adapter,raftengine: read the real status off a reset client stream

adapter/s3_blob_cluster.go

@@ -254,7 +254,13 @@ func (c *grpcS3BlobCluster) PushChunkBlob(ctx context.Context, replica S3BlobRep
 	if err != nil {
 		return errors.WithStack(err)
 	}
-	if err := sendS3ChunkBlobPushFrames(stream, digest, payload, commitTS); err != nil {
+	// gRPC reports a stream the server has already terminated to the sender as
+	// a bare io.EOF; the status it terminated with is only readable from the
+	// receive side. Falling through to CloseAndRecv is what turns an opaque EOF
+	// into the Unauthenticated or ResourceExhausted the replicator needs in
+	// order to decide whether this push is worth retrying.
+	if err := sendS3ChunkBlobPushFrames(stream, digest, payload, commitTS); err != nil &&
+		!errors.Is(err, io.EOF) {
 		return err
 	}
 	resp, err := stream.CloseAndRecv()

adapter/s3_blob_push_status_test.go

@@ -0,0 +1,57 @@
+package adapter
+
+import (
+	"context"
+	"crypto/sha256"
+	"net"
+	"testing"
+
+	pb "github.com/bootjp/elastickv/proto"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// rejectingS3BlobFetchServer refuses the push before reading a single frame,
+// which is what an auth interceptor or a fail-closed capability check does.
+type rejectingS3BlobFetchServer struct {
+	pb.UnimplementedS3BlobFetchServer
+}
+
+func (rejectingS3BlobFetchServer) PushChunkBlob(pb.S3BlobFetch_PushChunkBlobServer) error {
+	return status.Error(codes.Unauthenticated, "peer token required")
+}
+
+// TestPushChunkBlobSurfacesTheServersStatusWhenItRejectsEarly pins the gRPC
+// contract that a client-stream Send reports a server-terminated stream as a
+// bare io.EOF: the real status is only available from the receive side.
+//
+// A payload larger than the flow-control window guarantees the sender blocks
+// long enough to observe the reset, which is the case the caller must not see
+// as an opaque EOF -- the replicator decides whether to retry from this code.
+func TestPushChunkBlobSurfacesTheServersStatusWhenItRejectsEarly(t *testing.T) {
+	t.Parallel()
+
+	listener, err := (&net.ListenConfig{}).Listen(context.Background(), "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	server := grpc.NewServer()
+	pb.RegisterS3BlobFetchServer(server, rejectingS3BlobFetchServer{})
+	serveDone := make(chan error, 1)
+	go func() { serveDone <- server.Serve(listener) }()
+	t.Cleanup(func() {
+		server.Stop()
+		<-serveDone
+	})
+
+	cluster := NewGRPCS3BlobCluster("n1", staticS3BlobMembership{}, "read-only-admin", "peer-secret")
+	t.Cleanup(func() { require.NoError(t, cluster.Close()) })
+
+	payload := make([]byte, 4<<20)
+	digest := sha256.Sum256(payload)
+	replica := S3BlobReplica{NodeID: "n2", Address: listener.Addr().String(), Suffrage: "voter"}
+
+	err = cluster.PushChunkBlob(context.Background(), replica, digest, payload, 7)
+	require.Equal(t, codes.Unauthenticated, status.Code(err),
+		"a stream the server reset must report the server's status, not io.EOF")
+}

internal/raftengine/etcd/grpc_transport.go

@@ -518,7 +518,8 @@ func (t *GRPCTransport) streamFSMSnapshot(ctx context.Context, msg raftpb.Messag
 	// receiver-side total when a follower fails to restore. A mismatch points
 	// at transport truncation; a match points at a format/parsing issue.
 	counter := &countingReadCloser{inner: rc}
-	if err := sendSnapshotReaderChunks(stream, header, counter, t.chunkSize()); err != nil {
+	if err := sendSnapshotReaderChunks(stream, header, counter, t.chunkSize()); err != nil &&
+		!errors.Is(err, io.EOF) {
 		return err
 	}
 	if _, err := stream.CloseAndRecv(); err != nil {
@@ -1019,7 +1020,8 @@ func (t *GRPCTransport) sendSnapshot(ctx context.Context, msg raftpb.Message) er
 		return errors.WithStack(err)
 	}
 
-	if err := sendSnapshotChunks(stream, header, payload, t.chunkSize()); err != nil {
+	if err := sendSnapshotChunks(stream, header, payload, t.chunkSize()); err != nil &&
+		!errors.Is(err, io.EOF) {
 		return err
 	}
 	if _, err := stream.CloseAndRecv(); err != nil {
@@ -1047,7 +1049,8 @@ func (t *GRPCTransport) sendSnapshotSpool(ctx context.Context, msg raftpb.Messag
 	if err != nil {
 		return errors.WithStack(err)
 	}
-	if err := sendSnapshotReaderChunks(stream, header, reader, t.chunkSize()); err != nil {
+	if err := sendSnapshotReaderChunks(stream, header, reader, t.chunkSize()); err != nil &&
+		!errors.Is(err, io.EOF) {
 		return err
 	}
 	if _, err := stream.CloseAndRecv(); err != nil {
@@ -1170,6 +1173,11 @@ func sendSnapshotChunks(stream pb.EtcdRaft_SendSnapshotClient, header []byte, pa
 	return nil
 }
 
+// sendSnapshotChunk returns io.EOF unchanged when the receiver has already
+// terminated the stream. That is gRPC's signal that the real status is waiting
+// on the receive side, so callers must fall through to CloseAndRecv rather than
+// return it -- a follower that rejected the snapshot should appear in the log
+// as its actual reason, not as a bare EOF.
 func sendSnapshotChunk(stream pb.EtcdRaft_SendSnapshotClient, chunk *pb.EtcdRaftSnapshotChunk) error {
 	if err := stream.Send(chunk); err != nil {
 		return errors.WithStack(err)

internal/raftengine/etcd/grpc_transport_snapshot_status_test.go

@@ -0,0 +1,100 @@
+package etcd
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net"
+	"testing"
+
+	pb "github.com/bootjp/elastickv/proto"
+	"github.com/stretchr/testify/require"
+	raftpb "go.etcd.io/raft/v3/raftpb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// rejectingSnapshotServer refuses the snapshot before reading a single chunk,
+// the way a receiver out of spool space or holding a conflicting group ID does.
+type rejectingSnapshotServer struct {
+	pb.UnimplementedEtcdRaftServer
+}
+
+func (rejectingSnapshotServer) SendSnapshot(pb.EtcdRaft_SendSnapshotServer) error {
+	return status.Error(codes.FailedPrecondition, "receiver is not accepting snapshots")
+}
+
+// TestSnapshotSendersSurfaceTheReceiversStatus covers all three snapshot send
+// paths. gRPC reports a stream the receiver has already terminated to the
+// sender as a bare io.EOF, so a sender that returns the Send error loses the
+// receiver's actual reason -- and a snapshot that fails for a diagnosable
+// cause (no headroom, wrong group, unsupported format) shows up in the
+// operator's log as "EOF".
+//
+// The payload is deliberately larger than the flow-control window but smaller
+// than gRPC's 4 MiB send cap. Larger than the window is what guarantees the
+// sender is still writing when the reset arrives -- the case where the two
+// error paths differ. Under the cap matters because an oversized message makes
+// the client reject it locally with a status of its own, which would satisfy
+// the assertion without the stream ever being reset.
+func TestSnapshotSendersSurfaceTheReceiversStatus(t *testing.T) {
+	t.Parallel()
+
+	listener, err := (&net.ListenConfig{}).Listen(context.Background(), "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	server := grpc.NewServer()
+	pb.RegisterEtcdRaftServer(server, rejectingSnapshotServer{})
+	t.Cleanup(server.Stop)
+	go func() { _ = server.Serve(listener) }()
+
+	payload := make([]byte, 1<<20)
+	msg := raftpb.Message{
+		Type: messageTypePtr(raftpb.MsgSnap),
+		From: uint64Ptr(1),
+		To:   uint64Ptr(2),
+		Snapshot: &raftpb.Snapshot{
+			Data:     payload,
+			Metadata: testSnapshotMetadata(9, 3, nil),
+		},
+	}
+
+	newTransport := func(t *testing.T) *GRPCTransport {
+		t.Helper()
+		transport := NewGRPCTransport([]Peer{{NodeID: 2, Address: listener.Addr().String()}})
+		t.Cleanup(func() { require.NoError(t, transport.Close()) })
+		// The default 16 MiB chunk turns this payload into a single write, and
+		// a single write can win the race against the reset. Chunking small
+		// keeps the sender writing across the reset in every run, so the test
+		// measures the error path rather than the scheduler.
+		transport.snapshotChunkSize = 32 << 10
+		return transport
+	}
+
+	t.Run("in-memory payload", func(t *testing.T) {
+		t.Parallel()
+		err := newTransport(t).sendSnapshot(context.Background(), msg)
+		require.Equal(t, codes.FailedPrecondition, grpcStatusCode(err))
+	})
+
+	t.Run("spooled payload", func(t *testing.T) {
+		t.Parallel()
+		spool, err := newSnapshotSpool(t.TempDir())
+		require.NoError(t, err)
+		t.Cleanup(func() { require.NoError(t, spool.Close()) })
+		_, err = spool.Write(payload)
+		require.NoError(t, err)
+
+		err = newTransport(t).sendSnapshotSpool(context.Background(), msg, spool)
+		require.Equal(t, codes.FailedPrecondition, grpcStatusCode(err))
+	})
+
+	t.Run("streamed FSM snapshot", func(t *testing.T) {
+		t.Parallel()
+		open := func(uint64) (io.ReadCloser, error) {
+			return io.NopCloser(bytes.NewReader(payload)), nil
+		}
+		err := newTransport(t).streamFSMSnapshot(context.Background(), msg, 9, open)
+		require.Equal(t, codes.FailedPrecondition, grpcStatusCode(err))
+	})
+}