diff --git a/adapter/s3_blob_cluster.go b/adapter/s3_blob_cluster.go index c7f80233e..6965108b9 100644 --- a/adapter/s3_blob_cluster.go +++ b/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() diff --git a/adapter/s3_blob_push_status_test.go b/adapter/s3_blob_push_status_test.go new file mode 100644 index 000000000..c8f403275 --- /dev/null +++ b/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") +} diff --git a/internal/raftengine/etcd/grpc_transport.go b/internal/raftengine/etcd/grpc_transport.go index fdf5cc0b3..11322d0e5 100644 --- a/internal/raftengine/etcd/grpc_transport.go +++ b/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) diff --git a/internal/raftengine/etcd/grpc_transport_snapshot_status_test.go b/internal/raftengine/etcd/grpc_transport_snapshot_status_test.go new file mode 100644 index 000000000..2eca69022 --- /dev/null +++ b/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)) + }) +}