s3: add blob fetch rpc#1063
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Implemented the S3 blob fetch RPC slice on top of #1057. Changes:
Verification:
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the S3BlobFetch gRPC service to support peer-to-peer S3 chunk blob transfers. It implements the S3BlobFetchServer with FetchChunkBlob and PushChunkBlob streaming RPCs, registers the service in main.go, and adds comprehensive unit tests. The review feedback suggests a performance optimization in s3ChunkBlobReceiveState.finish() to return the payload bytes directly instead of cloning them, avoiding redundant memory allocation and copying.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if !s.seenEOF { | ||
| return s.digest, nil, s3BlobFetchStatus(codes.InvalidArgument, "missing s3 chunkblob eof") | ||
| } | ||
| return s.digest, bytes.Clone(s.payload.Bytes()), nil |
There was a problem hiding this comment.
Since s3ChunkBlobReceiveState is a local variable in recvChunkBlob and is not reused, its internal buffer will not be modified after finish() is called. Furthermore, the underlying store (mvccStore or pebbleStore) already clones or serializes the payload during PutAt. Therefore, calling bytes.Clone here is redundant and introduces an unnecessary allocation and memory copy of the entire chunk blob (which can be up to s3ChunkSize in size). We can safely return s.payload.Bytes() directly.
| return s.digest, bytes.Clone(s.payload.Bytes()), nil | |
| return s.digest, s.payload.Bytes(), nil |
|
Addressed the current-head S3BlobFetch review feedback. Changes:
Verification:
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 97c2749cec
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if !exists { | ||
| return nil, s3BlobFetchStatus(codes.NotFound, "s3 chunkblob not found") | ||
| } | ||
| payload, err := s.store.GetAt(ctx, key, readTS) |
There was a problem hiding this comment.
If the store compaction watermark has advanced past the chunkblob's write timestamp, this exact-timestamp GetAt returns ErrReadTSCompacted even though compaction keeps the newest version readable at current timestamps. In that normal post-compaction case FetchChunkBlob reports an Internal error for an existing blob; fetch should read at a current/max timestamp (or otherwise bypass the historical-read guard) after determining the key exists.
Useful? React with 👍 / 👎.
| if err := s.verifyChunkBlobDigest(digest, payload, codes.InvalidArgument); err != nil { | ||
| return err | ||
| } | ||
| if err := s.store.PutAt(stream.Context(), s3keys.ChunkBlobKey(digest), payload, s.clock.Next(), 0); err != nil { |
There was a problem hiding this comment.
This is a new MVCC write timestamp, but it is allocated with HLC.Next(), whose contract in kv/hlc.go says persistence-grade allocations that become MVCC write timestamps must use NextFenced(). When this node's HLC ceiling has expired (for example after lease renewal stalls during startup rotation or a leadership transition), PushChunkBlob will still write and acknowledge a local chunkblob instead of failing closed, letting direct chunkblob persistence bypass the same stale-leader timestamp fence as the rest of the store.
Useful? React with 👍 / 👎.
| grpcSvc := adapter.NewGRPCServer(shardStore, coordinate) | ||
| pb.RegisterRawKVServer(gs, grpcSvc) | ||
| pb.RegisterTransactionalKVServer(gs, grpcSvc) | ||
| pb.RegisterS3BlobFetchServer(gs, adapter.NewS3BlobFetchServer(rt.store, coordinate.Clock(), nil)) |
There was a problem hiding this comment.
The production registration constructs the only S3BlobFetchServer with a nil observer, so observeSHAMismatch() never reaches elastickv_s3_chunkblob_sha_mismatch_total for this RPC path. In the corrupt-local-blob or bad-push scenarios that this service explicitly detects, the request is rejected but operators get no S3 chunkblob mismatch signal even though the public S3 server is already wired with r.metricsRegistry.S3BlobOffloadObserver().
Useful? React with 👍 / 👎.
| // (nil when encryption is not wired); raftadmin.Server invokes | ||
| // it before AddVoter/AddLearner propose the conf-change. | ||
| internalraftadmin.RegisterOperationalServicesWithInterceptor(ctx, gs, rt.engine, []string{"RawKV"}, confChangeInterceptor) | ||
| internalraftadmin.RegisterOperationalServicesWithInterceptor(ctx, gs, rt.engine, []string{"RawKV", "S3BlobFetch"}, confChangeInterceptor) |
There was a problem hiding this comment.
Including S3BlobFetch in this leader-only health loop makes followers report NOT_SERVING for a service that is intentionally peer-local; currentHealthStatus in internal/raftadmin/health.go maps only the Raft leader to SERVING. Any gRPC client using health checking for service S3BlobFetch would therefore avoid the follower endpoints that the leader must push chunkblobs to and that readers may fan out to on a miss.
Useful? React with 👍 / 👎.
| if err := s.verifyChunkBlobDigest(digest, payload, codes.InvalidArgument); err != nil { | ||
| return err | ||
| } | ||
| if err := s.store.PutAt(stream.Context(), s3keys.ChunkBlobKey(digest), payload, s.clock.Next(), 0); err != nil { |
There was a problem hiding this comment.
If the leader retries after losing the response, or pushes the same content-addressed chunk for another reference, this unconditionally writes a fresh MVCC version because every call uses a new timestamp. That turns one !s3|chunkblob|<sha256> row into repeated full-value versions until compaction; after verifying the digest, duplicate pushes should detect an existing matching blob and acknowledge without appending another copy.
Useful? React with 👍 / 👎.
| if err != nil { | ||
| return err | ||
| } | ||
| if err := s.verifyChunkBlobDigest(digest, payload, codes.DataLoss); err != nil { |
There was a problem hiding this comment.
For a peer with a corrupted local copy, this returns DataLoss, but the S3 blob-offload design documents SHA mismatch during FetchChunkBlob as an INVALID_ARGUMENT fallback trigger alongside NOT_FOUND. A client implemented against that contract would treat this peer as a hard failure instead of trying the remaining replicas, so a single bad copy can break a read even when other peers still have the chunk.
Useful? React with 👍 / 👎.
|
Addressed the current-head S3BlobFetch review feedback. Changes:
Caller audit:
Verification:
|
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d207d49843
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| } | ||
|
|
||
| func (s *S3BlobFetchServer) nextChunkBlobCommitTS() (uint64, error) { | ||
| commitTS, err := s.clock.NextFenced() |
There was a problem hiding this comment.
Because this RPC is registered on every raft listener with that node's local clock, a leader→follower PushChunkBlob allocates from the follower's HLC here. In clusters where a follower has not yet applied the leader's renewed HLC ceiling, NextFenced can return ErrCeilingExpired even though the current leader can safely issue persistence timestamps, so chunkblob replication can fail/degrade and block offloaded PUTs when enough followers are lagging. Have the leader allocate and carry the commit timestamp for the receiver to use/observe instead of issuing it locally.
Useful? React with 👍 / 👎.
| if err := s.store.ApplyMutations(ctx, []*store.KVPairMutation{{ | ||
| Op: store.OpTypePut, | ||
| Key: key, | ||
| Value: payload, | ||
| }}, nil, 0, commitTS); err != nil { |
There was a problem hiding this comment.
When storage envelopes are active but this process load has not yet marked its writer registration (for example just after runtime EnableStorageEnvelope/RotateDEK, or while startup registration is still catching up), the direct ApplyMutations path consults the storage registration gate and returns store.ErrWriterNotRegistered. This new runtime direct-write caller is not wrapped in retryUntilRegistered, so PushChunkBlob reports an internal write failure instead of waiting, making synchronous chunkblob replication fail transiently on nodes that are otherwise advertised as S3BlobFetch SERVING.
Useful? React with 👍 / 👎.
| Op: store.OpTypePut, | ||
| Key: key, | ||
| Value: payload, | ||
| }}, nil, 0, commitTS); err != nil { |
There was a problem hiding this comment.
For a digest that was previously garbage-collected with a tombstone, chunkBlobAlreadyStored sees GetAt(..., MaxUint64) as not found, but this ApplyMutations call still uses startTS=0, so MVCC conflict checking rejects the newer tombstone as a write conflict and the retry path re-reads the tombstone as missing again. A later upload of the same content digest can therefore fail with an internal write error instead of recreating the local chunkblob; use a start timestamp at/after the latest version (or a direct sync put path with equivalent durability) for this content-addressed upsert.
Useful? React with 👍 / 👎.
| grpcSvc := adapter.NewGRPCServer(shardStore, coordinate) | ||
| pb.RegisterRawKVServer(gs, grpcSvc) | ||
| pb.RegisterTransactionalKVServer(gs, grpcSvc) | ||
| pb.RegisterS3BlobFetchServer(gs, adapter.NewS3BlobFetchServer(rt.store, coordinate.Clock(), s3BlobObserver)) |
There was a problem hiding this comment.
startRaftTransport opens these gRPC listeners before waitRotateOnStartup.Wait completes, and the existing startup gate is only a unary interceptor over RawKV/TransactionalKV/RaftAdmin/EncryptionAdmin methods. Since PushChunkBlob is a new client-streaming mutator and is registered here without any stream gate, a peer can write chunkblobs while --encryption-rotate-on-startup is still blocking other mutators, bypassing the rotation safety window that keeps writes from landing under the pre-rotation state.
Useful? React with 👍 / 👎.
|
Addressed the current-head S3BlobFetch review feedback. Changes:
Durability note:
Caller audit:
Verification:
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 0de5f9f36b
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if stored { | ||
| return nil |
There was a problem hiding this comment.
When a digest already has a visible payload, this fast path returns Durable=true without validating or appending the supplied commit_ts. If an object reuses an old chunk at timestamp 30 while the follower only has that chunk at timestamp 10, then a GC tombstone at timestamp 20 applied after this ack becomes the latest version and FetchChunkBlob at MaxUint64 returns NotFound, even though the push for timestamp 30 was acknowledged. Please only skip the write when the existing latest version is already at/after the pushed timestamp, or append the duplicate payload at the leader-supplied timestamp.
Useful? React with 👍 / 👎.
| return err | ||
| } | ||
| if err := s.applyChunkBlobUntilRegistered(stream.Context(), key, payload, startTS, commitTS); err != nil { | ||
| if stored, retryErr := s.storedAfterWriteConflict(stream.Context(), err, key, digest, payload); stored || retryErr != nil { |
There was a problem hiding this comment.
This conflict path only treats ErrWriteConflict as success when the payload is currently visible. If chunkBlobWriteStartTS reads version 10, a GC tombstone at 20 lands before ApplyMutations, and the pushed commit_ts is 30, the write is still valid but this branch falls through as an internal error instead of recomputing the latest timestamp and recreating the chunk. That makes recreation after GC fail under normal races unless the caller happens to retry the whole RPC.
Useful? React with 👍 / 👎.
|
Addressed the current-head S3BlobFetch review feedback. Changes:
Caller audit:
Verification:
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5288c41ec0
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
@codex review Current HEAD
Validation run on this head:
|
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5288c41ec0
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if err := s.store.ApplyMutations(ctx, []*store.KVPairMutation{{ | ||
| Op: store.OpTypePut, | ||
| Key: key, | ||
| Value: payload, | ||
| }}, nil, startTS, commitTS); err != nil { |
There was a problem hiding this comment.
When PushChunkBlob is exercised, this direct write goes into the same MVCCStore that backs the Raft FSM; the Pebble snapshot path streams every key from that store, so local-only !s3|chunkblob|... payloads will be captured in Raft snapshots and transferred/restored as FSM state. That defeats the offload path's snapshot-size goal and makes snapshots depend on whichever peer-local blobs happened to exist on the snapshotting node; store these blobs in snapshot-excluded local storage or filter this prefix from FSM snapshots.
Useful? React with 👍 / 👎.
TLA+ spec divergence review (auto-triggered)This PR touches files that the TLA+ safety spec has an anchor on (per Anchored files changed in this PR head (b06af8a):
What to check, by subsystem:
If the change is correct but requires a spec update, edit @claude review please verify TLA+ spec divergence per the checklist above. @codex review please verify TLA+ spec divergence per the checklist above. |
|
To use Codex here, create a Codex account and connect to github. |
|
@codex review Addressed current S3BlobFetch findings:
Caller audit:
Verification:
Note: broader |
|
Codex Review: Didn't find any major issues. Chef's kiss. Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
1 participant
Summary
Verification