tso: add dedicated ceiling fsm - #1095
Conversation
|
@codex review |
|
Caution Review failedAn error occurred during the review process. Please try again later. 📝 WalkthroughWalkthrough専用TSOのRaft予約、TSO FSMの永続状態、Phase Dのtimestamp検証、ReadTimestamp voucherを追加した。Redis、DynamoDB、S3、SQS、Distributionの書き込み経路を同じ読み取り時点で実行するよう更新した。専用TSO groupの起動配線とEncryptionAdmin制御も更新した。 ChangesTSO状態、予約、Phase D検証
RuntimeとRPC配線
アダプターの読み取り時点統一
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Adapter
participant Coordinator
participant TimestampAllocator
participant ShardedStore
Adapter->>Coordinator: BeginReadTimestampThrough
Coordinator->>TimestampAllocator: allocate or validate timestamp
TimestampAllocator->>ShardedStore: verify durable floor
ShardedStore-->>TimestampAllocator: validated watermark
TimestampAllocator-->>Coordinator: ReadTimestamp and voucher
Coordinator-->>Adapter: read timestamp
Adapter->>Coordinator: DispatchWithReadTimestamp
Coordinator->>ShardedStore: dispatch transaction with StartTS
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
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 |
![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 implements the TSOStateMachine, a minimal state machine for the dedicated timestamp group that handles HLC lease-renewal entries, along with its snapshotting, restoration, and volatile entry classification logic. It also updates the design documentation to reflect this progress and adds comprehensive unit tests for the new state machine. There are no review comments to evaluate, and I have no additional feedback to provide.
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.
![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
Update stale M6 status row
This milestone row still says the minimal TSOStateMachine remains open, but this same change marks it implemented in §1.0 and adds kv/tso_fsm.go. Readers using this design doc to plan M6 will incorrectly think the FSM work is still pending, so please update the row to match the new runtime-wiring-only remaining status.
ℹ️ 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".
| } | ||
| ceilingMs := int64(binary.BigEndian.Uint64(data[1:])) //nolint:gosec // value is a Unix ms timestamp encoded as uint64. | ||
| if f != nil && f.hlc != nil && ceilingMs > 0 { | ||
| f.hlc.SetPhysicalCeiling(ceilingMs) |
There was a problem hiding this comment.
When this FSM is used for the dedicated TSO group, applying a lease only updates physicalCeiling while leaving HLC.last at zero on followers/restarted nodes. If the current TSO leader has handed out timestamps under a future ceilingMs and then fails before the wall clock reaches it, the new leader has applied the same lease but has not observed any of those allocations, so its first NextBatchFenced can return (ceilingMs<<16)|0 again and duplicate timestamps. The TSO path needs to persist/observe an allocation floor or force a fresh higher lease before serving after leadership changes.
Useful? React with 👍 / 👎.
|
|
||
| func (f *TSOStateMachine) Apply(data []byte) any { | ||
| if len(data) != hlcLeaseEntryLen { | ||
| return errors.Wrapf(ErrTSOStateMachineInvalidEntry, "expected %d bytes, got %d", hlcLeaseEntryLen, len(data)) |
There was a problem hiding this comment.
If a malformed payload is ever committed to the dedicated TSO group, returning a bare error here does not actually reject it at the raft apply layer: applyNormalCommitted only stops before setApplied for responses implementing HaltApply() error, while plain error responses are still marked applied and only delivered back to the proposer. In a misrouted group-0 proposal or mixed-version entry scenario, the invalid entry would be permanently advanced past instead of failing closed, violating the FSM's “accepts only HLC lease entries” contract.
Useful? React with 👍 / 👎.
|
@codex review Updated commit: 4e49dfbe7f31f084fc98bb5bec47a0e1458c4c24 Addressed current-head review findings:
Caller/semantic audit:
Validation:
|
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 4e49dfbea8
ℹ️ 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".
| return | ||
| } | ||
| hlc.SetPhysicalCeiling(ceilingMs) | ||
| hlc.Observe(tsoLeaseAllocationFloor(ceilingMs)) |
There was a problem hiding this comment.
When the dedicated TSO FSM is wired, applying/restoring a lease with this Observe only records ceilingMs|maxLogical in memory while snapshots/WAL entries still persist just ceilingMs. If a leader applies ceiling C, serves one timestamp, NextBatchFenced returns (C+1,0) because the floor is already C|maxLogical; after a restart or re-election with only the same lease entry/snapshot, applyTSOLeaseToHLC recreates the same floor and can serve (C+1,0) again. This means the new floor does not actually prevent reissuing timestamps unless the allocation floor/current HLC is made durable or allocation is blocked until a higher ceiling is committed.
Useful? React with 👍 / 👎.
| if hlc == nil || ceilingMs <= 0 { | ||
| return |
There was a problem hiding this comment.
When a full HLC-lease entry encodes 0 or a high-bit value that decodes negative, this branch returns without error, so Apply returns nil and the engine advances the applied index. Renewals only generate positive Unix-ms ceilings, so these are malformed group-0 lease entries; accepting them masks the bad proposal instead of using the new fail-closed HaltApply path. Please reject non-positive ceilings in Apply while keeping snapshot restore's empty-state handling separate if needed.
Useful? React with 👍 / 👎.
| if f != nil { | ||
| hlc = f.hlc | ||
| } | ||
| return &tsoFSMSnapshot{ceilingMs: hlcCeilingFromHLC(hlc)}, nil |
There was a problem hiding this comment.
When group 0 is wired during the documented bridge, this FSM is likely to share the process HLC that shard FSMs also advance; then Snapshot can serialize a ceiling raised by some data-group lease even if the TSO group never committed that lease. If a shard renewal succeeds while the group-0 proposal fails or lags, installing/restoring the group-0 snapshot advances dedicated TSO state beyond its own consensus log. Keep a TSO-FSM-owned ceiling field updated only from this FSM's Apply/Restore path and snapshot that value instead.
Useful? React with 👍 / 👎.
|
@codex review Updated commit: 44e6c18 Addressed current-head review findings:
Caller/semantic audit:
Validation:
Additional check:
|
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 44e6c1852b
ℹ️ 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".
| case tsoSnapshotV1Len: | ||
| ceilingMs = int64(binary.BigEndian.Uint64(payload[:hlcLeasePayloadLen])) //nolint:gosec // legacy snapshot value. |
There was a problem hiding this comment.
When this restores an 8-byte snapshot emitted by the previous TSO FSM format, allocationFloor remains 0. Those legacy snapshots were produced by code that also observed tsoLeaseAllocationFloor(ceilingMs) on apply/restore to prevent a restarted or newly elected TSO leader from handing out logical slots from the old ceiling millisecond. After upgrading from such a snapshot, the HLC mirror has only the ceiling, so the next NextBatchFenced can start at ceiling<<16|0 and overlap timestamps already covered by the prior floor; derive the legacy floor from the decoded ceiling before calling restoreSnapshotState.
Useful? React with 👍 / 👎.
| f.ceilingMs.Store(ceilingMs) | ||
| f.allocationFloor.Store(allocationFloor) |
There was a problem hiding this comment.
When Restore is called on an FSM that has already applied a higher ceiling/floor (the monotonic-restore test exercises this shape), these unconditional stores regress the FSM-owned state even though the HLC mirror remains high via its monotonic setters. Because Snapshot() now serializes f.ceilingMs and f.allocationFloor rather than the HLC, the next snapshot from that FSM can persist the lower values and lose the higher applied TSO floor/ceiling; restore should max these fields the same way apply does.
Useful? React with 👍 / 👎.
|
@codex review Updated commit: 5b6fd30 Addressed current-head P2 findings:
Caller/semantic audit:
Validation:
|
|
Codex Review: Didn't find any major issues. Nice work! 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". |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 46be7a832f
ℹ️ 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".
| sg := &kv.ShardGroup{Engine: runtime.engine} | ||
| sg.Txn = kv.NewLeaderProxyForShardGroup(sg, kv.WithProposalObserver(proposalObserver)) |
There was a problem hiding this comment.
When group 0 is configured alongside a single data group, this returned ShardGroup is added to encWiring.attachRaftEnvelopeGroup in buildShardGroups. I checked adapter/encryption_admin.go's EnableRaftEnvelope path and main_raft_envelope_wiring.go's ValidateCutoverScope: the cutover precheck rejects unless exactly one group is attached, so enabling the dedicated TSO group turns a previously valid single-data-group raft-envelope rollout into FailedPrecondition (got 2) even though group 0 holds no user data. Exclude group 0 from that scope check or handle its cutover separately.
Useful? React with 👍 / 👎.
| legacy, err := hasLegacyKVFSMSnapshotHeader(br) | ||
| if err != nil || !legacy { | ||
| return legacy, err |
There was a problem hiding this comment.
When upgrading a reserved group-0 log that previously ran kvFSM from a version that emitted the supported headerless snapshot format, this probe returns legacy=false, so Restore falls through to readTSOSnapshotState and accepts only exactly 8 or 16 bytes. ReadSnapshotHeader deliberately leaves non-magic/headerless snapshots unconsumed for the inner store restore path, and those raw store payloads are not constrained to the TSO lengths, so affected nodes can fail to restore or misparse a 16-byte store payload instead of draining the legacy MVCC snapshot. Detect and drain headerless kvFSM snapshots before trying the raw TSO format.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e701cb221b
ℹ️ 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".
e701cb2 to
97a42ba
Compare
|
Evidence for current HEAD
History sanitation:
Validation:
@codex review |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
kv/tso_fsm.go (1)
94-94: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoff
//nolint:gosecの抑制は境界チェックへの置き換えを検討してください。コーディングガイドラインでは
//nolintの追加を避け、リファクタリングを優先することが求められています。タイムスタンプのuint64⇄int64変換については、変換前にmath.MaxInt64との境界チェックを行う小さなヘルパー（例:func unixMillisToInt64(v uint64) (int64, error)）を導入することで、複数箇所の//nolint:gosecを排除できます。既存コードで許容される慣例であれば据え置きで構いませんが、新規追加分については抑制の集約を推奨します。As per coding guidelines: "Avoid adding
//nolintunless absolutely required; prefer refactoring."Also applies to: 165-167, 197-197, 325-325
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/tso_fsm.go` at line 94, Replace the new `//nolint:gosec` suppressions around the `ceilingMs` conversion and the corresponding conversions at the other referenced sites with a shared checked conversion helper, such as `unixMillisToInt64`. Have the helper validate against `math.MaxInt64` before converting and return an error for overflow, then propagate or handle that error at each caller while preserving existing timestamp behavior.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@kv/tso_fsm.go`:
- Line 94: Replace the new `//nolint:gosec` suppressions around the `ceilingMs`
conversion and the corresponding conversions at the other referenced sites with
a shared checked conversion helper, such as `unixMillisToInt64`. Have the helper
validate against `math.MaxInt64` before converting and return an error for
overflow, then propagate or handle that error at each caller while preserving
existing timestamp behavior.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 378b480f-a7cd-42e8-a4ec-e03b8e9fa899
📒 Files selected for processing (7)
docs/design/2026_04_16_partial_centralized_tso.mdkv/tso_fsm.gokv/tso_fsm_test.gomain.gomain_encryption_admin.gomain_encryption_admin_test.gomultiraft_runtime_test.go
|
Codex Review: Didn't find any major issues. Delightful! 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". |
## Summary - route dedicated group-0 timestamp requests to the current TSO leader - commit every returned window end and fence each leader term above authoritative data-group commit floors - add synchronous fail-closed shadow migration and a durable one-way cutover marker - preserve rolling compatibility by rejecting legacy timestamp responses without durable reservation metadata - update the centralized TSO design status through M6 ## Migration safety - shadow candidates are serialized through group 0 before the legacy value is returned - overlapping legacy candidates are discarded and retried - cutover commits the marker before the first production window - group-0, shadow, and cutover failures stop timestamp issuance instead of falling back ## Validation - `go test ./kv -count=1 -timeout=10m` - `go test . -count=1 -timeout=10m` - `go test ./adapter -run 'Test(DistributionServerGetTimestamp|GRPCServer_RawLatestCommitTS_EmptyKeyReturnsGlobalWatermark)' -count=1 -timeout=5m` - `go test ./... -run '^$' -count=1 -timeout=10m` - `go test -race ./kv -run 'Test(RaftTSOAllocator|LeaderRoutedTSOAllocator|ShadowTimestampAllocator|ShardStoreGlobalCommittedTimestampFloor)' -count=1 -timeout=10m` - `golangci-lint --config=.golangci.yaml run ./kv ./adapter . --timeout=5m` - `make gen BREAKING_AGAINST='../.git#subdir=proto,branch=design/dedicated-tso-runtime'` ## Stack Base: #1103 Author: bootjp
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: aa94f4eedb
ℹ️ 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 min == ^uint64(0) { | ||
| return empty, errors.WithStack(ErrTxnCommitTSRequired) |
There was a problem hiding this comment.
When ReserveBatchAfter is called with n > 1 and a minimum close to MaxUint64 (for example Distribution.GetTimestamp{count:2, min_timestamp:MaxUint64-1}), this guard lets the request through; a.clock.Observe(min) then causes NextBatchFenced to wrap the reserved base to 0 and the HLC current value to 1 before the later response validation can fail, and commitAllocationFloor can persist that regressed floor. Reject any min that cannot fit the requested window before observing it.
Useful? React with 👍 / 👎.
| cfg.engine, | ||
| distCatalog, | ||
| adapter.WithDistributionCoordinator(coordinate), | ||
| adapter.WithDistributionTimestampAllocator(tsoWiring.serverAllocator), |
There was a problem hiding this comment.
When a client can reach the raft gRPC listener during startup rotation, this wiring makes Distribution.GetTimestamp call the dedicated allocator and commit group-0 allocation/cutover entries, but startupRotationGatedMethod only gates Distribution.SplitRange plus the RawKV/Transactional/Internal/Admin mutators. That lets timestamp proposals bypass the same startup mutator fence before waitRotateOnStartup completes and before public traffic is marked ready; include GetTimestamp in the gate or delay exposing the allocator until the gate opens.
Useful? React with 👍 / 👎.
| return &pb.GetTimestampResponse{Timestamp: s.engine.NextTimestamp()}, nil | ||
| } | ||
|
|
||
| reservation, err := s.allocateTimestampReservation(ctx, count, minTimestamp, activateCutover) |
There was a problem hiding this comment.
When group 0 is configured for Phase A/B but this node was not started with --tsoEnabled, any caller that can reach Distribution.GetTimestamp can set activate_cutover=true; this line forwards that untrusted request bit into the allocator, which commits the one-way cutover marker. Since main wires the server allocator even before production cutover, this bypasses the operator flag and the all-nodes-shadow rollout precondition, so keep activation tied to local/internal configuration rather than the public RPC field.
Useful? React with 👍 / 👎.
| return 0, errors.Wrapf(ErrTSOCommitFloorUnavailable, | ||
| "data group %d is not led by this node", groupID) | ||
| } | ||
| if _, err := linearizableReadEngineCtx(nonNilTSOContext(ctx), engine); err != nil { |
There was a problem hiding this comment.
When the TSO leader also leads a data group and the incoming timestamp request has no deadline, this local LinearizableRead inherits an unbounded context, unlike verifyLeaderEngineCtx and the remote floor RPC timeout. If that ReadIndex stalls, ReserveBatchAfter is still holding the allocator mutex while initializing the term floor, so one stuck local floor probe can wedge all subsequent TSO allocations; wrap this path in a bounded context.
Useful? React with 👍 / 👎.
Author: bootjp ## Summary - persist a one-way Phase D marker and floor after the production cutover marker - stop data-group HLC renewal and legacy/shadow timestamp issuance once Phase D is durable - validate caller-supplied cross-shard SSI timestamps at the current group-0 leader before allocating a commit timestamp or proposing writes - migrate adapter read-modify-write snapshots to dedicated TSO allocations while preserving single-shard and pre-Phase-D compatibility - retain the parent branch's same-term allocation fences and RawLatestCommitTS group_id/leader_fenced schema ## Safety - activation commits cutover, Phase D, and the first post-Phase-D allocation in order - validation accepts only timestamps in the durable post-marker allocation interval and fails closed on inactive state, stale leadership, unsupported routing, or out-of-range values - leader term is revalidated after floor/marker work and after the allocation-floor commit - proto output was regenerated from the combined schema ## Tests - make gen BREAKING_AGAINST='../.git#subdir=proto,branch=design/dedicated-tso-leader-routing' - go test ./kv ./adapter . -run 'TSO|PhaseD|RenewHLC|CrossShard|ValidateTimestamp|RawLatestCommitTS' -count=1 - go test ./kv -run 'TestRaftTSOAllocator(RejectsTermChange|DropsCommittedWindow|CommitsPhaseD)' -count=1 - go test ./adapter -run '^Test_consistency_satisfy_write_after_read_sequence$' -count=1 -timeout=10m - go test ./adapter -run '^TestRedis_StreamXReadLatencyIsConstant$' -count=1 -timeout=10m - golangci-lint run ./... --timeout=5m --allow-parallel-runners The full parallel go test ./... run completed all packages except two long adapter stress tests after their test HLC ceilings expired under concurrent local load; both tests passed when rerun independently.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ca1a812729
ℹ️ 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 timestamp == 0 || timestamp > end { | ||
| return errors.Wrapf(ErrTSOTimestampInvalid, | ||
| "timestamp=%d phase_d_floor=%d allocation_floor=%d", timestamp, floor, end) | ||
| } | ||
| if timestamp <= floor { |
There was a problem hiding this comment.
When Phase D is activated with no previous floor, or after a wall-clock/lease jump, the first reserved TSO Base can be far above PhaseDFloor (for example PhaseDFloor == 0 and Base == current_ms<<16). This validation accepts any timestamp in (floor, allocationFloor], so ValidateDurableTimestamp(1) can succeed even though group 0 never issued that timestamp, letting caller-supplied cross-shard StartTS values in the gap bypass the durable-TSO allocation gate. Track the first post-D base/ranges or set the marker floor to Base-1 before treating the interval as valid.
AGENTS.md reference: AGENTS.md:L35-L35
Useful? React with 👍 / 👎.
| func (r *RedisServer) beginTxnStartTS(ctx context.Context, label string) (uint64, error) { | ||
| readTimestamp, err := r.beginTxnReadTimestamp(ctx, label) | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
| return readTimestamp.Timestamp(), nil |
There was a problem hiding this comment.
When Phase D is active and the adapter read watermark predates the Phase-D floor, beginTxnReadTimestamp returns a voucher that authorizes that applied snapshot, but this helper discards it and returns only the numeric timestamp. Callers such as wide set/zset/list mutations then dispatch via dispatchElems/plain Dispatch; in a sharded catalog where the wide collection storage prefixes resolve to more than one group, the coordinator treats the StartTS as caller-supplied without the bound voucher and rejects the command as a pre-Phase-D timestamp. Return the ReadTimestamp through these paths or bind it with DispatchWithReadTimestamp at dispatch time.
Useful? React with 👍 / 👎.
| if err != nil { | ||
| return nil, err | ||
| } | ||
| defer revoke() |
There was a problem hiding this comment.
When the same ReadTimestamp is used by overlapping DispatchWithReadTimestamp calls during Phase D, each call increments the same (timestamp, ref) use count, but a call that successfully reaches consumeAppliedReadTimestampVoucher still runs this deferred revoke afterward. If one dispatch consumes its token while another prepared token is still in the map, the defer can delete/decrement the other dispatch's authorization before it validates, causing valid applied-watermark operations to fail as pre-Phase-D timestamps. Make the revoke conditional on the prepared token remaining unused, or mint a distinct ref per prepared dispatch.
Useful? React with 👍 / 👎.
| err := s.retryS3Mutation(r.Context(), func() error { | ||
| readTS := s.readTS() | ||
| startTS, err := s.txnStartTS(r.Context(), readTS) | ||
| readTimestamp, err := s.beginTxnReadTimestamp(r.Context(), readTS, "s3 delete object: begin read timestamp") |
There was a problem hiding this comment.
When Phase D is active and the applied S3 read watermark predates the Phase-D floor, this call can return a ReadTimestamp carrying the voucher for the bucket-metadata read at readTS, but the later delete dispatch uses bare s.coordinator.Dispatch with only the object-manifest mutation. If the bucket metadata route and object route are split, the coordinator sees a single mutation shard and no read keys, so it accepts the legacy StartTS without consuming the voucher even though the operation semantically read another shard. Dispatch this transaction with readTimestamp.WithDispatchVoucher/DispatchWithReadTimestamp.
Useful? React with 👍 / 👎.
|
Addressed the latest review and merge-resolution findings:
Tests:
|
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 (fea84f9):
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: Didn't find any major issues. Swish! 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". |
fea84f9 to
227aff5
Compare
|
Review feedback addressed on the latest head. Changes:
Validation:
|
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 (227aff5):
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. |
There was a problem hiding this comment.
💡 Codex Review
elastickv/adapter/dynamodb_migration.go
Line 99 in 227aff5
Bind migration dispatches to the read voucher
This migration path receives a ReadTimestamp but dispatches the cross-key table-generation update without binding/registering its voucher. Under Phase D, tables created before the cutover commonly read at a pre-D applied watermark; when the metadata/generation keys are on different shards, the coordinator rejects this caller-supplied StartTS as pre-D instead of consuming the intended applied-read voucher, blocking the legacy key migration. Dispatch through kv.DispatchWithReadTimestamp(readTimestamp.WithDispatchVoucher(ctx), ...) for these migration writes.
ℹ️ 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".
| return err | ||
| } | ||
| req.StartTS = readTS | ||
| if _, err := d.coordinator.Dispatch(ctx, req); err == nil { |
There was a problem hiding this comment.
When Phase D is required and readTS is an applied pre-D watermark, beginTxnReadTimestamp returns a voucher that is only registered by kv.DispatchWithReadTimestamp; this direct Dispatch drops that capability. If the table metadata and generation keys route to different shards, ShardedCoordinator treats StartTS=readTS as an arbitrary caller timestamp and rejects it with ErrTSOTimestampPrePhaseD, so CreateTable can fail after Phase-D activation on upgraded clusters with pre-D data. Use the read timestamp's dispatch helper here, as the item-write path does.
Useful? React with 👍 / 👎.
| readTimestamp, err := s.beginTxnReadTimestamp(ctx, "sqs send message: begin read timestamp") | ||
| if err != nil { | ||
| return nil, 0, errors.WithStack(err) | ||
| } | ||
| readTS := readTimestamp.Timestamp() |
There was a problem hiding this comment.
Here beginTxnReadTimestamp may mint a Phase-D applied-read voucher, but the helper immediately collapses it to a bare uint64; the standard SendMessage paths then dispatch StartTS=readTS via plain coordinator.Dispatch. On Phase-D clusters with pre-D queue metadata, any standard send whose queue metadata/read keys and message index writes route across shards is rejected as an arbitrary pre-D timestamp instead of consuming the voucher, so standard SQS sends can fail after cutover. Return and bind the ReadTimestamp through the send dispatch path.
Useful? React with 👍 / 👎.
| if err != nil { | ||
| return false, errors.WithStack(err) | ||
| } | ||
| readTS := readTimestamp.Timestamp() |
There was a problem hiding this comment.
This path keeps the ReadTimestamp only long enough to extract readTS; the later CreateQueue transaction, and the analogous DeleteQueue/PurgeQueue generation bumps, still dispatch with plain s.coordinator.Dispatch. Under Phase D, when the queue meta/gen/tombstone prefixes route to different shards and the applied watermark is pre-D, cross-shard validation rejects that StartTS as an arbitrary caller timestamp, so queue create/delete/purge can fail after cutover. Carry the ReadTimestamp through those dispatches and bind it with DispatchWithReadTimestamp.
Useful? React with 👍 / 👎.
|
Updated this PR against current main and addressed the current-head Phase-D voucher findings. Changes:
Validation:
Note: full go test ./adapter -count=1 -timeout=600s timed out while focused adapter coverage passed. @codex review |
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 (c231fdb):
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. |
There was a problem hiding this comment.
Actionable comments posted: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
adapter/s3_admin.go (1)
333-361: 🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy lift予約した applied-read voucher が dispatch へ届いていない S3 経路が 3 箇所あります。 共通の根本原因は、
beginTxnReadTimestampが返すkv.ReadTimestampを破棄し、WithDispatchVoucher+kv.DispatchWithReadTimestampへ渡していないことです。Phase-D ではBeginReadTimestampThroughが voucher を予約するため、未消費の予約が残ります。kv/sharded_coordinator_txn_test.goのTestDispatchWithReadTimestampVouchesEveryBoundDispatchは、未消費の voucher が残った状態で素のDispatchを呼ぶとErrTSOTimestampPrePhaseDになることを示しています。
adapter/s3_admin.go#L333-L361: Line 354 のs.coordinator.DispatchをreadTimestamp.WithDispatchVoucher(ctx)+kv.DispatchWithReadTimestampへ置き換えてください。adapter/s3_admin.go#L444-L483: Line 474 のs.coordinator.Dispatchを同じ形へ置き換えてください。adapter/s3_upload_part.go#L30-L39: Line 35 で得たkv.ReadTimestampをs3UploadPartStateへ保存しcommitS3UploadPartまで引き回すか、allocateS3UploadPartVersion（Line 144）の 1 回の確保へ統合してください。
adapter/s3_admin.goのadminCreateBucketTxn（Line 298-299）が正しい形です。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/s3_admin.go` around lines 333 - 361, 消費されずに残る applied-read voucher を各 S3 経路で dispatch に引き渡してください。adapter/s3_admin.go の 333-361（該当する ACL 更新処理）と 444-483 の Dispatch を、adminCreateBucketTxn の実装にならい、readTimestamp.WithDispatchVoucher(ctx) と kv.DispatchWithReadTimestamp に置き換えてください。adapter/s3_upload_part.go の 30-39 では取得した kv.ReadTimestamp を s3UploadPartState に保存して commitS3UploadPart まで引き回すか、allocateS3UploadPartVersion の単一の確保に統合し、予約した voucher が最終 dispatch で消費されるようにしてください。adapter/dynamodb_schema.go (1)
206-217: 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win重大: dispatch のエラーがシャドーイングされ、失敗が成功として返ります。
行208 は
if _, err := ...; err == nilでerrをifスコープ内に新規宣言します。このerrは行209 のreturn nilにしか使われません。行211 の
isRetryableTransactWriteError(err)は、行208 の dispatch エラーではなく、外側スコープのerrを読みます。外側のerrは行186／191／198／202 の各チェックを通過した時点で必ずnilです。結果として次の動作になります。
- 行211 は常に
isRetryableTransactWriteError(nil)を評価します。nilが retryable と判定されない場合、行212 はerrors.WithStack(nil)を返します。これはnilです。- 呼び出し元は、テーブルが作成されていないのに
CreateTableが成功したと判断します。nilが retryable と判定される場合、実際のエラー種別に関係なく必ず再試行し、非再試行エラーでも試行回数を使い切ります。同じファイルの
deleteTableWithRetry（行323-330）はifスコープ内でエラーを完結して処理しており、正しい形です。create 側も同じ形に揃えてください。🐛 提案する修正
req.StartTS = readTS dispatchCtx := readTimestamp.WithDispatchVoucher(ctx) - if _, err := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); err == nil { - return nil - } - if !isRetryableTransactWriteError(err) { - return errors.WithStack(err) + if _, dispatchErr := kv.DispatchWithReadTimestamp(dispatchCtx, d.coordinator, req); dispatchErr != nil { + if !isRetryableTransactWriteError(dispatchErr) { + return errors.WithStack(dispatchErr) + } + } else { + return nil } if err := waitRetryWithDeadline(ctx, deadline, backoff); err != nil { return errors.WithStack(err) }
adapter/phase_d_voucher_test.goのTestDynamoDBCreateTablePhaseDBindsReadVoucherは成功パスのみを検証します。dispatch 失敗時にエラーが伝播することを確認する回帰テストの追加も推奨します。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/dynamodb_schema.go` around lines 206 - 217, 修正対象は CreateTable のリトライ処理で、DispatchWithReadTimestamp のエラーが外側の err に正しく渡らず成功扱いされています。dispatch のエラーを外側スコープで保持し、成功時だけ return nil、失敗時はその同じエラーを isRetryableTransactWriteError と errors.WithStack に渡すよう、deleteTableWithRetry と同じ形に揃えてください。TestDynamoDBCreateTablePhaseDBindsReadVoucher には dispatch 失敗が呼び出し元へ伝播する回帰テストを追加してください。adapter/s3_upload_part.go (1)
30-39: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
prepareS3UploadPartが予約した voucher を破棄しています。Line 35 は
beginTxnReadTimestampを呼びますが、Line 39 はTimestamp()だけをstate.readTSへ保存し、kv.ReadTimestamp自体を捨てます。Phase-D ではBeginReadTimestampThroughが voucher を予約するため、この予約は消費されません。さらにallocateS3UploadPartVersion（Line 144）が同じアップロード処理でもう一度beginTxnReadTimestampを呼びます。1 リクエストで 2 つの read timestamp を確保する形になっています。
prepareS3UploadPartで確保したkv.ReadTimestampを state に保存してcommitS3UploadPartまで引き回すか、prepareS3UploadPart側の確保を廃止してallocateS3UploadPartVersionの 1 回に統合してください。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/s3_upload_part.go` around lines 30 - 39, Update prepareS3UploadPart and the surrounding upload flow so the read timestamp is acquired only once: either retain the full kv.ReadTimestamp returned by beginTxnReadTimestamp in s3UploadPartState and pass it through commitS3UploadPart, or remove that acquisition from prepareS3UploadPart and rely on allocateS3UploadPartVersion. Ensure the reserved voucher is consumed rather than discarded.
🧹 Nitpick comments (26)
kv/shard_store.go (1)
2339-2353: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
sort.Sliceの代わりにslices.Sortを使えます。
slicesは Line 10 で既に import 済みです。uint64のスライスに対してはslices.Sort(ids)が簡潔で、比較関数の割り当てもありません。同じ指摘は Line 2633 のLocalStoresにも当てはまります。♻️ 変更案
- sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + slices.Sort(ids)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/shard_store.go` around lines 2339 - 2353, Replace the sort.Slice call in tsoCommitFloorGroupIDs with slices.Sort(ids), preserving ascending uint64 ordering; apply the same simplification to the LocalStores implementation around line 2633.adapter/redis_lists.go (1)
102-106: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueリトライ理由の文字列が 2 つの経路で同一です。
listPushCore（Line 102）とlistPushCoreWithDedup（Line 398）はどちらも"redis list push: begin read timestamp"を渡します。この文字列が診断ログや監視に出る場合、dedup 有効／無効のどちらの経路かを区別できません。後者を"redis list push (dedup): begin read timestamp"のように分けると、障害解析が容易になります。Also applies to: 398-402
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/redis_lists.go` around lines 102 - 106, Update the retry-reason string passed to beginTxnReadTimestamp in listPushCoreWithDedup to identify the dedup path, using a distinct message such as “redis list push (dedup): begin read timestamp”; leave the non-dedup listPushCore message unchanged.kv/tso_fsm.go (2)
141-150: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
applyCutoverEntryの長さ検証は到達不能です。
Applyの分岐（Line 71）はbytes.Equal(data, []byte(tsoCutoverEnvelope))で既に完全一致を確認します。したがって Line 142 の再検証は常に偽になり、エラーパスは死にコードです。防御的に残す方針であれば問題ありませんが、applyPhaseDEntryと対称にするならApply側をbytes.HasPrefixにして長さ検証をapplyCutoverEntryに委ねる方が一貫します。♻️ 一貫性のための変更案
- case bytes.Equal(data, []byte(tsoCutoverEnvelope)): + case bytes.HasPrefix(data, []byte(tsoCutoverEnvelope)): return f.applyCutoverEntry(data)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/tso_fsm.go` around lines 141 - 150, Update the cutover dispatch in Apply to use bytes.HasPrefix instead of requiring bytes.Equal, allowing applyCutoverEntry to perform the complete envelope-length validation. Keep applyCutoverEntry’s exact validation and error handling unchanged, matching the applyPhaseDEntry flow.
458-467: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
IsVolatileOnlyPayloadの条件式は演算子優先度に依存しており読みにくいです。Go では
&&が||より強く束縛するため、現在の式は意図どおり「3 つの (長さ AND プレフィックス) 条件の OR」として評価されます。動作は正しいですが、括弧または early return を使うと将来の誤読を防げます。♻️ 可読性の改善案
func (f *TSOStateMachine) IsVolatileOnlyPayload(payload []byte) bool { if bytes.Equal(payload, []byte(tsoCutoverEnvelope)) { return true } - return len(payload) == hlcLeaseEntryLen && payload[0] == raftEncodeHLCLease || - len(payload) == len(tsoAllocationFloorEnvelope)+hlcLeasePayloadLen && - bytes.HasPrefix(payload, []byte(tsoAllocationFloorEnvelope)) || - len(payload) == len(tsoPhaseDEnvelope)+hlcLeasePayloadLen && - bytes.HasPrefix(payload, []byte(tsoPhaseDEnvelope)) + if len(payload) == hlcLeaseEntryLen && payload[0] == raftEncodeHLCLease { + return true + } + if len(payload) == len(tsoAllocationFloorEnvelope)+hlcLeasePayloadLen && + bytes.HasPrefix(payload, []byte(tsoAllocationFloorEnvelope)) { + return true + } + return len(payload) == len(tsoPhaseDEnvelope)+hlcLeasePayloadLen && + bytes.HasPrefix(payload, []byte(tsoPhaseDEnvelope)) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/tso_fsm.go` around lines 458 - 467, Improve readability in TSOStateMachine.IsVolatileOnlyPayload by making the three OR-separated payload checks explicit with parentheses or early returns, rather than relying on &&/|| precedence. Preserve the existing matching behavior for HLC lease, allocation-floor, and phase-D payloads, as well as the tsoCutoverEnvelope check.adapter/redis_compat_helpers.go (1)
785-788: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
normalizeStartTSを再利用してください。Line 786-788 は
^uint64(0)センチネルを 0 に正規化するロジックを手書きしています。同じ処理を行うnormalizeStartTSヘルパーが既に存在し、rewriteListTxn（Line 1313）やredis_lists.goの各所で使われています。重複を避けると、センチネル判定の変更が 1 箇所で済みます。♻️ 変更案
- startTS := readTimestamp.Timestamp() - if startTS == ^uint64(0) { - startTS = 0 - } + startTS := normalizeStartTS(readTimestamp.Timestamp())🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/redis_compat_helpers.go` around lines 785 - 788, Replace the inline ^uint64(0)-to-zero normalization after readTimestamp.Timestamp() with the existing normalizeStartTS helper, preserving the resulting startTS behavior and reusing the shared sentinel handling.adapter/redis_zset_cmds.go (1)
1319-1321: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
r.blockWaitFallbackがゼロのとき 2 か所で扱いが異なります。
waitForBlockedCommandUpdate（Line 1400-1403）はゼロ以下の値をdefaultRedisBlockWaitFallbackに置換します。一方 Line 1350 のfast = signaled && time.Since(lastFullCheck) < r.blockWaitFallbackは生の値を使うため、ゼロだと常にfalseになり fast モードが無効化されます。動作は安全側（常に full check）に倒れるため、現状のバグではありません。ただし将来の読み手が混乱します。解決済みの fallback 値を 1 か所で算出し、両方で使う方が明確です。
Also applies to: 1349-1350, 1399-1403
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/redis_zset_cmds.go` around lines 1319 - 1321, Normalize r.blockWaitFallback once to defaultRedisBlockWaitFallback when it is zero or negative, then reuse that resolved duration in both the fast-mode check and waitForBlockedCommandUpdate. Ensure both paths apply identical fallback semantics without changing the existing safety behavior.adapter/redis_delta_compactor.go (1)
613-628: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
dispatchCompactionは voucher を context 経由で暗黙に受け取ります。シグネチャは
readTS uint64だけを受け取り、voucher はctxに埋め込まれた状態を前提とします。呼び出し元がWithDispatchVoucherを忘れると、コンパイルは通り、voucher なしで dispatch されます。kv.ReadTimestampを明示的な引数にすると、この失敗を型で防げます。ファイル内の他の Redis アダプター（例:dispatchAndSignalZSet）は既にこの形式を採用しています。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/redis_delta_compactor.go` around lines 613 - 628, Update DeltaCompactor.dispatchCompaction to accept an explicit kv.ReadTimestamp argument instead of relying on a voucher embedded in context, and pass that value through the timestamp allocation and operation-group dispatch. Update all callers, including the compaction dispatch path, to provide the ReadTimestamp explicitly while preserving existing normalization behavior.distribution/catalog_test.go (1)
368-373: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
snapshot.Routes[0]の前に長さを検証してください。
SnapshotAtが空のルート集合を返した場合、Line 372 は index out of range で panic します。テストは失敗しますが、原因がスタックトレースからしか読み取れません。require.Lenを先に置くと失敗メッセージが明確になります。💚 変更案
require.Equal(t, first.Version, snapshot.Version) require.Equal(t, firstTS, snapshot.ReadTS) + require.Len(t, snapshot.Routes, 1) require.Equal(t, uint64(1), snapshot.Routes[0].RouteID)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@distribution/catalog_test.go` around lines 368 - 373, Validate the route collection length before indexing it in the SnapshotAt test. Add a require.Len assertion for snapshot.Routes before the snapshot.Routes[0].RouteID check, preserving the existing assertions and expected route count.kv/tso_raft.go (3)
940-975: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winシャドウ再試行ループの上限を検討してください。
nextAfterはレガシー候補が TSO フロア以下の間、無制限に再試行します。ctx のチェックはありますが、ctx に期限がない場合、フロアが常に先行する状況ではループが長時間続きます。ログも 1 反復ごとに出ます。反復回数の上限、またはバックオフの追加を検討してください。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/tso_raft.go` around lines 940 - 975, Add a bounded retry policy or backoff to the retry loop in ShadowTimestampAllocator.nextAfter when legacyTS does not exceed reservation.PreviousAllocationFloor, while preserving context cancellation handling and successful return paths. Avoid unbounded per-iteration warning logs by limiting or throttling the corresponding discard logging.
665-679: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
uint32(n)の二重変換をまとめてください。Line 666 と Line 674 は同じ
uint32(n)を別々に計算します。nはvalidateTSOMinimumWindowでmaxHLCBatchSize以下に検証済みなので、静的解析の桁あふれ警告は誤検知です。可読性のため、変換を 1 回にまとめてください。♻️ 提案する変更
+ count := uint32(n) //nolint:gosec // n is bounded by maxHLCBatchSize. resp, err := pb.NewDistributionClient(conn).GetTimestamp(ctx, &pb.GetTimestampRequest{ - Count: uint32(n), //nolint:gosec // n is bounded by maxHLCBatchSize. + Count: count, MinTimestamp: min, ActivateCutover: activate, ActivatePhaseD: activatePhaseD, }) if err != nil { return empty, errors.Wrap(err, "tso request leader batch") } - count := uint32(n) //nolint:gosec // n is validated before this request.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/tso_raft.go` around lines 665 - 679, In the GetTimestamp request flow, compute the validated n-to-uint32 conversion once and reuse that variable for both the request Count field and the response count comparison. Keep the existing overflow-suppression justification on the single conversion and preserve the protocol validation behavior.Source: Linters/SAST tools
44-46: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winインターフェースの引数に名前を付けてください。
ReserveBatchAfter(context.Context, int, uint64, bool, bool)は bool が 2 つ連続します。呼び出し側はactivateCutoverとactivatePhaseDの順序を取り違えやすいです。順序を誤ると Phase D が意図せず起動します。引数名を宣言に付けると誤用を減らせます。♻️ 提案する変更
type TSOReservationAllocator interface { - ReserveBatchAfter(context.Context, int, uint64, bool, bool) (TSOReservation, error) + ReserveBatchAfter( + ctx context.Context, + n int, + min uint64, + activateCutover bool, + activatePhaseD bool, + ) (TSOReservation, error) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/tso_raft.go` around lines 44 - 46, TSOReservationAllocator の ReserveBatchAfter 宣言に全引数の名前を追加し、特に2つの bool を activateCutover、activatePhaseD の順序で明示してください。既存の型と戻り値は変更せず、呼び出し側が引数の意図を確認できる宣言に更新してください。kv/tso_test.go (1)
333-380: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
BeginReadTimestampThroughの 4 つのテストをテーブル駆動にまとめてください。4 つのテストは同じ構造です。差分は
phaseDActive、phaseDRequired、validateErrと期待値だけです。テーブル駆動にすると重複が減ります。また、このファイルの他のテストはt.Parallel()を使いますが、Line 333-407 のテストは使っていません。整合させてください。コーディングガイドラインに従っています: "Unit tests must be co-located with packages in
*_test.gofiles; prefer table-driven test cases"。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/tso_test.go` around lines 333 - 380, 4つのBeginReadTimestampThroughテストを1つのテーブル駆動テストに統合し、各ケースでphaseDActive、phaseDRequired、validateErrと期待する戻り値・呼び出し回数・検証値を定義してください。各サブテストでは既存の検証内容を維持し、t.Parallel()を追加してファイル内のテスト慣行に合わせてください。Source: Coding guidelines
adapter/redis_txn.go (1)
471-495: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
cancelOnceは不要です。エラーの選択も非決定的です。2つの点があります。
context.CancelFuncは複数回呼び出しても安全です。sync.Onceによる保護は不要です。cancel()を直接呼べます。- 失敗したゴルーチンが
cancel()を呼ぶと、残りのゴルーチンはcontext.CanceledをerrChに送ります。for err := range errChは最初に読み取った1件を返します。チャネルの受信順は非決定的なため、本来の原因エラーではなくcontext.Canceledが返る場合があります。障害調査が難しくなります。
errgroup.WithContextを使うと、最初に発生したエラーだけが保持され、この両方が解消されます。♻️ errgroup を使う修正案
- leaseCtx, cancel := context.WithCancel(ctx) - defer cancel() - - errCh := make(chan error, len(groupKeys)) - var wg sync.WaitGroup - var cancelOnce sync.Once + eg, leaseCtx := errgroup.WithContext(ctx) for _, key := range groupKeys { - wg.Add(1) - go func(k []byte) { - defer wg.Done() + k := key + eg.Go(func() error { if _, err := kv.LeaseReadForKeyThrough(r.coordinator, leaseCtx, k); err != nil { - errCh <- errors.WithStack(err) - cancelOnce.Do(cancel) + return errors.WithStack(err) } - }(key) + return nil + }) } - wg.Wait() - close(errCh) - for err := range errCh { - if err != nil { - return err - } - } - return nil + return errors.WithStack(eg.Wait()) }
golang.org/x/sync/errgroupの import が必要です。syncの import が他で使われていない場合は削除してください。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/redis_txn.go` around lines 471 - 495, Replace the manual WaitGroup, error channel, and cancelOnce logic in the transaction lease-read flow with errgroup.WithContext, using the derived group context for LeaseReadForKeyThrough and returning the group’s first error. Remove now-unused synchronization imports and preserve cancellation of remaining reads after the first failure.adapter/dynamodb_item_write.go (1)
449-456: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoff
WithDispatchVoucherの呼び出しが二重になっています。
commitItemWriteはreadTimestamp.WithDispatchVoucher(ctx)で voucher を context に載せ、その context をkv.DispatchWithReadTimestampに渡します。kv.DispatchWithReadTimestampは内部で context からReadTimestampを取り出し、prepareした後に再度WithDispatchVoucherを呼びます。動作は正しいですが、呼び出し側が context への詰め込みを知っている必要があります。
kv.DispatchWithReadTimestampがReadTimestampを引数で受け取る形にすると、この結合がなくなります。この変更はkv/tso.goと全アダプター呼び出し側に及ぶため、別 PR での対応でも問題ありません。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/dynamodb_item_write.go` around lines 449 - 456, commitItemWrite 内の readTimestamp.WithDispatchVoucher 呼び出しを削除し、kv.DispatchWithReadTimestamp が ReadTimestamp を引数として受け取る API に変更して、内部で必要な context 処理を一元化してください。kv/tso.go の実装と全アダプター呼び出し側を更新し、context への voucher 設定を呼び出し側に要求しない形を維持してください。main.go (1)
2122-2147: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
floorProviders ...kv.TSOCutoverFloorProviderを単一引数にしてください。
configureCoordinatorTSOは可変長引数を受け取りますが、L2143-2145 で先頭の1要素だけを使い、残りを黙って破棄します。呼び出し側（L530）も1つだけ渡します。可変長にすると、2つ目以降を渡した呼び出しがコンパイルエラーにならず、実行時に無視されます。
floorProvider kv.TSOCutoverFloorProviderの単一引数に変更すると、この曖昧さがなくなります。♻️ 修正案
func configureCoordinatorTSO( coordinate *kv.ShardedCoordinator, shardGroups map[uint64]*kv.ShardGroup, - floorProviders ...kv.TSOCutoverFloorProvider, + floorProvider kv.TSOCutoverFloorProvider, ) (coordinatorTSOWiring, error) { @@ tsoGroup, dedicated := shardGroups[dedicatedTSORaftGroupID] if !dedicated { return configureLegacyCoordinatorTSO(coordinate) } - var floorProvider kv.TSOCutoverFloorProvider - if len(floorProviders) > 0 { - floorProvider = floorProviders[0] - } return configureDedicatedCoordinatorTSO(coordinate, tsoGroup, floorProvider) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@main.go` around lines 2122 - 2147, Change configureCoordinatorTSO to accept a single kv.TSOCutoverFloorProvider parameter instead of the variadic floorProviders argument, and pass that value directly to configureDedicatedCoordinatorTSO. Update all callers, including the existing call site, to provide the single floor provider explicitly.kv/tso_raft_test.go (1)
100-100: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
//nolint:gosecの代わりに変換ヘルパーを使えます。このファイルには HLC タイムスタンプ生成のための
//nolint:gosecが4箇所あります。同ファイル内ではpositiveIntToUint64のような変換ヘルパーが既に使われています。ミリ秒値をuint64へ変換する小さなテストヘルパー（例:testHLCFromWallMillis(t time.Time) uint64）を1つ用意すると、4箇所の//nolintを削除できます。コーディングガイドラインは
//nolintの追加を避け、リファクタを優先することを求めています。Also applies to: 287-287, 301-301, 372-372
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/tso_raft_test.go` at line 100, Replace the four HLC millisecond conversions currently suppressed with //nolint:gosec by adding one file-local helper such as testHLCFromWallMillis(t time.Time) uint64. Have the helper perform the Unix-millisecond-to-uint64 conversion, update all four call sites to use it, and remove the corresponding nolint directives while preserving the existing HLC timestamp behavior.Source: Coding guidelines
adapter/redis_hash_cmds.go (1)
567-598: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
*ReadTimestamp版の新設時に、旧関数の本体をそのまま複製しています。 どちらの新関数もディスパッチ呼び出し以外は旧関数と同一です。要素生成ロジックが二重化するため、TTL 形式やメタ形式の変更が片方だけに入るリスクがあります。要素生成を共通ヘルパーへ抽出し、各関数はディスパッチのみを担当させてください。
adapter/redis_hash_cmds.go#L567-L598:persistHashTxn（L532-565）と共通の要素生成ヘルパー（例:persistHashElems）を抽出し、両関数から呼び出してください。adapter/redis_collection_ttl.go#L151-L166:dispatchCollectionExpire（L135-149）と共通化してください。旧関数の呼び出し元が残っていない場合は旧関数を削除してください。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/redis_hash_cmds.go` around lines 567 - 598, 重複した要素生成ロジックを共通ヘルパーへ抽出し、adapter/redis_hash_cmds.go:567-598 の persistHashReadTimestampTxn と persistHashTxn の両方から利用して、各関数はディスパッチに専念させてください。adapter/redis_collection_ttl.go:151-166 の処理も dispatchCollectionExpire と要素生成を共通化し、旧関数の呼び出し元が残っていなければ削除してください。kv/sharded_coordinator_txn_test.go (2)
386-388: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
t.Parallel()の欠落を確認してください。このファイルで追加した他のテストはすべて
t.Parallel()を呼びます。TestReadTimestampVoucherBindingShadowsParentCapabilityだけ呼びません。allocとcoordはテストローカルで生成するため、並列実行しても他テストと状態を共有しません。意図的な除外でなければ、一貫性のために追加してください。♻️ 提案する修正
func TestReadTimestampVoucherBindingShadowsParentCapability(t *testing.T) { + t.Parallel() prePhaseDErr := errors.Join(ErrTSOTimestampInvalid, ErrTSOTimestampPrePhaseD)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/sharded_coordinator_txn_test.go` around lines 386 - 388, 「TestReadTimestampVoucherBindingShadowsParentCapability」に t.Parallel() を追加し、同ファイル内の他テストと並列実行の扱いを統一してください。
484-485: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
require.Nilではなくrequire.NoErrorを使ってください。
state.Applyはerrorを返します。require.NoErrorはエラー内容をメッセージに出力するため、失敗時の診断が容易になります。testifylintを有効にしている場合はrequire-error系の指摘対象にもなります。♻️ 提案する修正
- require.Nil(t, state.Apply(marshalTSOCutover())) - require.Nil(t, state.Apply(marshalTSOPhaseD(0))) + require.NoError(t, state.Apply(marshalTSOCutover())) + require.NoError(t, state.Apply(marshalTSOPhaseD(0)))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@kv/sharded_coordinator_txn_test.go` around lines 484 - 485, Replace the require.Nil assertions around state.Apply in the affected test with require.NoError, preserving both Apply calls and their arguments so failures report the returned error details.adapter/redis_expire_cmds.go (1)
95-99: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winストレージエラーと「キー無し」を区別してください。
readRedisStringAtが返すすべてのエラーを「キーが存在しない」として扱います。この結果、Pebble の I/O エラーやデコード失敗でも GETDEL は削除を行わず、クライアントにはnilを返します。呼び出し元はエラーを検出できません。store.ErrKeyNotFoundだけを「キー無し」に分類し、それ以外は伝播させてください。🛡️ 提案する修正
raw, _, err := r.readRedisStringAt(key, readTS) if err != nil { - // Key may have expired or been deleted between type check and read. - return nil, false, nil //nolint:nilerr // treat not-found/expired as nil value + // Key may have expired or been deleted between type check and read. + if cockerrors.Is(err, store.ErrKeyNotFound) { + return nil, false, nil + } + return nil, false, cockerrors.WithStack(err) } return raw, true, nilコーディングガイドラインは
//nolintの追加を避け、リファクタリングを優先することを求めています。この変更はnolintも同時に除去します。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/redis_expire_cmds.go` around lines 95 - 99, readRedisStringAt のエラー処理を更新し、store.ErrKeyNotFound の場合だけキー無しとして nil 値を返し、それ以外のストレージエラーやデコードエラーは呼び出し元へ伝播させてください。GETDEL の削除処理と戻り値が非NotFoundエラーを隠さないよう維持し、該当ブロックの //nolint:nilerr も削除してください。Source: Coding guidelines
adapter/redis_stream_cmds.go (1)
1334-1346: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low value
r.readTS()をループの外へ移動してください。
xreadCheckTypesは各キーごとにr.readTS()を呼びます。型チェックは 1 つのスナップショットで行うほうが一貫します。またキー数に比例した呼び出しを避けられます。♻️ 修正案
func (r *RedisServer) xreadCheckTypes(ctx context.Context, req xreadRequest) error { + readTS := r.readTS() for _, key := range req.keys { - readTS := r.readTS() typ, err := r.keyTypeAtExpect(ctx, key, readTS, redisTypeStream)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/redis_stream_cmds.go` around lines 1334 - 1346, Move the single r.readTS() call in xreadCheckTypes outside the key loop, store its result, and reuse that snapshot for every keyTypeAtExpect invocation while preserving the existing error and wrong-type handling.adapter/s3_hlc_fence_test.go (2)
259-259: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
context.Contextを第 1 引数へ移動してください。Go の慣習では
context.Contextが第 1 引数です。テストヘルパーではt *testing.Tの後ろに置くとreviveのcontext-as-argumentに抵触する場合があります。seedS3ObjectForReadVoucherTest(ctx, t, st, bucket, key)の順序、またはctxを引数から外して内部でt.Context()を使う形を検討してください。コーディングガイドラインに従い、
.golangci.yamlのリンター設定に適合させてください。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/s3_hlc_fence_test.go` at line 259, Update seedS3ObjectForReadVoucherTest so context.Context is the first parameter, using the signature and all call sites in the order seedS3ObjectForReadVoucherTest(ctx, t, st, bucket, key). Ensure the helper and its callers comply with the context-as-argument lint rule.Source: Coding guidelines
221-222: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueこの 2 つのテストに
t.Parallel()がありません。同じファイル内の他のテストはすべて
t.Parallel()を呼びます。TestS3CommitUploadPartRechecksUploadAtLatestAppliedWatermarkとTestS3CommitUploadPartIncludesUploadMetaInReadSetは共有状態を持たないため、並列化できます。♻️ 修正案
func TestS3CommitUploadPartRechecksUploadAtLatestAppliedWatermark(t *testing.T) { + t.Parallel() + st := store.NewMVCCStore()Also applies to: 240-241
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/s3_hlc_fence_test.go` around lines 221 - 222, 対象の2テスト、TestS3CommitUploadPartRechecksUploadAtLatestAppliedWatermark と TestS3CommitUploadPartIncludesUploadMetaInReadSet の先頭で t.Parallel() を呼び出し、同一ファイル内の他テストと同様に並列実行できるようにする。main_tso_routing_test.go (1)
248-259: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueエラー生成を
cockroachdb/errorsに揃えることを検討してください。行255 は
fmt.Errorfを使います。同等のモックであるkv/tso_raft_test.goのrecordingTSOEngineはerrors.Newfを使います。リポジトリのガイドラインはエラーをgithub.com/cockroachdb/errorsで扱うことを求めます。テストコードでも表記を揃えると、スタックトレースの扱いが一貫します。♻️ 提案する変更
- return nil, fmt.Errorf("unexpected TSO apply result %T", result) + return nil, errors.Newf("unexpected TSO apply result %T", result)インポートを差し替えてください。
- "fmt" "sync/atomic" "testing" "time" + "github.com/cockroachdb/errors" "github.com/bootjp/elastickv/distribution"コーディングガイドラインの「Wrap errors with
github.com/cockroachdb/errorsat boundaries in Go code」に基づく指摘です。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@main_tso_routing_test.go` around lines 248 - 259, Update mainTSOEngine.Propose to use github.com/cockroachdb/errors for the unexpected TSO apply result instead of fmt.Errorf, and adjust imports accordingly. Preserve the existing error message and type formatting.Source: Coding guidelines
adapter/grpc.go (1)
146-149: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
GroupCommittedTimestampFloorのエラーを一律FailedPreconditionに変換しています。
GroupCommittedTimestampFloorは複数の原因でエラーを返します。kv/tso_floor_test.goはローカルリーダーシップ不足時のErrTSOCommitFloorUnavailableと、境界付き ReadIndex を確認しています。したがって context のキャンセルやデッドライン超過も到達しえます。現状はすべて
codes.FailedPreconditionになります。クライアントはリーダー未確定（再試行して別ノードへ）と、呼び出しタイムアウト（バックオフして同一ノードへ）を区別できません。kv.NewLeaderRoutedTSOAllocatorのフェンス処理はこの応答でリーダー到達可否を判断するため、判定精度が下がります。原因に応じたコードへ分岐することを検討してください。
♻️ 提案する変更
ts, err := reader.GroupCommittedTimestampFloor(ctx, groupID) if err != nil { + switch { + case errors.Is(err, context.Canceled): + return nil, errors.WithStack(status.Error(codes.Canceled, err.Error())) + case errors.Is(err, context.DeadlineExceeded): + return nil, errors.WithStack(status.Error(codes.DeadlineExceeded, err.Error())) + } return nil, errors.WithStack(status.Error(codes.FailedPrecondition, err.Error())) }また、行148 は
err.Error()をそのままクライアントへ返します。内部エラー文字列の露出範囲が要件に合うかを確認してください。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/grpc.go` around lines 146 - 149, Update the error handling around GroupCommittedTimestampFloor to map context cancellation and deadline errors to the corresponding gRPC status codes, while preserving FailedPrecondition for ErrTSOCommitFloorUnavailable and other leader-readiness failures. Keep the existing wrapped error behavior as appropriate, but avoid exposing raw internal err.Error() text to clients unless the API contract requires it.adapter/distribution_server_test.go (1)
263-274: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winこのテストは「legacy server の拒否」を証明できていない可能性があります。
TestDistributionServerGetTimestamp_LeaderRoutedRejectsLegacyServerは 75ms のデッドラインを設定し、context.DeadlineExceededを期待します。デッドライン超過は複数の原因で発生します。
- サーバーが
FailedPreconditionを返し、LeaderRoutedTSOAllocatorが再試行を続けてデッドラインに達した（意図した挙動）。- 負荷の高い CI 環境で、gRPC の初回ダイヤルまたは1回目の RPC 自体が 75ms を超えた（意図しない経路）。
どちらの場合も同じ結果になるため、テストは fail-closed 動作を区別できません。
サーバー側のリクエスト受信回数を記録するか、
serverAllocを持たないサーバーが返すエラー種別を直接検証する形に変更することを検討してください。デッドラインに依存しない検証であればt.Parallel()も追加できます。🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@adapter/distribution_server_test.go` around lines 263 - 274, Update TestDistributionServerGetTimestamp_LeaderRoutedRejectsLegacyServer to verify the legacy server rejection directly instead of inferring it from a 75ms context deadline. Assert the server-side error/status or track and validate the request count so the test distinguishes FailedPrecondition rejection from dial or RPC timing; then remove the deadline-dependent assertion and add t.Parallel() if the revised test is independent.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@adapter/distribution_server.go`:
- Around line 244-252: Update the ValidateTimestamp response construction around
ValidateDurableTimestamp to derive PhaseDActive from the allocator state instead
of always setting it to true. Extend the optional timestamp allocator interface
with PhaseDActive() bool, assign the returned value to resp.PhaseDActive
alongside PhaseDFloor() and AllocationFloor(), and preserve the existing
fallback behavior when the allocator does not implement the interface.
In `@adapter/redis_delta_compactor.go`:
- Around line 316-324: 各 dispatch ごとに ReadTimestamp voucher
を再取得するよう、adapter/redis_delta_compactor.go:316-324 の SyncOnce と
compactBackgroundHandlers の処理を更新してください。dispatch 直前に BeginReadTimestampThrough
を呼び、各呼び出しから voucher 付き context を生成して、同じ tickCtx や券を複数 dispatch で再利用しないでください。関連する
adapter/redis_lists.go:221-222 および adapter/redis_compat_helpers.go:781-784 でも同じ
dispatch 単位の取得・context 生成に統一してください。
In `@adapter/s3.go`:
- Around line 2514-2530: Replace direct BeginReadTimestampThrough-based
dispatches with kv.DispatchWithReadTimestamp using the read timestamp’s
WithDispatchVoucher(ctx). Update AdminPutBucketAcl and the admin delete/other
dispatches in adapter/s3_admin.go (lines 354, 474, and 518), cleanup/delete
dispatches in adapter/s3.go (lines 1530, 1601, and 1850), and multipart admin
dispatch in adapter/s3_admin_objects.go (line 266); adapter/s3.go lines
2514-2530 requires the corresponding read-timestamp voucher flow, and
adapter/sqs_messages.go lines 1002-1007 requires the same replacement. Preserve
each operation’s existing request and error handling.
In `@docs/design/2026_04_16_partial_centralized_tso.md`:
- Around line 3-7: Resolve the lifecycle marker inconsistency for the design
document: either rename it to the implemented marker when M1–M7 are the complete
milestone set, updating every reference found by searching for the old filename,
or define the remaining runtime reload and production latency/alerting work as
M8 in the milestone table so the partial marker remains accurate. Keep the
status text, Remaining section, and milestone table consistent with the chosen
policy.
- Around line 62-63: Increase the leading indentation of the continuation line
beginning “malformed control entries” to four spaces so it remains part of
numbered item 10, matching the indentation of the other continuation lines.
In `@kv/keyviz_label.go`:
- Around line 124-130: Add an exported capability check on
keyVizLabeledCoordinator, such as supportsAppliedReadVoucher, that reports
whether inner implements AppliedReadTimestampVoucher rather than relying on the
wrapper’s method set. Update validateAppliedReadTimestamp in tso.go to use this
check and reject unsupported coordinators before BeginReadTimestampThrough
returns a voucher-bearing ReadTimestamp, while preserving the existing dispatch
behavior.
In `@kv/shard_store.go`:
- Around line 2419-2430: Update the fallback flow around
GroupCommittedTimestampFloor and leaderAddrFromEngine so that, after a local
leader fence failure, it detects when the resolved leader address belongs to the
current node and immediately returns the original fence error instead of
forwarding an RPC to itself. Preserve the existing remote-leader forwarding
behavior for addresses belonging to other nodes.
In `@kv/tso_fsm.go`:
- Around line 433-439: Update hasHeaderlessLegacyKVFSMSnapshotPayload to
identify headerless legacy snapshots using the recognized magic headers
EKVMVCC2, EKVPBBL1, or EKVSSTI1, matching restoreLegacyKVFSMSnapshot, rather
than relying only on len(peeked) > tsoSnapshotV4Len. Preserve the existing peek
error handling and return false for oversized payloads without a recognized
legacy header.
In `@kv/tso.go`:
- Around line 179-199: Define and enforce a clear voucher-use contract: in
kv/tso.go lines 179-199, either implement the prepared-count limit using
ErrTSOReadVoucherLimit or remove prepared and the unused error and document
unlimited ReadTimestamp reuse; if enforcing a limit, update
adapter/sqs_reaper.go lines 76-93 so each deletion dispatch obtains a
ReadTimestamp without exceeding that limit, such as acquiring one per record.
---
Outside diff comments:
In `@adapter/dynamodb_schema.go`:
- Around line 206-217: 修正対象は CreateTable のリトライ処理で、DispatchWithReadTimestamp
のエラーが外側の err に正しく渡らず成功扱いされています。dispatch のエラーを外側スコープで保持し、成功時だけ return
nil、失敗時はその同じエラーを isRetryableTransactWriteError と errors.WithStack
に渡すよう、deleteTableWithRetry
と同じ形に揃えてください。TestDynamoDBCreateTablePhaseDBindsReadVoucher には dispatch
失敗が呼び出し元へ伝播する回帰テストを追加してください。
In `@adapter/s3_admin.go`:
- Around line 333-361: 消費されずに残る applied-read voucher を各 S3 経路で dispatch
に引き渡してください。adapter/s3_admin.go の 333-361（該当する ACL 更新処理）と 444-483 の Dispatch
を、adminCreateBucketTxn の実装にならい、readTimestamp.WithDispatchVoucher(ctx) と
kv.DispatchWithReadTimestamp に置き換えてください。adapter/s3_upload_part.go の 30-39 では取得した
kv.ReadTimestamp を s3UploadPartState に保存して commitS3UploadPart
まで引き回すか、allocateS3UploadPartVersion の単一の確保に統合し、予約した voucher が最終 dispatch
で消費されるようにしてください。
In `@adapter/s3_upload_part.go`:
- Around line 30-39: Update prepareS3UploadPart and the surrounding upload flow
so the read timestamp is acquired only once: either retain the full
kv.ReadTimestamp returned by beginTxnReadTimestamp in s3UploadPartState and pass
it through commitS3UploadPart, or remove that acquisition from
prepareS3UploadPart and rely on allocateS3UploadPartVersion. Ensure the reserved
voucher is consumed rather than discarded.
---
Nitpick comments:
In `@adapter/distribution_server_test.go`:
- Around line 263-274: Update
TestDistributionServerGetTimestamp_LeaderRoutedRejectsLegacyServer to verify the
legacy server rejection directly instead of inferring it from a 75ms context
deadline. Assert the server-side error/status or track and validate the request
count so the test distinguishes FailedPrecondition rejection from dial or RPC
timing; then remove the deadline-dependent assertion and add t.Parallel() if the
revised test is independent.
In `@adapter/dynamodb_item_write.go`:
- Around line 449-456: commitItemWrite 内の readTimestamp.WithDispatchVoucher
呼び出しを削除し、kv.DispatchWithReadTimestamp が ReadTimestamp を引数として受け取る API
に変更して、内部で必要な context 処理を一元化してください。kv/tso.go の実装と全アダプター呼び出し側を更新し、context への
voucher 設定を呼び出し側に要求しない形を維持してください。
In `@adapter/grpc.go`:
- Around line 146-149: Update the error handling around
GroupCommittedTimestampFloor to map context cancellation and deadline errors to
the corresponding gRPC status codes, while preserving FailedPrecondition for
ErrTSOCommitFloorUnavailable and other leader-readiness failures. Keep the
existing wrapped error behavior as appropriate, but avoid exposing raw internal
err.Error() text to clients unless the API contract requires it.
In `@adapter/redis_compat_helpers.go`:
- Around line 785-788: Replace the inline ^uint64(0)-to-zero normalization after
readTimestamp.Timestamp() with the existing normalizeStartTS helper, preserving
the resulting startTS behavior and reusing the shared sentinel handling.
In `@adapter/redis_delta_compactor.go`:
- Around line 613-628: Update DeltaCompactor.dispatchCompaction to accept an
explicit kv.ReadTimestamp argument instead of relying on a voucher embedded in
context, and pass that value through the timestamp allocation and
operation-group dispatch. Update all callers, including the compaction dispatch
path, to provide the ReadTimestamp explicitly while preserving existing
normalization behavior.
In `@adapter/redis_expire_cmds.go`:
- Around line 95-99: readRedisStringAt のエラー処理を更新し、store.ErrKeyNotFound
の場合だけキー無しとして nil 値を返し、それ以外のストレージエラーやデコードエラーは呼び出し元へ伝播させてください。GETDEL
の削除処理と戻り値が非NotFoundエラーを隠さないよう維持し、該当ブロックの //nolint:nilerr も削除してください。
In `@adapter/redis_hash_cmds.go`:
- Around line 567-598:
重複した要素生成ロジックを共通ヘルパーへ抽出し、adapter/redis_hash_cmds.go:567-598 の
persistHashReadTimestampTxn と persistHashTxn
の両方から利用して、各関数はディスパッチに専念させてください。adapter/redis_collection_ttl.go:151-166 の処理も
dispatchCollectionExpire と要素生成を共通化し、旧関数の呼び出し元が残っていなければ削除してください。
In `@adapter/redis_lists.go`:
- Around line 102-106: Update the retry-reason string passed to
beginTxnReadTimestamp in listPushCoreWithDedup to identify the dedup path, using
a distinct message such as “redis list push (dedup): begin read timestamp”;
leave the non-dedup listPushCore message unchanged.
In `@adapter/redis_stream_cmds.go`:
- Around line 1334-1346: Move the single r.readTS() call in xreadCheckTypes
outside the key loop, store its result, and reuse that snapshot for every
keyTypeAtExpect invocation while preserving the existing error and wrong-type
handling.
In `@adapter/redis_txn.go`:
- Around line 471-495: Replace the manual WaitGroup, error channel, and
cancelOnce logic in the transaction lease-read flow with errgroup.WithContext,
using the derived group context for LeaseReadForKeyThrough and returning the
group’s first error. Remove now-unused synchronization imports and preserve
cancellation of remaining reads after the first failure.
In `@adapter/redis_zset_cmds.go`:
- Around line 1319-1321: Normalize r.blockWaitFallback once to
defaultRedisBlockWaitFallback when it is zero or negative, then reuse that
resolved duration in both the fast-mode check and waitForBlockedCommandUpdate.
Ensure both paths apply identical fallback semantics without changing the
existing safety behavior.
In `@adapter/s3_hlc_fence_test.go`:
- Line 259: Update seedS3ObjectForReadVoucherTest so context.Context is the
first parameter, using the signature and all call sites in the order
seedS3ObjectForReadVoucherTest(ctx, t, st, bucket, key). Ensure the helper and
its callers comply with the context-as-argument lint rule.
- Around line 221-222:
対象の2テスト、TestS3CommitUploadPartRechecksUploadAtLatestAppliedWatermark と
TestS3CommitUploadPartIncludesUploadMetaInReadSet の先頭で t.Parallel()
を呼び出し、同一ファイル内の他テストと同様に並列実行できるようにする。
In `@distribution/catalog_test.go`:
- Around line 368-373: Validate the route collection length before indexing it
in the SnapshotAt test. Add a require.Len assertion for snapshot.Routes before
the snapshot.Routes[0].RouteID check, preserving the existing assertions and
expected route count.
In `@kv/shard_store.go`:
- Around line 2339-2353: Replace the sort.Slice call in tsoCommitFloorGroupIDs
with slices.Sort(ids), preserving ascending uint64 ordering; apply the same
simplification to the LocalStores implementation around line 2633.
In `@kv/sharded_coordinator_txn_test.go`:
- Around line 386-388: 「TestReadTimestampVoucherBindingShadowsParentCapability」に
t.Parallel() を追加し、同ファイル内の他テストと並列実行の扱いを統一してください。
- Around line 484-485: Replace the require.Nil assertions around state.Apply in
the affected test with require.NoError, preserving both Apply calls and their
arguments so failures report the returned error details.
In `@kv/tso_fsm.go`:
- Around line 141-150: Update the cutover dispatch in Apply to use
bytes.HasPrefix instead of requiring bytes.Equal, allowing applyCutoverEntry to
perform the complete envelope-length validation. Keep applyCutoverEntry’s exact
validation and error handling unchanged, matching the applyPhaseDEntry flow.
- Around line 458-467: Improve readability in
TSOStateMachine.IsVolatileOnlyPayload by making the three OR-separated payload
checks explicit with parentheses or early returns, rather than relying on &&/||
precedence. Preserve the existing matching behavior for HLC lease,
allocation-floor, and phase-D payloads, as well as the tsoCutoverEnvelope check.
In `@kv/tso_raft_test.go`:
- Line 100: Replace the four HLC millisecond conversions currently suppressed
with //nolint:gosec by adding one file-local helper such as
testHLCFromWallMillis(t time.Time) uint64. Have the helper perform the
Unix-millisecond-to-uint64 conversion, update all four call sites to use it, and
remove the corresponding nolint directives while preserving the existing HLC
timestamp behavior.
In `@kv/tso_raft.go`:
- Around line 940-975: Add a bounded retry policy or backoff to the retry loop
in ShadowTimestampAllocator.nextAfter when legacyTS does not exceed
reservation.PreviousAllocationFloor, while preserving context cancellation
handling and successful return paths. Avoid unbounded per-iteration warning logs
by limiting or throttling the corresponding discard logging.
- Around line 665-679: In the GetTimestamp request flow, compute the validated
n-to-uint32 conversion once and reuse that variable for both the request Count
field and the response count comparison. Keep the existing overflow-suppression
justification on the single conversion and preserve the protocol validation
behavior.
- Around line 44-46: TSOReservationAllocator の ReserveBatchAfter
宣言に全引数の名前を追加し、特に2つの bool を activateCutover、activatePhaseD
の順序で明示してください。既存の型と戻り値は変更せず、呼び出し側が引数の意図を確認できる宣言に更新してください。
In `@kv/tso_test.go`:
- Around line 333-380:
4つのBeginReadTimestampThroughテストを1つのテーブル駆動テストに統合し、各ケースでphaseDActive、phaseDRequired、validateErrと期待する戻り値・呼び出し回数・検証値を定義してください。各サブテストでは既存の検証内容を維持し、t.Parallel()を追加してファイル内のテスト慣行に合わせてください。
In `@main_tso_routing_test.go`:
- Around line 248-259: Update mainTSOEngine.Propose to use
github.com/cockroachdb/errors for the unexpected TSO apply result instead of
fmt.Errorf, and adjust imports accordingly. Preserve the existing error message
and type formatting.
In `@main.go`:
- Around line 2122-2147: Change configureCoordinatorTSO to accept a single
kv.TSOCutoverFloorProvider parameter instead of the variadic floorProviders
argument, and pass that value directly to configureDedicatedCoordinatorTSO.
Update all callers, including the existing call site, to provide the single
floor provider explicitly.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: e0a05717-d126-47c2-ac27-ade840a29c79
⛔ Files ignored due to path filters (3)
proto/distribution.pb.gois excluded by!**/*.pb.goproto/distribution_grpc.pb.gois excluded by!**/*.pb.goproto/service.pb.gois excluded by!**/*.pb.go
📒 Files selected for processing (65)
adapter/distribution_server.goadapter/distribution_server_test.goadapter/dynamodb_item_write.goadapter/dynamodb_locks.goadapter/dynamodb_migration.goadapter/dynamodb_onephase_dedup_test.goadapter/dynamodb_schema.goadapter/dynamodb_transact.goadapter/grpc.goadapter/grpc_test.goadapter/phase_d_voucher_test.goadapter/redis_collection_ttl.goadapter/redis_compat_helpers.goadapter/redis_delta_compactor.goadapter/redis_expire_cmds.goadapter/redis_hash_cmds.goadapter/redis_list_dedup_test.goadapter/redis_lists.goadapter/redis_lua_context.goadapter/redis_lua_phase_d_test.goadapter/redis_retry_test.goadapter/redis_set_cmds.goadapter/redis_stream_cmds.goadapter/redis_strings.goadapter/redis_txn.goadapter/redis_zset_cmds.goadapter/s3.goadapter/s3_admin.goadapter/s3_admin_objects.goadapter/s3_hlc_fence_test.goadapter/s3_multipart_complete.goadapter/s3_put_object.goadapter/s3_upload_part.goadapter/sqs_catalog.goadapter/sqs_fifo.goadapter/sqs_messages.goadapter/sqs_messages_batch.goadapter/sqs_purge.goadapter/sqs_reaper.goadapter/sqs_redrive.goadapter/sqs_tags.godistribution/catalog.godistribution/catalog_test.godocs/design/2026_04_16_partial_centralized_tso.mdkv/coordinator.gokv/keyviz_label.gokv/leader_routed_store_test.gokv/lease_warmup_test.gokv/shard_store.gokv/sharded_coordinator.gokv/sharded_coordinator_txn_test.gokv/tso.gokv/tso_floor_test.gokv/tso_fsm.gokv/tso_fsm_test.gokv/tso_raft.gokv/tso_raft_test.gokv/tso_test.gomain.gomain_encryption_admin.gomain_encryption_admin_test.gomain_encryption_rotate_on_startup_test.gomain_tso_routing_test.goproto/distribution.protoproto/service.proto
🚧 Files skipped from review as they are similar to previous changes (1)
- main_encryption_admin_test.go
| resp := &pb.ValidateTimestampResponse{Valid: true, PhaseDActive: true} | ||
| if state, ok := s.timestampAllocator.(interface { | ||
| PhaseDFloor() uint64 | ||
| AllocationFloor() uint64 | ||
| }); ok { | ||
| resp.PhaseDFloor = state.PhaseDFloor() | ||
| resp.AllocationFloor = state.AllocationFloor() | ||
| } | ||
| return resp, nil |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
PhaseDActive を無条件に true として返しています。
Line 244 は検証成功時に常に PhaseDActive: true を設定します。この値が正しいのは、ValidateDurableTimestamp が Phase D 非アクティブ時に必ず kv.ErrTSOPhaseDInactive を返す場合に限られます。この前提はコード上に明示されていません。
Line 245 のオプショナルインタフェースには既に PhaseDFloor() と AllocationFloor() があります。ここに PhaseDActive() bool を加えて実際の状態を返すか、少なくとも前提をコメントで明記してください。adapter/distribution_server_test.go:1411 には PhaseDActive() の実装が既にあります。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@adapter/distribution_server.go` around lines 244 - 252, Update the
ValidateTimestamp response construction around ValidateDurableTimestamp to
derive PhaseDActive from the allocator state instead of always setting it to
true. Extend the optional timestamp allocator interface with PhaseDActive()
bool, assign the returned value to resp.PhaseDActive alongside PhaseDFloor() and
AllocationFloor(), and preserve the existing fallback behavior when the
allocator does not implement the interface.
| tickCtx, cancel := context.WithTimeout(ctx, c.timeout) | ||
| defer cancel() | ||
| readTimestamp, err := kv.BeginReadTimestampThrough(tickCtx, c.coord, snapshotTS(c.coord.Clock(), c.st), | ||
| "redis delta compactor: begin read timestamp") | ||
| if err != nil { | ||
| return errors.WithStack(err) | ||
| } | ||
| tickCtx = readTimestamp.WithDispatchVoucher(tickCtx) | ||
| readTS := readTimestamp.Timestamp() |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# voucher の消費・解放契約を確定させる
set -euo pipefail
ast-grep outline kv/tso.go --items all
rg -n -C15 'appliedReadDispatchVoucher' --type=go
rg -n -C10 'func .*WithDispatchVoucher|func DispatchWithReadTimestamp|func BeginReadTimestampThrough|func .*beginTxnReadTimestamp' --type=goRepository: bootjp/elastickv
Length of output: 7100
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== kv/tso.go relevant lines =="
sed -n '80,190p' kv/tso.go
echo
echo "== dispatcher implementation and other tso functions =="
sed -n '195,330p' kv/tso.go
echo
echo "== adapter sites =="
sed -n '298,335p' adapter/redis_delta_compactor.go
sed -n '185,230p' adapter/redis_lists.go
sed -n '768,792p' adapter/redis_compat_helpers.go
echo
echo "== usage count / imports for ReadTimestamp voucher methods =="
rg -n '\.WithDispatchVouncer|WithDispatchVoucher|DispatchWithReadTimestamp|BeginReadTimestampThrough|RevokeAppliedReadTimestamp|VouchAppliedReadTimestamp' --type=goRepository: bootjp/elastickv
Length of output: 13753
各 dispatch で ReadTimestamp voucher を再取得してください。 DispatchWithReadTimestamp は同じ ReadTimestamp を渡されない限り券を発行せず、BeginReadTimestampThrough の券は dispatch 単位で消費される設計です。現在の実装では、SyncOnce の 1 券を複数 dispatch に再利用しています。compactBackgroundHandlers が dispatch を呼ぶ直前に BeginReadTimestampThrough を読み直し、SyncOnce の tickCtx は各 BeginReadTimestampThrough 呼び出しごと生成するか、dispatch 毎に券付き context を作り直してください。
📍 Affects 3 files
adapter/redis_delta_compactor.go#L316-L324(this comment)adapter/redis_lists.go#L221-L222adapter/redis_compat_helpers.go#L781-L784
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@adapter/redis_delta_compactor.go` around lines 316 - 324, 各 dispatch ごとに
ReadTimestamp voucher を再取得するよう、adapter/redis_delta_compactor.go:316-324 の
SyncOnce と compactBackgroundHandlers の処理を更新してください。dispatch 直前に
BeginReadTimestampThrough を呼び、各呼び出しから voucher 付き context を生成して、同じ tickCtx や券を複数
dispatch で再利用しないでください。関連する adapter/redis_lists.go:221-222 および
adapter/redis_compat_helpers.go:781-784 でも同じ dispatch 単位の取得・context 生成に統一してください。
| func (s *S3Server) beginTxnReadTimestamp(ctx context.Context, readTS uint64, label string) (kv.ReadTimestamp, error) { | ||
| if readTS == ^uint64(0) { | ||
| if alloc, ok := kv.TimestampAllocatorThrough(s.coordinator); ok { | ||
| if phaseD, phaseDOK := alloc.(kv.TSOPhaseDState); phaseDOK && (phaseD.PhaseDRequired() || phaseD.PhaseDActive()) { | ||
| readTS = 1 | ||
| readTimestamp, err := kv.BeginReadTimestampThrough(ctx, s.coordinator, readTS, label) | ||
| return readTimestamp, errors.WithStack(err) | ||
| } | ||
| } | ||
| } | ||
| startTS, err := s.txnStartTS(ctx, readTS) | ||
| if err != nil { | ||
| return kv.ReadTimestamp{}, err | ||
| } | ||
| readTimestamp, err := kv.BeginReadTimestampThrough(ctx, s.coordinator, startTS, label) | ||
| return readTimestamp, errors.WithStack(err) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# beginTxnReadTimestamp を使う関数内に残る直接 Dispatch を検出する。
rg -nP --type=go -l 'beginTxnReadTimestamp' | while IFS= read -r f; do
echo "=== $f ==="
rg -nP -C30 'beginTxnReadTimestamp\(' "$f" \
| rg -nP 'beginTxnReadTimestamp\(|coordinator\.Dispatch\(|DispatchWithReadTimestamp\(|WithDispatchVoucher\('
done
echo '=== all remaining direct dispatches in adapters ==='
rg -nP --type=go -C2 'coordinator\.Dispatch\(' -g 'adapter/**'Repository: bootjp/elastickv
Length of output: 154
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- locate files ---'
git ls-files | rg '(^adapter/s3\.go$|^adapter/s3_admin\.go$|^adapter/sqs_messages\.go$|^adapter/sqs_fifo\.go$|^.*sharded_coordinator_txn_test\.go$)$' || true
echo '--- search beginTxnReadTimestamp definitions/usages ---'
rg -nP --type=go -n 'beginTxnReadTimestamp|BeginReadTimestampThrough|WithDispatchVoucher|DispatchWithReadTimestamp|coordinator\.Dispatch\(' adapter kv || true
echo '--- relevant line ranges ---'
for f in $(git ls-files | rg '(^adapter/s3\.go$|^adapter/s3_admin\.go$|^adapter/sqs_messages\.go$|^adapter/sqs_fifo\.go$)'); do
echo "### $f"
wc -l "$f"
done
echo '### adapter/s3.go lines 2400-2560'
sed -n '2400,2560p' adapter/s3.go | nl -ba -v2400
echo '### adapter/sqs_messages.go lines 960-1035'
sed -n '960,1035p' adapter/sqs_messages.go | nl -ba -v960
echo '### adapter/s3_admin.go relevant slices'
rg -nP -C20 'AdminPutBucketAcl|beginTxnReadTimestamp|coordinator\.Dispatch\(|DispatchWithReadTimestamp\(|WithDispatchVoucher\(' adapter/s3_admin.go || trueRepository: bootjp/elastickv
Length of output: 28942
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '### S3 beginTxnReadTimestamp callers and direct dispatches'
grep -nP 'beginTxnReadTimestamp|DispatchWithReadTimestamp|coordinator\.Dispatch\(' adapter/s3.go
echo '--- lines 860-900 ---'
sed -n '860,900p' adapter/s3.go
echo '--- lines 620-690 ---'
sed -n '620,690p' adapter/s3.go
echo '--- lines 1520-1610 ---'
sed -n '1520,1610p' adapter/s3.go
echo '--- lines 1840-1860 ---'
sed -n '1840,1860p' adapter/s3.go
echo '### S3 admin beginTxnReadTimestamp caller and direct dispatch'
grep -nP 'AdminPutBucketAcl|beginTxnReadTimestamp|DispatchWithReadTimestamp|coordinator\.Dispatch\(' adapter/s3_admin.go
echo '--- lines 320-365 ---'
sed -n '320,365p' adapter/s3_admin.go
echo '### SQS messages receive path caller and lower dispatches'
grep -nP 'beginTxnReadTimestamp|scanAndDeliverPartition|DispatchWithReadTimestamp|coordinator\.Dispatch\(' adapter/sqs_messages.go
echo '--- lines 980-1045 ---'
sed -n '980,1045p' adapter/sqs_messages.go
echo '### SQS fifo dispatches'
grep -nP 'DispatchWithReadTimestamp|coordinator\.Dispatch\(' adapter/sqs_fifo.go
echo '### Admin objects relevant dispatch'
grep -nP 'beginTxnReadTimestamp|DispatchWithReadTimestamp|coordinator\.Dispatch\(' adapter/s3_admin_objects.go
echo '--- lines 240-280 ---'
sed -n '240,280p' adapter/s3_admin_objects.goRepository: bootjp/elastickv
Length of output: 18004
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '### kv tso relevant implementation'
sed -n '90,180p' kv/tso.go
echo '### kv test around phase D direct dispatch rejection'
sed -n '240,310p' kv/sharded_coordinator_txn_test.goRepository: bootjp/elastickv
Length of output: 6454
BeginReadTimestampThrough 経由の直接 dispatch を DispatchWithReadTimestamp に置き換えてください。 Phase-D 有効時、coord.Dispatch(ctx, req) は ErrTSOTimestampPrePhaseD で StartTS の有効性を拒否します。対応箇所は以下の直接 dispatch です。
adapter/s3_admin.go#L354:AdminPutBucketAclはreadTimestamp.WithDispatchVoucher(ctx)+kv.DispatchWithReadTimestampを使う必要がありますadapter/s3_admin.go#L474,#L518: admin delete/other 直接 dispatch も同様に置き換えてくださいadapter/s3.go#L1530,#L1601,#L1850: cleanup/delete 直接 dispatch も開始時 read timestamp の voucher で dispatch する必要がありますadapter/s3_admin_objects.go#L266: multipart admin direct dispatch も同様に置き換えてください
📍 Affects 2 files
adapter/s3.go#L2514-L2530(this comment)adapter/sqs_messages.go#L1002-L1007
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@adapter/s3.go` around lines 2514 - 2530, Replace direct
BeginReadTimestampThrough-based dispatches with kv.DispatchWithReadTimestamp
using the read timestamp’s WithDispatchVoucher(ctx). Update AdminPutBucketAcl
and the admin delete/other dispatches in adapter/s3_admin.go (lines 354, 474,
and 518), cleanup/delete dispatches in adapter/s3.go (lines 1530, 1601, and
1850), and multipart admin dispatch in adapter/s3_admin_objects.go (line 266);
adapter/s3.go lines 2514-2530 requires the corresponding read-timestamp voucher
flow, and adapter/sqs_messages.go lines 1002-1007 requires the same replacement.
Preserve each operation’s existing request and error handling.
| - Status: Partial — M1-M7 are implemented, including the dedicated group-0 | ||
| FSM, leader-routed durable windows, strict term bootstrap, serialized shadow | ||
| migration, one-way rolling cutover, durable Phase-D retirement, and | ||
| cross-shard SSI timestamp validation. Runtime config reload and production | ||
| latency/alerting work remain open. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
ライフサイクルマーカーの妥当性を確認してください。
行3 は「M1-M7 are implemented」と記載します。セクション8のマイルストーン表（行831-837）でも M1 から M7 までのすべてが shipped です。
コーディングガイドラインは、設計文書のファイル名マーカーを次のように定義します。*_partial_*.md は「some milestones shipped, full proposal not complete」、*_implemented_*.md は「all milestones shipped」です。
現在のファイル名は 2026_04_16_partial_centralized_tso.md です。すべてのマイルストーンが shipped であるため、2026_04_16_implemented_centralized_tso.md への変更が該当する可能性があります。
行144-147 の Remaining 項目（ランタイム設定リロード、本番ベンチマークとアラート閾値）をマイルストーンとして扱う方針であれば、セクション8の表に M8 として追加してください。そうすれば _partial_ の維持が表と整合します。どちらの方針を採るかを決めてください。
ファイル名を変更する場合は、他文書からの参照も更新が必要です。
#!/bin/bash
# この設計文書への参照箇所を洗い出す。
rg -n '2026_04_16_partial_centralized_tso' --glob '!**/node_modules/**'🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/design/2026_04_16_partial_centralized_tso.md` around lines 3 - 7,
Resolve the lifecycle marker inconsistency for the design document: either
rename it to the implemented marker when M1–M7 are the complete milestone set,
updating every reference found by searching for the old filename, or define the
remaining runtime reload and production latency/alerting work as M8 in the
milestone table so the partial marker remains accurate. Keep the status text,
Remaining section, and milestone table consistent with the chosen policy.
Source: Coding guidelines
| decoded and deterministically rejected without halting the TSO apply loop; | ||
| malformed control entries still halt fail-closed. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
リスト継続行のインデントが不足しています。
行63 の先頭インデントは3スペースです。項目10 は 10. で始まるため、継続行は4スペースのインデントが必要です。同じ項目の他の継続行（行54-62）は4スペースです。
3スペースのままでは、一部の Markdown レンダラが行63 を項目10 の継続として扱わず、独立した段落として表示します。
📝 提案する修正
decoded and deterministically rejected without halting the TSO apply loop;
- malformed control entries still halt fail-closed.
+ malformed control entries still halt fail-closed.
11. `RaftTSOAllocator` verifies group-0 leadership and commits every returned🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/design/2026_04_16_partial_centralized_tso.md` around lines 62 - 63,
Increase the leading indentation of the continuation line beginning “malformed
control entries” to four spaces so it remains part of numbered item 10, matching
the indentation of the other continuation lines.
| func (c keyVizLabeledCoordinator) VouchAppliedReadTimestamp(timestamp uint64, ref AppliedReadTimestampVoucherRef) error { | ||
| voucher, ok := c.inner.(AppliedReadTimestampVoucher) | ||
| if !ok { | ||
| return errors.WithStack(ErrTSOProtocolUnsupported) | ||
| } | ||
| return errors.WithStack(voucher.VouchAppliedReadTimestamp(timestamp, ref)) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
装飾側が常に AppliedReadTimestampVoucher を実装するため、失敗の検出が遅れます。
keyVizLabeledCoordinator はこのメソッドを常に持ちます。そのため kv/tso.go の validateAppliedReadTimestamp（Line 352）の型アサーションは、内側の Coordinator が voucher に非対応でも成功します。結果として BeginReadTimestampThrough は voucher 付きの ReadTimestamp を返し、ErrTSOProtocolUnsupported は dispatch 時まで表面化しません。読み取り開始時点で失敗させるほうが安全です。内側の対応可否を問い合わせる小さなヘルパー（例: supportsAppliedReadVoucher()）を公開し、validateAppliedReadTimestamp 側でそれを使う方法を検討してください。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@kv/keyviz_label.go` around lines 124 - 130, Add an exported capability check
on keyVizLabeledCoordinator, such as supportsAppliedReadVoucher, that reports
whether inner implements AppliedReadTimestampVoucher rather than relying on the
wrapper’s method set. Update validateAppliedReadTimestamp in tso.go to use this
check and reject unsupported coordinators before BeginReadTimestampThrough
returns a voucher-bearing ReadTimestamp, while preserving the existing dispatch
behavior.
| if isLeaderEngine(engine) { | ||
| if ts, err := s.GroupCommittedTimestampFloor(ctx, groupID); err == nil { | ||
| return ts, nil | ||
| } | ||
| // Leadership may have changed between State and ReadIndex. Resolve the | ||
| // newly published leader below instead of trusting the local watermark. | ||
| } | ||
| addr := leaderAddrFromEngine(engine) | ||
| if addr == "" { | ||
| return 0, errors.Wrapf(ErrTSOCommitFloorUnavailable, | ||
| "data group %d has no known leader", groupID) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
ローカルリーダーの fence 失敗時に自ノードへ RPC を再送する可能性があります。
Line 2419 でこのノードがリーダーだと判定し、GroupCommittedTimestampFloor の ReadIndex が失敗した場合、Line 2426 で leaderAddrFromEngine にフォールバックします。リーダー情報がまだ更新されていなければ、この値は自ノードのアドレスです。その場合、自ノードへ gRPC 往復してから同じ GroupCommittedTimestampFloor を再実行し、同じ理由で失敗します。無限ループにはなりませんが、TSO term 初期化の失敗検知が proxyForwardTimeout 分だけ遅延します。
解決したアドレスが自ノードのものであるかを判定し、その場合は元の fence エラーを直接返す方が速く失敗できます。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@kv/shard_store.go` around lines 2419 - 2430, Update the fallback flow around
GroupCommittedTimestampFloor and leaderAddrFromEngine so that, after a local
leader fence failure, it detects when the resolved leader address belongs to the
current node and immediately returns the original fence error instead of
forwarding an RPC to itself. Preserve the existing remote-leader forwarding
behavior for addresses belonging to other nodes.
| func hasHeaderlessLegacyKVFSMSnapshotPayload(br *bufio.Reader) (bool, error) { | ||
| peeked, err := br.Peek(tsoSnapshotV4Len + 1) | ||
| if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) { | ||
| return false, errors.Wrap(err, "tso fsm snapshot: peek headerless legacy payload") | ||
| } | ||
| return len(peeked) > tsoSnapshotV4Len, nil | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# ヘッダなし旧kvFSMスナップショットの実際の形式と長さ前提を確認する
rg -n -C5 'hasHeaderlessLegacyKVFSMSnapshotPayload|restoreHeaderlessLegacyKVFSMSnapshot' --type=go
rg -n -C8 'legacyKVFSMStoreSnapshotMagics|EKVMVCC2|EKVPBBL1|EKVSSTI1' --type=go
rg -n -C5 'hlcSnapshotMagic|isV1Magic|isV2Magic|isUnknownEKVTHLC' --type=goRepository: bootjp/elastickv
Length of output: 154
🏁 Script executed:
#!/bin/bash
set -u
printf 'repo files around kv/tso_fsm.go:\n'
git ls-files | rg '(^|/)tso_fsm\.go$|(^|/)kv/' | sed -n '1,120p'
printf '\nFind functions/consts names broadly:\n'
rg -n 'Legacy|legacy|Headerless|snapshot|tsoSnapshot|KVFSM|SnapshotPayload|\bEKV\b|HLC|Magic' . --glob '*.go' --glob '!vendor/**' --glob '!**/testdata/**' | sed -n '1,240p'Repository: bootjp/elastickv
Length of output: 27916
🏁 Script executed:
#!/bin/bash
set -u
printf 'tso_fsm.go size and relevant lines:\n'
wc -l kv/tso_fsm.go
sed -n '400,470p' kv/tso_fsm.go
printf '\nAll exact identifiers in all files:\n'
rg -n 'hasHeaderlessLegacyKVFSMSnapshotPayload|restoreHeaderlessLegacyKVFSMSnapshot|legacyKVFSMStoreSnapshotMagics|ReadTSOSnapshotState|readTSOSnapshotState|tsoSnapshotV4Len|EKVMVCC2|EKVPBBL1|EKVSSTI1|hlcSnapshotMagic|isV1Magic|isV2Magic|isUnknownEKVTHLC' . --glob '*.go' --glob '!vendor/**' --glob '!**/testdata/**' || trueRepository: bootjp/elastickv
Length of output: 10767
🏁 Script executed:
#!/bin/bash
set -u
sed -n '1,120p' kv/tso_fsm.go
sed -n '250,390p' kv/tso_fsm.go
sed -n '470,510p' kv/tso_fsm_test.go
printf '\nConstants across codebase:\n'
rg -n 'tsoSnapshotV[0-9]Len|hlcSnapshotMagic|PebbleSnapshotMagicLen|legacyKVFSMStoreSnapshotMagics|Restore\(' kv/tso_fsm.go kv/tso_fsm_test.go internal/raftengine/etcd/persisted_snapshot_export_test.goRepository: bootjp/elastickv
Length of output: 17276
🏁 Script executed:
#!/bin/bash
set -u
rg -n 'const|hlcLeasePayloadLen|hlcLeaseEntryLen|PebbleSnapshotMagicLen|ReadSnapshotHeader' kv internal/backup --glob '*.go'
printf '\nReadSnapshotHeader implementation:\n'
sed -n '95,155p' internal/backup/snapshot_reader.go
printf '\nStoreSnapshot implementations mentioning length/magic:\n'
sed -n '3060,3160p' store/lsm_store.go
sed -n '3215,3280p' store/lsm_store.goRepository: bootjp/elastickv
Length of output: 49557
🏁 Script executed:
#!/bin/bash
set -u
python3 - <<'PY'
from pathlib import Path
import re
p = Path('kv/tso_fsm.go')
text = p.read_text()
m = re.search(r'const\((.*?)\)', text, re.S)
const_block = m.group(1) if m else ''
print('const block contains tsoSnapshot constants:')
for part in ['tsoAllocationFloorEnvelope', 'tsoCutoverEnvelope', 'tsoPhaseDEnvelope', 'tsoSnapshotV1Len', 'tsoSnapshotV2Len', 'tsoSnapshotV3Len', 'tsoSnapshotV4Len']:
print(part, bool(re.search(rf'\b{part}\b(?!.*=)', const_block)))
# Behavioral probe of current length-only predicate for truncated/corrupt payloads.
# hlcLeasePayloadLen = 8; tsoSnapshotV4Len computed from source constants below.
hlc_lease_payload_len = 8
tso_snapshot_v1_len = hlc_lease_payload_len
tso_snapshot_v2_len = hlc_lease_payload_len * 2
tso_snapshot_v3_len = tso_snapshot_v2_len + 1
tso_snapshot_v4_len = tso_snapshot_v3_len + 1 + hlc_lease_payload_len
print('computed tsoSnapshotV4Len=', tso_snapshot_v4_len)
print('computed hlcLeasePayloadLen=', hlc_lease_payload_len)
payloads = {
'truncated_unknown_25': bytearray(25),
'invalid_ekvthlc_25': b'EKVTHLC' + bytes(18),
'invalid_kvfsm_magics_25': b'EKVPBBL1' + bytes(17),
'valid_exact_v4': bytearray(tso_snapshot_v4_len),
'short_valid_v4': bytearray(tso_snapshot_v4_len - 1),
}
for name, payload in payloads.items():
peeked_len = min(len(payload), tso_snapshot_v4_len + 1)
result = peeked_len > tso_snapshot_v4_len
print(name, len(payload), 'headerless_legacy_predicate=', result)
PY
printf '\nWriteTo implementation and constants around kvFSM snapshot write:\n'
sed -n '35,90p' kv/fsm.go
sed -n '540,630p' kv/tso_fsm.go
sed -n '690,735p' kv/fsm.goRepository: bootjp/elastickv
Length of output: 8533
ヘッダなし旧 kvFSM スナップショットは魔法値で判定してください。
len(peeked) > tsoSnapshotV4Len（>26バイト）だけで旧形式を判定すると、破損または不正な長尺 TSO スナップショットを restoreHeaderlessLegacyKVFSMSnapshot が全量 discard します。このままだと readTSOSnapshotState の長さ検証を通らず、ceiling と allocation floor が 0 のまま復元成功になります。判定条件は restoreLegacyKVFSMSnapshot と同じく EKVMVCC2、EKVPBBL1、EKVSSTI1 のヘッダに限定するか、長さ上限と魔法値を併用してください。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@kv/tso_fsm.go` around lines 433 - 439, Update
hasHeaderlessLegacyKVFSMSnapshotPayload to identify headerless legacy snapshots
using the recognized magic headers EKVMVCC2, EKVPBBL1, or EKVSSTI1, matching
restoreLegacyKVFSMSnapshot, rather than relying only on len(peeked) >
tsoSnapshotV4Len. Preserve the existing peek error handling and return false for
oversized payloads without a recognized legacy header.
| func (v *appliedReadDispatchVoucher) prepare(coord Coordinator, timestamp uint64) (ReadTimestamp, func(), error) { | ||
| v.mu.Lock() | ||
| defer v.mu.Unlock() | ||
| voucher, ok := coord.(AppliedReadTimestampVoucher) | ||
| if !ok { | ||
| return ReadTimestamp{}, nil, errors.WithStack(ErrTSOProtocolUnsupported) | ||
| } | ||
| preparedVoucher := newAppliedReadDispatchVoucher() | ||
| if err := voucher.VouchAppliedReadTimestamp(timestamp, preparedVoucher.ref); err != nil { | ||
| return ReadTimestamp{}, nil, errors.WithStack(err) | ||
| } | ||
| v.prepared++ | ||
| revoke := func() {} | ||
| if revoker, ok := coord.(AppliedReadTimestampVoucherRevoker); ok { | ||
| ref := preparedVoucher.ref | ||
| revoke = func() { | ||
| revoker.RevokeAppliedReadTimestamp(timestamp, ref) | ||
| } | ||
| } | ||
| return ReadTimestamp{timestamp: timestamp, voucher: preparedVoucher}, revoke, nil | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
voucher の使用回数契約が未確定です。 appliedReadDispatchVoucher.prepare は prepared を加算しますが、その値を読む箇所も上限判定もありません。ErrTSOReadVoucherLimit も未使用です。そのため 1 つの ReadTimestamp から dispatch を何度でも実行できます。アダプタ側はこの緩い契約に依存しています。
kv/tso.go#L179-L199: 上限を実装するか、preparedとErrTSOReadVoucherLimitを削除して「無制限に再利用可能」という契約を doc コメントへ明記してください。adapter/sqs_reaper.go#L76-L93: 1 つのReadTimestampを 1 パス内の多数の削除 dispatch で再利用しています。上限を実装する場合は、リーパーが上限に達しない設計（レコード単位でReadTimestampを取得するなど）へ変更してください。
📍 Affects 2 files
kv/tso.go#L179-L199(this comment)adapter/sqs_reaper.go#L76-L93
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@kv/tso.go` around lines 179 - 199, Define and enforce a clear voucher-use
contract: in kv/tso.go lines 179-199, either implement the prepared-count limit
using ErrTSOReadVoucherLimit or remove prepared and the unused error and
document unlimited ReadTimestamp reuse; if enforcing a limit, update
adapter/sqs_reaper.go lines 76-93 so each deletion dispatch obtains a
ReadTimestamp without exceeding that limit, such as acquiring one per record.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: c231fdbb50
ℹ️ 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 := validateTSOMinimumWindow(floor, n); err != nil { | ||
| return 0, 0, err | ||
| } | ||
| base := floor + 1 |
There was a problem hiding this comment.
When Phase D is active, advancing directly from floor ignores both the current wall clock and the HLC's newer physical ceiling. After an idle interval, the next timestamp therefore retains the previous allocation's physical millisecond, and that component advances by only one millisecond per 65,536 issued timestamps. This breaks consumers that interpret the HLC physical bits as time: for example, S3 stores the commit timestamp in LastModifiedHLC and adapter/s3.go:2473-2481 converts those bits directly into HTTP dates, so post-idle PUTs can report arbitrarily stale Last-Modified values. Preserve the contiguous validation property without freezing the physical component, such as by tracking committed allocation ranges instead of treating every integer gap as issued.
AGENTS.md reference: AGENTS.md:L34-L34
Useful? React with 👍 / 👎.
## Summary - add atomic runtime reload for the one-way `legacy -> shadow -> cutover -> phase-d` sequence - preserve durable cutover and Phase-D overrides while rejecting operator-requested rollback and phase skipping - export bounded TSO latency, shadow divergence, mode, reload, and durable-state metrics with checked Prometheus alerts - add a 16-writer, 1/3/8-way fanout benchmark and record three-run batch-size evidence - promote the centralized TSO design from partial to implemented and update all references ## Safety Legacy mode keeps the coordinator allocator logically absent, preserving existing HLC fallback callers. Cutover activation is published before batch invalidation, durable group-0 markers can only advance the local mode, and invalid reloads leave the active allocator unchanged. ## Validation - `go test ./kv ./monitoring . -count=1` - `go test -race ./kv ./monitoring . -count=1` - `golangci-lint run ./... --timeout=5m --allow-parallel-runners` - `go test ./kv -run ^'$' -bench ^'BenchmarkTSOWriteFanout$' -benchmem -benchtime=1s -count=3`\n- `promtool check config monitoring/prometheus/prometheus.yml` (8 rules)\n- `go test ./store -count=1 -timeout=15m`\n- `git diff --check`\n- caller audit across allocator providers, coordinator fields, runtime flags, activation, and Phase-D state\n- zero references to the former partial design path
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 (96c7573):
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. |
1 participant
Summary
Validation
Notes
Author: bootjp
Summary by CodeRabbit