From b1216333f3012d9e986342114ec1b9a6c22edf3c Mon Sep 17 00:00:00 2001 From: bootjp Date: Tue, 15 Sep 2026 23:28:07 +0900 Subject: [PATCH] adapter,raftengine: read the real status off a reset client stream 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. Four client-stream senders returned that EOF instead of falling through to CloseAndRecv, so the peer's actual reason never reached the caller. PushChunkBlob is the one with a visible consequence: the replicator decides whether a push is worth retrying from this error, and an Unauthenticated from a peer missing its token looked identical to a transport hiccup. It is also a real CI failure -- TestGRPCS3BlobCluster- AuthenticatedCapabilityPushAndFetch asserted Unauthenticated and got Unknown whenever the reset won the race, which is exactly this. The three snapshot senders lose diagnosability rather than behaviour: a follower that rejected a snapshot for a nameable reason appeared in the sender's log as "EOF". The tests push more than the flow-control window so the sender is still writing when the reset lands, and reverting each of the four guards fails exactly the one test that covers it. They stay under gRPC's 4 MiB send cap: an oversized message makes the client reject it locally with a status of its own, which satisfied the assertion without the stream ever being reset -- the first version of this test passed against the bug for that reason. The snapshot test also shrinks the chunk size, because the 16 MiB default turns the payload into a single write that can win the race, which left the in-memory path uncovered. Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE --- adapter/s3_blob_cluster.go | 8 +- adapter/s3_blob_push_status_test.go | 57 ++++++++++ internal/raftengine/etcd/grpc_transport.go | 14 ++- .../grpc_transport_snapshot_status_test.go | 100 ++++++++++++++++++ 4 files changed, 175 insertions(+), 4 deletions(-) create mode 100644 adapter/s3_blob_push_status_test.go create mode 100644 internal/raftengine/etcd/grpc_transport_snapshot_status_test.go 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)) + }) +}