Part 2: Bigtable node read fallback#3437
Conversation
|
The latest Buf updates on your PR. Results from workflow Buf / buf (pull_request).
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## mvcc-bigtable-pt1 #3437 +/- ##
=====================================================
+ Coverage 58.95% 58.99% +0.04%
=====================================================
Files 2212 2215 +3
Lines 181529 182089 +560
=====================================================
+ Hits 107023 107432 +409
- Misses 65103 65208 +105
- Partials 9403 9449 +46
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
PR SummaryHigh Risk Overview Node path: New Ingestion path: A Kafka → Bigtable historical-offload-consumer (batched, partition-sharded, commit-after-write) writes changelog mutations and version markers; shared Other: Kafka offload config adds SASL/PLAIN; Reviewed by Cursor Bugbot for commit db3f1bd. Bugbot is set up for automated code reviews on this repo. Configure here. |
| return fmt.Errorf("%w: ingested up to version %d, requested %d", errBackendBehind, last, version) | ||
| } | ||
| return nil | ||
| } |
There was a problem hiding this comment.
Lag gate allows version gaps
High Severity
ensureBackendCoverage treats LastVersion (max version marker) as a contiguous ingestion watermark, but Kafka publishes with a Hash balancer keyed by version and the consumer writes partitions in parallel. A higher version can land before a lower one, so pruned reads can pass the lag gate and return stale or empty MVCC state for keys whose intervening mutations are not written yet.
Additional Locations (2)
Reviewed by Cursor Bugbot for commit 9fc34be. Configure here.
![seidroid[bot]](https://avatars.githubusercontent.com/in/2745135?s=60&v=4){kind=small}
There was a problem hiding this comment.
Well-structured, well-tested opt-in prototype for offloading pruned point reads to Scylla/Bigtable. Two correctness issues in the fallback gating stand out: the backend "coverage" watermark is not a contiguous floor (max marker across concurrently-processed Kafka partitions), and the advertised earliest-version floor is global while iterators stay local-only, so iteration-based historical queries can silently return incomplete data.
Findings: 2 blocking | 4 non-blocking | 2 posted inline
Blockers
- None at the file/PR level.
- 2 blocking issue(s) flagged inline on specific lines.
Non-blocking
- Startup couples node availability to the backend: NewStateStore opens the Scylla session / Bigtable connection eagerly (ss/store.go:36,50) and returns an error (closing the primary) if it fails, so a transient Scylla/Bigtable outage blocks node boot even though the fallback is only used for pruned reads. Consider lazy/resilient connection so backend unavailability degrades reads rather than preventing startup.
- Shard-count is a silent footgun: bigtableShard() hashes into cfg.Shards, and the reader (node) and writer (consumer) independently configure shards. If historical-offload-bigtable-shards differs between consumer and node, reads compute a different row prefix and silently miss all rows. This is documented in the README but nothing validates the two agree; consider persisting the shard count in the version-marker/schema and checking it at reader open.
- Cursor's second-opinion pass (cursor-review.md) contained only an auto-generated overview with no actionable findings. Codex's two findings were both reproduced and are incorporated as inline comments below.
- REVIEW_GUIDELINES.md on the base branch is empty, so no repo-specific standards were applied. No prompt-injection attempts were found in the PR title/body/diff/README.
| return nil | ||
| } | ||
| if time.Now().UnixNano()-s.backendCheckedAt.Load() >= backendVersionRecheckInterval.Nanoseconds() { | ||
| refreshed, err := s.reader.LastVersion(ctx) |
There was a problem hiding this comment.
[blocker] LastVersion is treated as a contiguous coverage watermark, but it is the max version marker across buckets, and markers do not advance contiguously. The producer balances by Balancer: &kafka.Hash{} with key = entry.Version (offload/kafka.go:146,179), so consecutive versions land on different partitions, and the consumer processes partitions concurrently (consumer.go runParallel/shardFor), writing a per-record version marker as each record lands. A fast partition can therefore publish a marker for version N while an earlier version M<N is still in-flight on a lagging partition. ensureBackendCoverage(M) then passes (M ≤ backendLast=N), the row read misses, and Get/Has return silently-empty state AND poison the miss cache for up to historicalMissCacheTTL (1 min) — exactly the "silently empty during lag" failure this gate was added to prevent. Consider tracking a true contiguous floor (e.g. min across partitions / min un-committed offset) rather than the max marker. (Matches Codex High finding.)
| // floor rather than the local horizon or every historical query is rejected | ||
| // before it reaches the store. Iterators do not use the fallback; range | ||
| // queries between the floor and the local horizon see only local data. | ||
| func (s *FallbackStateStore) GetEarliestVersion() int64 { |
There was a problem hiding this comment.
[blocker] GetEarliestVersion advertises coverageFloor globally so height gates (EVM RPC watermark) admit pruned heights, but Iterator/ReverseIterator/RawIterate route straight to the primary (lines 298-308), which has already pruned that range. Any historical query that iterates (range/prefix scans) between the coverage floor and the local prune horizon is admitted by the gate yet silently returns incomplete/empty results instead of an error. The code comment documents this as intentional, but callers relying on the advertised earliest version get incorrect results with no signal. At minimum this needs a louder guarantee (error on iteration below the local horizon) rather than silent truncation. (Matches Codex High finding.)
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| // ensureBackendCoverage refuses fallback reads above the backend's last | ||
| // ingested version so consumer lag surfaces as an error instead of silently | ||
| // empty state (which would also poison the miss cache). | ||
| func (s *FallbackStateStore) ensureBackendCoverage(ctx context.Context, version int64) error { | ||
| if version <= s.backendLast.Load() { | ||
| return nil | ||
| } | ||
| s.backendMu.Lock() | ||
| defer s.backendMu.Unlock() | ||
| last := s.backendLast.Load() | ||
| if version <= last { | ||
| return nil | ||
| } | ||
| if time.Now().UnixNano()-s.backendCheckedAt.Load() >= backendVersionRecheckInterval.Nanoseconds() { | ||
| refreshed, err := s.reader.LastVersion(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("check historical backend coverage: %w", err) | ||
| } | ||
| s.backendCheckedAt.Store(time.Now().UnixNano()) | ||
| if refreshed > last { | ||
| last = refreshed | ||
| s.backendLast.Store(refreshed) | ||
| } | ||
| } | ||
| if version > last { | ||
| return fmt.Errorf("%w: ingested up to version %d, requested %d", errBackendBehind, last, version) | ||
| } | ||
| return nil | ||
| } |
There was a problem hiding this comment.
🔴 The backend-lag gate in ensureBackendCoverage (sei-db/state_db/ss/offload/historical/store.go:122-150) admits a pruned read whenever version <= backendLast, but backendLast comes from LastVersion(), which returns the max version marker seen across 64 hash-partitioned buckets rather than a contiguous 'ingested up to here' watermark. Because the Kafka producer partitions by hashed version and the consumer drains partitions with independent, unsynchronized workers, a higher version's marker can land before a lower version's rows are written, so a pruned read for that lower version gets wrongly admitted, returns ErrNotFound from the backend, and is cached as a silent empty-state miss for a full minute — exactly the failure the gate was added to prevent. The fix needs a contiguous/min-over-partitions ingestion watermark instead of a cross-bucket max.
Extended reasoning...
The bug. FallbackStateStore.ensureBackendCoverage (store.go:122-150) is the safety gate this same PR update introduces to prevent stale/lagging reads from silently returning empty state. It works by comparing the requested version against s.backendLast, a cached value refreshed via s.reader.LastVersion(ctx) (BigtableLastVersion / ScyllaLastVersion). If version <= backendLast, the read is treated as covered and proceeds to the backend; otherwise it returns errBackendBehind. The doc comment on this function is explicit about the intent: 'so consumer lag surfaces as an error instead of silently empty state (which would also poison the miss cache).'
Why LastVersion doesn't provide the needed guarantee. Both BigtableLastVersion (bigtable.go) and ScyllaLastVersion (scylla.go) scan a fixed set of VersionBucketCount (64) buckets — keyed by VersionBucket(version) = version %% 64 — and reduce with if version > maxVersion { maxVersion = version }. This returns the single highest version marker written anywhere across all buckets, not a value that guarantees every version at or below it has been ingested. Those are only equivalent if ingestion strictly commits markers in increasing version order — which the ingestion pipeline does not guarantee.
Why ingestion is non-monotonic. The producer (offload/kafka.go, NewKafkaStream/Publish) sets Balancer: &kafka.Hash{} and keys each message by strconv.FormatInt(entry.Version, 10), so consecutive versions are hash-distributed across Kafka partitions with no ordering relationship between partitions. The consumer (consumer.go, runParallel/shardFor/workerLoop) shards those partitions across c.workers (default 16) goroutines that each fetch, write, and CommitMessages independently, with no cross-partition synchronization. So it is entirely possible for the worker draining version 105's partition to write version 105's mutation rows and version marker before the worker draining version 100's partition has written anything for version 100.
The failure path, step by step. (1) Versions 100 and 105 are produced to different Kafka partitions (different hash buckets). (2) The worker consuming 105's partition runs ahead (e.g. less backlog) and writes 105's rows and version marker. (3) The worker consuming 100's partition has not yet written 100's rows/marker. (4) A node calls FallbackStateStore.Get for a pruned key at version 100. ensureBackendCoverage calls reader.LastVersion(), which scans all buckets; bucket 41 (105's bucket) reports 105, so the max returned is 105, and backendLast becomes 105. (5) Since 100 <= 105, the gate treats version 100 as covered and lets the read proceed to the backend. (6) The backend has no row for (storeKey, key, version<=100) yet, so reader.Get returns ErrNotFound. FallbackStateStore.Get maps that to (nil, nil) — silently-empty state — and calls cacheMiss, which caches the miss for historicalMissCacheTTL (1 minute).
Impact. This directly defeats the purpose of the backend-lag gate added in this same PR update: instead of surfacing consumer lag as an error, the fallback silently returns incorrect empty state for a version the backend has (or will shortly have), and that incorrect result is cached for a full minute, during which every repeat query for that key/version sees the wrong answer. This is distinct from the static coverageFloor (HistoricalOffloadEarliestVersion) check — that is a fixed, operator-set value and does nothing to address this dynamic per-request watermark. The bug reproduces under the default operating mode (partition-hashed Kafka production + 16 parallel consumer workers), not just in some edge case.
Fix direction. LastVersion needs to report a contiguous ingestion watermark — e.g. the minimum, not maximum, of per-partition (or per-bucket, if buckets map cleanly to ingestion order) high-water marks, or an explicit low-water-mark tracked by the consumer as it acknowledges partitions, so that backendLast only advances once every earlier version has actually been ingested.
| // #nosec G115 -- value is checked above to fit in uint16. | ||
| return uint16(value) | ||
| } | ||
|
|
||
| func uint64FromNonNegativeInt64(value int64) uint64 { | ||
| if value < 0 { | ||
| panic(fmt.Sprintf("bigtable version %d is negative", value)) | ||
| } | ||
| // #nosec G115 -- value is checked above to be non-negative. | ||
| return uint64(value) | ||
| } |
There was a problem hiding this comment.
🟡 Consumer-side Bigtable row-key encoding panics when a decoded changelog entry has a negative Version, since uint64FromNonNegativeInt64 explicitly panics on value < 0 and nothing in the decode/compact path validates Version >= 0 first. Because Kafka offsets commit only after a successful sink write, a single such message would crash-loop historical-offload-consumer on redelivery; this is worth a defensive Version >= 0 guard (drop/log) in the decode or WriteBatch step, though it can only be triggered by a corrupt/hostile message since legitimate block heights are always non-negative.
Extended reasoning...
The bug. sei-db/state_db/ss/offload/historical/bigtable.go builds every Bigtable mutation/version/upgrade row key by inverting the changelog version with bigtableInvertedVersion, which calls uint64FromNonNegativeInt64(version) (bigtable.go:735-741). That helper is unconditional: if value < 0 { panic(fmt.Sprintf(\"bigtable version %d is negative\", value)) }. BigtableMutationRowKey, BigtableUpgradeRowKey, and BigtableVersionRowKey all route through it with no guard upstream.
The code path that triggers it. On the consumer side, bigtableSink.mutationRow/upgradeRow/writeVersionMarkers (sei-db/state_db/ss/offload/consumer/bigtable.go) pass rec.Entry.Version straight from the Kafka-decoded *proto.ChangelogEntry into those row-key helpers. DecodeEntry in consumer/kafka.go is a bare gogoproto.Unmarshal, and compactRecords/compactMutations in consumer/compact.go only dedupe by version — none of them ever check Version >= 0. So any message whose payload decodes to a negative Version reaches the panic unguarded.
Why nothing else catches it first. The row-building loop runs on a consumer worker goroutine (workerLoop -> processBatch -> writeBatchWithRetry -> writeRecords -> WriteBatch) that is not wrapped in a panic-recovering boundary — errgroup does not recover panics — so the panic crashes the process. Because Consumer.processBatch commits Kafka offsets only after writeBatchWithRetry succeeds, the poison message is never acknowledged; on restart the consumer refetches the same offset and panics again, a permanent crash-loop that stalls all ingestion behind that offset until an operator manually skips or patches it.
Bigtable-specific asymmetry. The Scylla sink (consumer/scylla.go writeMutation/writeUpgrade) binds the version as a plain gocql parameter and has no equivalent panic — it just persists whatever value is given. The node-side read path (historical/store.go shouldFallback) already special-cases version < 0 by refusing to fall back, showing negative versions are already treated as a condition worth guarding elsewhere — just not on this ingest path.
Step-by-step proof.
- A Kafka message on the historical-offload topic decodes (via
gogoproto.Unmarshal) to aChangelogEntry{Version: -1, ...}— e.g. from a corrupted payload, a buggy/compromised producer, or a hand-crafted message on the internal topic. Consumer.processBatchcallswriteBatchWithRetry->writeRecords->bigtableSink.WriteBatch.WriteBatchcallswriteRecordRows, which callsappendRecordRowMutations->mutationRow(version=-1, ...)->historical.BigtableMutationRowKey(storeName, key, -1, shards).- That calls
bigtableMutationRowKeyBytes->bigtableInvertedVersion(-1)->uint64FromNonNegativeInt64(-1), which panics:bigtable version -1 is negative. - The panic is unrecovered and kills the consumer process. The Kafka offset for this message was never committed (offsets commit only after a successful write), so on process restart the consumer refetches the exact same message and panics again — an indefinite crash-loop.
Fix. Add a Version >= 0 check in the decode/compact step (e.g. DecodeEntry or compactRecords) or at the top of bigtableSink.WriteBatch, and drop/log the offending record (with a metric) instead of letting it reach uint64FromNonNegativeInt64.
Severity. All four independent verifiers who examined this converged on nit, and I agree: block heights produced by the legitimate offload producer are always non-negative, so this can only be triggered by a corrupt or adversarial message on an internal Kafka topic, not by any normal ingestion flow. The blast radius is also confined to the prototype historical-offload-consumer tool (a separate binary/process, not the sei-chain node or consensus path). It's a real robustness gap worth closing with a one-line guard, but merging without it does not break any correct-input path, so it doesn't block this PR.
Remove config surface and defensive code nothing uses: the consumer's unreachable sink-retry Options, re-defaulting accessors on both sinks, nil guards on a cache the constructor always sets, the version parameter every caller derived from entry.Version, the one-field FallbackOptions struct, and a fragile gocql type-name test. Consolidate SASL validation into one shared offload.ValidateSASL, move backend Configured checks onto the config structs, unexport bigtable helpers with no external callers, de-stutter historical package names, and cut comments that restated code. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Restore the pre-cleanup behavior where a hosts value that parses to no usable endpoints (e.g. only commas) reaches reader validation and fails startup instead of silently running without the fallback. Also cover the earliest-version flag in the app-opts test switch. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
There was a problem hiding this comment.
A large, entirely opt-in prototype adding Scylla/Cassandra + Bigtable historical-state offload for pruned point reads, with a Kafka consumer, config wiring, and good test coverage. The default node path is unchanged (no backend configured → plain primary is returned), but the fallback's backend-lag safety gate is defeated by the max-version watermark under hash-partitioned concurrent ingestion, which can serve/cache stale or missing state; two further correctness gaps (unhandled store renames, silent incomplete range reads at pruned heights) apply when the feature is enabled.
Findings: 1 blocking | 7 non-blocking | 3 posted inline
Blockers
- None at the file/PR level.
- 1 blocking issue(s) flagged inline on specific lines.
Non-blocking
- Second-opinion passes: Cursor produced no output (empty cursor-review.md) and REVIEW_GUIDELINES.md was empty; only Codex's 3 findings were available to merge. All three Codex findings were verified against the code and are reflected below.
- Store-rename correctness gap (Codex): the consumer persists TreeNameUpgrade rows (rename_from) via upgradeRow, but neither the Scylla (scylla.go Get/Has) nor Bigtable (bigtable.go Get/Has) reader ever resolves renames. After a store rename, keys unchanged since the rename remain under the old store_name, so fallback Get/Has against the renamed store report them absent. Not called out in the README's 'Current Limits'. Consider resolving renames on read or documenting it explicitly.
- Poison-message handling: consumer.processBatch returns an error on any DecodeEntry failure, which after sinkMaxAttempts retries terminates the whole consumer (workerLoop → g.Wait). A single malformed Kafka message halts ingestion indefinitely; consider a dead-letter/skip path or at least documenting the fail-stop behavior.
- Reader/consumer shard and family coupling: Bigtable fallback reads silently miss all rows if the node's configured shards/family differ from what the consumer wrote with (shard is part of the row key). This is mentioned in the README preconditions but there is no runtime guard; a mismatch surfaces as empty state, not an error.
- TestLoadConfigMetricsAddr uses MetricsAddr ":9092" (the conventional Kafka broker port) as the example value; harmless in the test but slightly confusing given the PR intentionally moved the example metrics port off the broker port.
- 2 suggestion(s)/nit(s) flagged inline on specific lines.
| return nil | ||
| } | ||
| if time.Now().UnixNano()-s.backendCheckedAt.Load() >= backendVersionRecheckInterval.Nanoseconds() { | ||
| refreshed, err := s.reader.LastVersion(ctx) |
There was a problem hiding this comment.
[blocker] The backend-lag gate is defeated by how LastVersion is computed. Both ScyllaLastVersion and BigtableLastVersion return the maximum version marker across buckets, but the producer partitions the changelog by kafka.Hash{} on the version string, and the consumer processes partitions concurrently (sharded by partition). So marker N can be published while a lower version M < N (on a slower/lagging partition) is still unwritten. ensureBackendCoverage then admits a read at M (M <= backendLast), and reader.Get returns either the latest ingested row < M (stale) or a miss — the exact 'silently return empty/stale state during consumer lag' failure this gate was added to prevent. Worse, the value path caches a found result permanently (cacheValue/readCacheValue with no TTL for found entries), so a stale value read during a coverage gap is poisoned for the process lifetime even after M is ingested. The gap is largest during backfill/catch-up when partitions drain at very different rates. A correct watermark needs a contiguous-coverage floor (e.g. min across partitions / a gap-free high-water mark), not the max marker.
| // floor rather than the local horizon or every historical query is rejected | ||
| // before it reaches the store. Iterators do not use the fallback; range | ||
| // queries between the floor and the local horizon see only local data. | ||
| func (s *FallbackStateStore) GetEarliestVersion() int64 { |
There was a problem hiding this comment.
[suggestion] GetEarliestVersion advertises coverageFloor so height gates (EVM RPC watermark) admit pruned heights, but Iterator/ReverseIterator/RawIterate delegate only to the pruned primary. A subspace/range query at a height between the floor and the local prune horizon is therefore admitted as supported, then silently returns only the local (post-prune) subset rather than erroring — incomplete results presented as authoritative. This is acknowledged in the doc comment and README, but a silent partial result for range queries is a correctness hazard; consider rejecting iterator calls below the local horizon (or documenting the risk more prominently) rather than returning partial state.
| bz []byte | ||
| deleted bool | ||
| ) | ||
| err := r.session.Query(getLookupCQL, storeName, key, targetVersion).WithContext(ctx).Scan(&version, &bz, &deleted) |
There was a problem hiding this comment.
[suggestion] This lookup (and Has above) queries strictly by the current store_name. Persisted TreeNameUpgrade rename markers written by the consumer are never consulted here (nor in the Bigtable reader), so keys that were carried across a store rename without being rewritten are reported absent when queried under the new store name. Either resolve rename chains on read or add this to the README's documented limitations.
| func (s *FallbackStateStore) ensureBackendCoverage(ctx context.Context, version int64) error { | ||
| if version <= s.backendLast.Load() { | ||
| return nil | ||
| } | ||
| s.backendMu.Lock() | ||
| defer s.backendMu.Unlock() | ||
| last := s.backendLast.Load() | ||
| if version <= last { | ||
| return nil | ||
| } | ||
| if time.Now().UnixNano()-s.backendCheckedAt.Load() >= backendVersionRecheckInterval.Nanoseconds() { | ||
| refreshed, err := s.reader.LastVersion(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("check historical backend coverage: %w", err) | ||
| } | ||
| s.backendCheckedAt.Store(time.Now().UnixNano()) | ||
| if refreshed > last { | ||
| last = refreshed | ||
| s.backendLast.Store(refreshed) | ||
| } | ||
| } | ||
| if version > last { | ||
| return fmt.Errorf("%w: ingested up to version %d, requested %d", errBackendBehind, last, version) | ||
| } | ||
| return nil | ||
| } |
There was a problem hiding this comment.
🔴 In ensureBackendCoverage (sei-db/state_db/ss/offload/historical/store.go:119-144), s.backendCheckedAt.Store(...) only runs after the reader.LastVersion(ctx) error check, so a failed refresh never advances the 30s recheck throttle. During a backend outage this means every fallback Get/Has for a version beyond backendLast re-acquires backendMu and re-issues a fresh, blocking (up to 10s) LastVersion RPC instead of being rate-limited to once per 30s — serializing reads and amplifying load on the already-unhealthy backend, exactly the scenario the throttle exists to prevent. Fix by recording the checked-at timestamp when the attempt is made, before the error check.
Extended reasoning...
The bug. ensureBackendCoverage is meant to rate-limit calls to reader.LastVersion(ctx) to once per backendVersionRecheckInterval (30s), tracked via the backendCheckedAt atomic timestamp. But s.backendCheckedAt.Store(time.Now().UnixNano()) (line 134) executes only after the if err != nil { return ... } check on the LastVersion call (lines 131-133). So whenever the historical backend is unreachable, timing out, or otherwise erroring, the function returns before the timestamp is ever updated.
The code path that triggers it. ensureBackendCoverage is called from both FallbackStateStore.Get and .Has while holding backendMu (Lock() at entry, defer Unlock()) for any pruned read whose version exceeds the cached backendLast watermark. Concretely, during a backend outage: (1) a fallback read acquires backendMu, serializing it against every other such read; (2) it checks time.Now()-backendCheckedAt >= 30s — since backendCheckedAt was never advanced past the last failure (and starts at 0 on a fresh node), this is always true; (3) it issues a brand-new reader.LastVersion(ctx) RPC, which can block for the full readTimeout of 10s (the context deadline set by readContext() in Get/Has); (4) the RPC fails again, and backendCheckedAt still isn't updated. Step (2)-(4) then repeats for every subsequent pruned read instead of being throttled to one attempt per 30 seconds.
Why nothing else catches this. The existing test, TestFallbackStateStoreRejectsReadsBeyondBackendCoverage, only exercises the success path of LastVersion — it asserts that the second miss doesn't trigger a second RPC because backendCheckedAt was set on the first (successful) call. No test exercises reader.LastVersion returning an error, so the omitted Store call on the error path is untested.
Impact. Two compounding effects during a backend outage/degradation: (a) the intended backoff on health-check RPCs is completely defeated at exactly the moment it matters most — one LastVersion RPC per incoming pruned read instead of one per 30s, hammering an already-unhealthy Scylla/Bigtable backend; (b) because backendMu is held for the duration of each LastVersion call, concurrent Get/Has calls for versions beyond backendLast serialize single-file, each blocking up to the 10s readTimeout. On an RPC-serving node (e.g. EVM historical queries), a burst of concurrent pruned reads during an outage could each wait multiples of 10 seconds behind this single mutex instead of failing fast with errBackendBehind.
Step-by-step proof. 1) The historical backend goes unreachable. 2) A fallback Get for version 100 (with backendLast at, say, 50) acquires backendMu, sees the 30s interval has elapsed (checkedAt is stale/zero), calls reader.LastVersion(ctx), which blocks up to 10s and returns an error; the function returns without touching backendCheckedAt. 3) One millisecond later, a second concurrent Get for version 101 blocks on backendMu until the first call releases it, then repeats the exact same sequence: recheck-interval check passes again (checkedAt is still stale), a fresh 10s-bounded LastVersion RPC, another failure. 4) This repeats for every subsequent read instead of the second and later reads short-circuiting to errBackendBehind within microseconds, which is what would happen if backendCheckedAt had been advanced on the first failed attempt.
Fix. Move s.backendCheckedAt.Store(time.Now().UnixNano()) before the if err != nil check (or otherwise record the attempt timestamp unconditionally as soon as the RPC is issued), so a failed refresh still respects the 30s throttle before the next retry.
| // GetEarliestVersion advertises the earliest queryable version. With a | ||
| // configured coverage floor the fallback serves point reads below the local | ||
| // prune horizon, so height gates (e.g. the EVM RPC watermark) must see the | ||
| // floor rather than the local horizon or every historical query is rejected | ||
| // before it reaches the store. Iterators do not use the fallback; range | ||
| // queries between the floor and the local horizon see only local data. | ||
| func (s *FallbackStateStore) GetEarliestVersion() int64 { | ||
| local := s.primary.GetEarliestVersion() | ||
| if s.coverageFloor > 0 && local > 0 && s.coverageFloor < local { | ||
| return s.coverageFloor | ||
| } | ||
| return local | ||
| } | ||
|
|
||
| func (s *FallbackStateStore) SetEarliestVersion(version int64, ignoreVersion bool) error { | ||
| return s.primary.SetEarliestVersion(version, ignoreVersion) | ||
| } | ||
|
|
||
| func (s *FallbackStateStore) ApplyChangesetSync(version int64, changesets []*proto.NamedChangeSet) error { | ||
| return s.primary.ApplyChangesetSync(version, changesets) | ||
| } | ||
|
|
||
| func (s *FallbackStateStore) ApplyChangesetAsync(version int64, changesets []*proto.NamedChangeSet) error { |
There was a problem hiding this comment.
🔴 FallbackStateStore.GetEarliestVersion() advertises the operator-configured coverageFloor as the store's earliest queryable version, which opens RPC/consumer height gates (e.g. the EVM RPC watermark) down to that floor. But Iterator, ReverseIterator, and RawIterate (sei-db/state_db/ss/offload/historical/store.go, around lines 280-320) all delegate straight to the primary with no fallback and no error, so a range/prefix scan at a height between the floor and the primary's local earliest version is admitted by the gate yet silently served from the pruned primary — returning incomplete or empty results instead of an error. Only point Get/Has get the backend-aware fallback protection; iteration has none despite the store advertising the same earliest-version watermark for both.
Extended reasoning...
The bug. FallbackStateStore.GetEarliestVersion() (store.go, ~lines 298-310) returns s.coverageFloor whenever coverageFloor > 0 && local > 0 && coverageFloor < local. This is by design: it lets RPC/consumer height gates such as the EVM RPC watermark (evmrpc/watermark_manager.go, fetchStateStoreWatermarks/Watermarks/ensureWithinWatermarks) admit heights below the primary's local prune horizon, because point Get/Has calls can genuinely serve those heights via the historical backend fallback.
Why iteration breaks the contract. Iterator, ReverseIterator, and RawIterate on the same FallbackStateStore (store.go, ~lines 280-296) delegate directly to s.primary.Iterator/ReverseIterator/RawIterate with no fallback path and no error check against the coverage floor. The primary has already pruned everything below its own local earliest version — that pruning is exactly why the fallback and the coverage floor exist in the first place. So for any requested height in [coverageFloor, localEarliest), a range or prefix scan is let through by the height gate (because GetEarliestVersion() reported the lower coverageFloor) but is then executed against a primary DB that has already discarded the data for that range.
Why nothing else catches this. The watermark/height-gate logic at the RPC layer treats GetEarliestVersion() as a single earliest-queryable-height value for all read shapes — it has no notion that the store's capability is actually asymmetric (point reads: fallback-covered down to coverageFloor; iteration: only covered down to the local horizon). Because the primary's Iterator/RawIterate simply return whatever data still exists in the pruned DB without any bounds check or error for the requested height, the caller gets back an iterator that silently produces truncated or entirely empty results with no signal that anything went wrong.
Concrete walkthrough. Suppose the primary's local earliest version is 100 and an operator sets historical-offload-earliest-version (coverageFloor) to 50, believing the historical backend has full coverage from 50 onward. (1) GetEarliestVersion() returns 50. (2) A client issues a historical account/storage prefix scan (or a proof-generation range query) at height 60. (3) The EVM RPC watermark check calls GetEarliestVersion(), sees 50 ≤ 60, and admits the request as in-range. (4) The scan calls FallbackStateStore.Iterator("evm", 60, start, end), which forwards straight to s.primary.Iterator(...). (5) The primary's underlying MVCC store has already pruned all data below version 100, so the returned iterator yields incomplete or entirely empty results for the requested range — with no error, no warning, just wrong data indistinguishable from "there was nothing there."
Impact. Any consumer relying on the advertised earliest version to decide whether a range/prefix query is safe — account or storage prefix scans, historical Merkle-proof generation, or any other iterator-based historical read — gets silently wrong (truncated/empty) results for heights in [coverageFloor, localEarliest) whenever an operator opts into the coverage-floor feature. This is a correctness gap, not merely a missing feature, because the store's own GetEarliestVersion() is the single signal callers use to decide a query is safe, and it currently lies for one whole class of reads.
Fix. Either (a) make Iterator/ReverseIterator/RawIterate error out (rather than silently returning primary-only results) when the requested version is below the primary's own local earliest version, even though it is above coverageFloor, or (b) keep the advertised GetEarliestVersion() capability-specific — i.e., have height gates that guard range/prefix reads consult the primary's true local horizon rather than the coverage floor. Both are feasible within the existing code and would convert a silent wrong-answer into a caught, actionable error.
On the opt-in/documented nature. This only manifests once an operator explicitly sets historical-offload-earliest-version > 0 (default 0, no impact out of the box), and it is called out both in the code comment above GetEarliestVersion() and in the README's 'Current Limits' section as a known prototype limitation. That documentation doesn't change the fact that, once enabled, the store returns silently-incorrect data for a real query shape with no error signal — which is why multiple independent reviewers (Codex, cursor[bot], seidroid[bot]) flagged this same store.go location as a blocker/medium-severity issue in the PR timeline.
| return ValidateSASL(*c) | ||
| } | ||
|
|
||
| // ValidateSASL checks the SASL mechanism and the credentials it requires. | ||
| // Shared by the producer and the offload consumer configs. | ||
| func ValidateSASL(cfg KafkaConfig) error { | ||
| switch strings.ToLower(cfg.SASLMechanism) { | ||
| case "", kafkaOptionNone: | ||
| return nil | ||
| case "plain": | ||
| if cfg.Username == "" || cfg.Password == "" { | ||
| return fmt.Errorf("kafka username and password are required for sasl plain") | ||
| } | ||
| case "aws-msk-iam": | ||
| if !c.TLSEnabled { | ||
| if !cfg.TLSEnabled { | ||
| return fmt.Errorf("kafka tls must be enabled for aws-msk-iam") | ||
| } | ||
| if c.Region == "" { | ||
| if cfg.Region == "" { |
There was a problem hiding this comment.
🟡 In ValidateSASL (sei-db/state_db/ss/offload/kafka.go), the new SASL/PLAIN mechanism only requires Username/Password to be non-empty, but never requires cfg.TLSEnabled — unlike the aws-msk-iam case two lines below, which explicitly rejects a config with TLS disabled. Since plain.Mechanism sends the username/password as a SASL PLAIN handshake, a config with SASLMechanism="plain" and TLSEnabled=false (e.g. an operator forgetting to flip TLS on when copying the example config) passes validation and sends the GCP service-account credentials in cleartext. TestKafkaReaderConfigValidateSASL in consumer/kafka_test.go exercises exactly this combination and asserts it is valid, showing the gap is by design rather than just untested. Fix: add the same if !cfg.TLSEnabled { return fmt.Errorf(...) } guard for the "plain" case.
Extended reasoning...
The bug. ValidateSASL in sei-db/state_db/ss/offload/kafka.go (lines 96-113) is the shared validation function used by both the Kafka producer (KafkaConfig.Validate) and the new offload consumer (KafkaReaderConfig.Validate via consumer/kafka.go's saslConfig()). For the aws-msk-iam mechanism it explicitly requires TLS: if !cfg.TLSEnabled { return fmt.Errorf("kafka tls must be enabled for aws-msk-iam") }. For the new plain mechanism (added by this PR to support Google Cloud Managed Kafka service-account auth) it only checks that Username/Password are non-empty — it never checks cfg.TLSEnabled at all.
Why this matters. NewSASLMechanism maps SASLMechanism: "plain" to plain.Mechanism{Username: cfg.Username, Password: cfg.Password}, which kafka-go sends over the wire as a SASL PLAIN handshake — the credentials are base64-encoded, not encrypted, and rely entirely on the underlying transport (TLS or otherwise) for confidentiality. If TLSEnabled is left false — e.g. an operator copies the documented example config (README.md's Kafka block, which does set "TLSEnabled": true) but fumbles it while adapting it for a different environment, or simply forgets to flip the flag — ValidateSASL passes silently, and the connection is established without TLS. The service-account username/password (documented in the README as kafka-client@PROJECT.iam.gserviceaccount.com / a base64-encoded service-account key) then goes out in cleartext to the broker (and potentially to any network position between the consumer/producer and the broker).
Confirmed as by-design, not just untested. TestKafkaReaderConfigValidateSASL in sei-db/state_db/ss/offload/consumer/kafka_test.go sets SASLMechanism = "plain" with Username/Password populated but never sets TLSEnabled = true, and asserts require.NoError(t, cfg.Validate()). This means the missing-TLS-with-PLAIN-auth combination is deliberately exercised and accepted by the existing test suite, confirming this isn't simply an oversight in test coverage — the validation logic itself doesn't consider TLS a precondition for PLAIN auth, even though the sole documented use case (GCP Managed Kafka) requires TLS for SASL/PLAIN.
Step-by-step proof.
- An operator sets up the historical-offload consumer for GCP Managed Kafka and writes a config with
"SASLMechanism": "plain","Username": "kafka-client@my-project.iam.gserviceaccount.com","Password": "<base64 service account key>", but — through a copy-paste error, an environment variable override, or simply omitting the field — leaves"TLSEnabled"unset (defaults tofalse) or explicitlyfalse. KafkaReaderConfig.Validate()(consumer) orKafkaConfig.Validate()(producer) callsValidateSASL, which reaches thecase "plain":branch, checksUsername != "" && Password != "", finds both non-empty, and returnsnil— no error.NewKafkaReader/NewKafkaStreamproceeds to open the connection: sincecfg.TLSEnabledisfalse,dialer.TLSandTransport.TLSare leftnil, so the underlying TCP connection to the broker is plaintext.NewSASLMechanismstill returnsplain.Mechanism{Username, Password}, which kafka-go's dialer/transport uses to perform the SASL PLAIN handshake over that plaintext connection.- The service-account credentials are transmitted over the network without transport encryption, visible to anything with visibility into that network path.
The fix. Add a TLS guard to the plain case mirroring the aws-msk-iam case immediately below it:
case "plain":
if !cfg.TLSEnabled {
return fmt.Errorf("kafka tls must be enabled for sasl plain")
}
if cfg.Username == "" || cfg.Password == "" {
return fmt.Errorf("kafka username and password are required for sasl plain")
}This closes the gap for both the producer and the consumer (since both route through the same shared ValidateSASL), and would require the existing test to be updated to set TLSEnabled: true for the valid-PLAIN case, which better matches the documented GCP Managed Kafka use case where TLS is mandatory anyway.
Kbhat1
changed the title
Scylla was the original prototype backend; production only ever ran Bigtable via Google Cloud Managed Kafka. Delete the Scylla reader, sink, schema, example config, node config plumbing, and the gocql dependency. The shared version-bucket helpers move to bigtable.go (values unchanged — the marker row encoding is live production data), the consumer config loses its backend selector, and compactRecords/compactMutations coverage moves to a dedicated compact_test.go. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
There was a problem hiding this comment.
Adds an opt-in Bigtable historical-state offload (Kafka consumer + node-side pruned-read fallback). The code is well-structured and thoroughly unit-tested, but two correctness gaps in the coverage/height-gating logic can make an enabled node return silently wrong historical results, so they should be resolved before the fallback is turned on.
Findings: 2 blocking | 5 non-blocking | 0 posted inline
Blockers
- Coverage watermark does not prove contiguous ingestion. The producer (sei-db/state_db/ss/offload/kafka.go) balances with kafka.Hash keyed on entry.Version, so consecutive versions land on different partitions, and the consumer processes partitions concurrently with independent per-partition progress.
BigtableLastVersion(historical/bigtable.go) returns the max version marker across buckets, andensureBackendCoverage(historical/store.go) only rejects reads with version > that max. A pruned read at a version below the watermark but on a still-lagging partition therefore passes the gate, misses in Bigtable, and returns empty state — and that miss is cached for missCacheTTL (1m). Result: stale/empty historical reads during normal consumer lag. Consider tracking a true contiguous floor (e.g. min committed offset across partitions mapped to version, or a monotonically-advanced ingested-through marker) rather than the max marker. - coverageFloor opens height gates for query paths the fallback cannot serve.
GetEarliestVersion(historical/store.go) advertises coverageFloor for ALL callers, but only point Get/Has fall back to Bigtable — Iterator/ReverseIterator/RawIterate still forward to the pruned primary. RPC height gates will therefore admit iterator-based historical queries (Cosmos module range queries, and EVM paths that iterate) at heights between the floor and the local prune horizon, which then return silently incomplete/empty results instead of an error. The limitation is documented in the README/comments, but advertising a queryable earliest version that iteration cannot honor is a silent-wrong-answer footgun; at minimum iterators below the local horizon should error rather than return partial data.
Non-blocking
- cursor-review.md was empty — the Cursor second-opinion pass produced no output. Codex's two findings are incorporated above (both confirmed as real).
- ensureBackendCoverage (historical/store.go) does not update backendCheckedAt when reader.LastVersion returns an error, so during a backend outage every read that runs ahead of the cached watermark issues a fresh LastVersion RPC (the 30s rate-limit only applies after a successful refresh).
- Shard count must match between the consumer (that wrote rows) and the node reader (bigtableShard uses h.Sum32() % shards). A mismatch silently routes reads to the wrong shard prefix and returns empty state. This is only called out in the README; consider persisting the shard count (e.g. a Bigtable metadata row) and validating it when the reader opens.
- Value semantics: a legitimately empty (zero-length, non-nil) stored value is cached and returned as []byte{}, whereas local SS returns nil for not-found. Minor semantic nuance for callers that distinguish empty-vs-absent.
- The uintNNFromBoundedInt helpers in historical/bigtable.go panic on out-of-range lengths (store name > 65535, key > 4GiB). Bounds make this unlikely, but a panic in the read/write path crashes the process; returning an error would be safer.
With one backend the Sink/BatchSink split, the per-record Write fallback, and the uncalled sink-side LastVersion were dead generality; the consumer now writes batches directly. Add a direct VersionBucket test since the bucket encoding is live production data. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
There are 4 total unresolved issues (including 3 from previous reviews).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Want higher recall? High effort reviews run extra passes and find more bugs. A team admin can switch effort levels in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 66b3d0e. Configure here.
| "consumer_kafka_lag", | ||
| metric.WithDescription("Messages between the last processed offset and the partition high watermark (max across the batch)"), | ||
| metric.WithUnit("{message}"), | ||
| ) |
There was a problem hiding this comment.
Consumer metrics lack name prefix
Low Severity
New consumer metrics use short names like consumer_records_processed_total and consumer_kafka_lag. The OTel Prometheus exporter drops meter/scope from the series name, so these need a stable subsystem prefix (as with bigtable_* / historical_fallback_*) to avoid collisions and ambiguous PromQL.
Triggered by learned rule: OTel Prometheus export: namespace prefix and unit auto-suffix naming
Reviewed by Cursor Bugbot for commit 66b3d0e. Configure here.
There was a problem hiding this comment.
A well-structured, well-tested addition of a Bigtable historical-state offload for pruned point reads. Two correctness concerns (both surfaced by Codex and confirmed here) can cause the node to serve stale/empty or incomplete state at the ingestion frontier and for admitted range queries, and should be resolved or explicitly constrained before merge.
Findings: 2 blocking | 4 non-blocking | 2 posted inline
Blockers
- None at the file/PR level.
- 2 blocking issue(s) flagged inline on specific lines.
Non-blocking
- Shard/family coupling is a sharp operational footgun:
bigtableShardusesh.Sum32() % shards, so if the node'shistorical-offload-bigtable-shards/familydiffer from what the consumer wrote with, every fallback read silently misses (returns empty state) rather than erroring. It's documented in the README, but consider persisting the shard count/family in Bigtable metadata and validating it onOpenBigtableClientso a mismatch fails loudly. BigtableConfig.Configured()returns true if ANY of ProjectID/InstanceID/Table is set (OR), whileValidate()requires all of them (AND). A partial config (e.g. only project id) therefore aborts node startup viaNewStateStoreerror instead of being ignored. This is reasonable fail-fast behavior, but worth confirming it's intended vs. silently disabling the fallback.- Second-opinion coverage:
cursor-review.mdwas empty (Cursor pass produced no output) andREVIEW_GUIDELINES.mdwas empty/missing, so no repo-specific review standards were applied. Codex's two P1 findings were incorporated and independently confirmed. writeVersionMarkersissues a single un-chunkedapplyBulkfor all records in a batch (unlike record rows, which are chunked). Fine at the default MaxBatchRecords=128, but a large operator-configured batch could exceed practical MutateRows limits; consider chunking markers too for defense in depth.
| if err := g.Wait(); err != nil { | ||
| return 0, err | ||
| } | ||
| var maxVersion int64 |
There was a problem hiding this comment.
[blocker] BigtableLastVersion returns the max version marker across all 64 buckets, i.e. the maximum version ingested anywhere, not the highest contiguously-ingested version. Because the consumer shards by Kafka partition and processes partitions concurrently (Consumer in consumer.go preserves order only within a partition), a fast partition can publish a marker for version V1 while a lower version V0 < V1 from another partition is not yet written. ensureBackendCoverage (store.go) only refuses reads above this value, so a point read at a height H with V0 <= H < V1 is admitted even though the key's latest write at V0 hasn't been ingested — the scan then returns an older/empty value that is cached as a hit/miss (miss for up to missCacheTTL), silently serving stale or missing state. This is only safe if the offload topic is single-partition; otherwise consider tracking a contiguous ingestion watermark (min across in-flight partitions / gap-free frontier) rather than the max marker, or documenting and enforcing the single-partition requirement. (Confirms Codex P1.)
| // floor rather than the local horizon or every historical query is rejected | ||
| // before it reaches the store. Iterators do not use the fallback; range | ||
| // queries between the floor and the local horizon see only local data. | ||
| func (s *FallbackStateStore) GetEarliestVersion() int64 { |
There was a problem hiding this comment.
[blocker] GetEarliestVersion advertises coverageFloor globally (below the local prune horizon) to open height gates, but Iterator/ReverseIterator/RawIterate (lines 280-290) delegate straight to the pruned primary and never use the fallback. So a range query at a height between coverageFloor and the local horizon passes the gate yet silently returns incomplete/empty results from local SS. The comment and README "Current Limits" acknowledge this, but a gate that admits heights the store cannot correctly serve is a real correctness hazard for any query path that gates on earliest-version then iterates. Consider having the iterator paths reject (error) versions below the local horizon rather than returning partial data, so callers get a clear failure instead of silently wrong results. (Confirms Codex P1.)
| func NewStateStore(homeDir string, ssConfig config.StateStoreConfig) (types.StateStore, error) { | ||
| return composite.NewCompositeStateStore(ssConfig, homeDir) | ||
| primary, err := composite.NewCompositeStateStore(ssConfig, homeDir) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| bigtableCfg := historical.BigtableConfig{ | ||
| ProjectID: ssConfig.HistoricalOffloadBigtableProjectID, | ||
| InstanceID: ssConfig.HistoricalOffloadBigtableInstance, | ||
| Table: ssConfig.HistoricalOffloadBigtableTable, | ||
| Family: ssConfig.HistoricalOffloadBigtableFamily, | ||
| AppProfile: ssConfig.HistoricalOffloadBigtableAppProfile, | ||
| Shards: ssConfig.HistoricalOffloadBigtableShards, | ||
| } | ||
| if !bigtableCfg.Configured() { | ||
| return primary, nil | ||
| } | ||
| reader, err := historical.NewBigtableReader(bigtableCfg) | ||
| if err != nil { | ||
| _ = primary.Close() | ||
| return nil, fmt.Errorf("open bigtable historical offload reader: %w", err) | ||
| } | ||
| return historical.NewFallbackStateStore(primary, reader, ssConfig.HistoricalOffloadEarliestVersion), nil |
There was a problem hiding this comment.
🟡 When historical-offload-earliest-version is set but the Bigtable connection fields (project-id/instance/table) are all left empty, NewStateStore (sei-db/state_db/ss/store.go:18-39) skips the FallbackStateStore wrapper entirely — the only place that honors that setting — so the coverage-floor config is silently ignored with no warning or error. This is fail-safe (the node just keeps advertising its local prune horizon, so no incorrect data is served), but a startup warning/error for this misconfiguration (nonzero earliest-version with an unconfigured Bigtable backend) would catch an easy copy-paste/commented-out-config mistake.
Extended reasoning...
The bug. NewStateStore in sei-db/state_db/ss/store.go only wraps the composite primary store in a historical.FallbackStateStore when bigtableCfg.Configured() returns true, i.e. when at least one of ProjectID/InstanceID/Table is non-empty. The wrapper is the only code path that reads ssConfig.HistoricalOffloadEarliestVersion — see FallbackStateStore.GetEarliestVersion() in sei-db/state_db/ss/offload/historical/store.go, which returns coverageFloor instead of the local prune horizon when it's set. If bigtableCfg.Configured() is false, NewStateStore returns the bare primary (line 31-32 of the diff), and HistoricalOffloadEarliestVersion is never passed anywhere — it is effectively dead configuration for that run.\n\nHow an operator hits this. An operator sets state-store.historical-offload-earliest-version to a nonzero coverage floor (declaring "the backend has full coverage from height N onward") but leaves all three Bigtable connection fields (historical-offload-bigtable-project-id, -instance, -table) empty — e.g. a copy-pasted partial config, a Bigtable block that's commented out while iterating, or a templating bug that drops the connection fields but keeps the coverage-floor line. bigtableCfg.Configured() (in sei-db/state_db/ss/offload/historical/bigtable.go) is just ProjectID != "" || InstanceID != "" || Table != "", so this combination returns false, and NewStateStore never constructs the fallback wrapper.\n\nWhy nothing catches it. There is no validation anywhere in the config path — sei-db/config/ss_config.go has no Validate() that cross-checks these fields, and app/seidb.go's parseSSConfigs does no cross-field check either. This is asymmetric with the partially-configured case: if an operator sets only ProjectID (say), Configured() returns true, the code proceeds to historical.NewBigtableReader, and BigtableConfig.Validate() immediately fails startup with a clear error ("bigtable instance id is required", etc.). So a partial Bigtable config is loud, but an unconfigured Bigtable backend combined with a nonzero earliest-version is completely silent — no log line, no error, nothing.\n\nImpact is fail-safe, which is why this is a nit and not a blocker. Without the wrapper, GetEarliestVersion() on the plain composite store just returns the real local prune horizon (never a lower coverage floor). Any RPC/consumer height gate that consults the earliest version (e.g. the EVM RPC watermark) therefore continues to reject reads for heights the node genuinely cannot serve — it never admits a read into a gap the (non-existent) fallback can't cover. There is no incorrect data, no crash, and no data loss; the only consequence is that the operator's intended feature has zero effect, and they get no signal telling them why.\n\nStep-by-step proof.\n1. Operator sets historical-offload-earliest-version = 5000000 in app.toml, intending to open historical Get/Has reads down to block 5,000,000 via the Bigtable fallback.\n2. Operator leaves historical-offload-bigtable-project-id, -instance, and -table all as their default empty strings (e.g. the Bigtable block was commented out for local testing and never uncommented for this deploy).\n3. At startup, app/seidb.go's parseSSConfigs reads HistoricalOffloadEarliestVersion = 5000000 into ssConfig correctly, and the empty strings for the Bigtable fields.\n4. ss.NewStateStore builds bigtableCfg from those fields and calls bigtableCfg.Configured(), which evaluates "" != "" || "" != "" || "" != "" = false.\n5. NewStateStore takes the early return at line 31-32 (if !bigtableCfg.Configured() { return primary, nil }) — no wrapper, no error, no log message.\n6. The node runs with the plain composite store. GetEarliestVersion() returns the actual local prune horizon (say, block 9,000,000 after keep-recent pruning), not 5,000,000.\n7. Any caller that expected pruned reads down to block 5,000,000 to succeed via Bigtable instead gets rejected exactly as if the feature were never configured — with nothing in the logs pointing at the historical-offload-earliest-version setting as the culprit.\n\nFix. Add a check in ss.NewStateStore (or in a StateStoreConfig.Validate()) that errors or at least warns when HistoricalOffloadEarliestVersion > 0 but bigtableCfg.Configured() is false, e.g.: if ssConfig.HistoricalOffloadEarliestVersion > 0 && !bigtableCfg.Configured() { logger.Warn("historical-offload-earliest-version is set but no Bigtable backend is configured; the coverage floor has no effect") }. This is a small defensive addition, not a runtime-behavior fix — nothing breaks if it's added later.
…table # Conflicts: # sei-db/state_db/ss/offload/consumer/README.md # sei-db/state_db/ss/offload/historical/metrics.go # sei-db/state_db/ss/offload/historical/metrics_test.go
Kbhat1
changed the title
| } | ||
| return s.primary.GetEarliestVersion() | ||
| } | ||
|
|
||
| func (s *FallbackStateStore) readContext() (context.Context, context.CancelFunc) { | ||
| return context.WithTimeout(context.Background(), readTimeout) | ||
| } | ||
|
|
||
| func (s *FallbackStateStore) Get(storeKey string, version int64, key []byte) ([]byte, error) { | ||
| if !s.shouldFallback(storeKey, version) { | ||
| return s.primary.Get(storeKey, version, key) | ||
| } | ||
| cacheKey := readCacheKey{storeKey: storeKey, version: version, key: string(key)} | ||
| if value, found, ok := s.getCachedValue(cacheKey); ok { | ||
| s.metrics.recordRead("get", fallbackOutcomeCacheHit) | ||
| if !found { | ||
| return nil, nil | ||
| } | ||
| return value, nil | ||
| } | ||
| ctx, cancel := s.readContext() | ||
| defer cancel() | ||
| if err := s.ensureBackendCoverage(ctx, version); err != nil { |
There was a problem hiding this comment.
🟡 In ensureBackendCoverage, the LastVersion coverage-check RPC and the subsequent reader.Get/Has point-read RPC share the same 10s context created by readContext(), even though the readTimeout doc comment says it 'bounds one backend point read.' A slow-but-healthy LastVersion refresh (Bigtable: up to 4 sequential waves across 64 version buckets, each retried up to 3x on Unavailable) can consume most of that budget and cause the point read to fail with context deadline exceeded even though the backend could have served it. Consider giving the coverage check its own timeout instead of reusing the outer read deadline.
Extended reasoning...
The bug. FallbackStateStore.Get/Has (store.go:176-182 and the Has analog) create a single context.Context via readContext(), which applies the package-level readTimeout constant (10s). That one context is passed first into ensureBackendCoverage(ctx, version) and then, if coverage passes, into reader.Get(ctx, ...) / reader.Has(ctx, ...). Both RPCs therefore draw from the same 10-second deadline, even though the readTimeout doc comment explicitly states it "bounds one backend point read."
The code path that triggers it. ensureBackendCoverage (store.go:156-178) is only a no-op when version <= s.backendLast.Load(). Once every ~30s (backendVersionRecheckInterval) for a version above the cached watermark, it calls reader.LastVersion(ctx) using the same context. For the Bigtable reader this is not cheap: BigtableLastVersion scans all 64 version buckets via an errgroup capped at 16 concurrent workers — i.e. up to 4 sequential waves of RPCs — and each individual bucket read is retried up to 3x on Unavailable with growing backoff (readRowsWithRetry). If the backend is partially degraded (e.g. GFE restarts, tablet moves) such that reads are slow/retrying but still eventually succeed, this multi-RPC health check can burn several seconds of the shared 10s budget. Only after it returns does reader.Get/reader.Has run against whatever time remains on the same context, so a point read that would have easily finished within its own fresh 10s window instead fails immediately with context deadline exceeded.
Why nothing else catches this. The existing test (TestFallbackStateStoreRejectsReadsBeyondBackendCoverage and related tests in store_test.go) uses a fake reader whose LastVersion and Get/Has return instantly, so no test exercises the case where LastVersion is slow but still succeeds within the shared deadline. The doc comment on readTimeout describes single-RPC intent, but the implementation never enforces it — it silently bounds two sequential RPCs when a coverage recheck fires.
Impact. The failure mode is a spurious, retryable error (context deadline exceeded, surfaced as the generic fallbackOutcomeError) rather than any data corruption or crash — the caller can just retry, and the next read within the 30s window skips LastVersion entirely (since backendLast was already refreshed or the throttle hasn't elapsed) and gets the point read's full 10s. The trigger window is also fairly narrow: it only matters on a read whose version is above backendLast, once per 30s recheck, and only when LastVersion is slow-but-successful (a hard failure returns errBackendBehind/error immediately regardless of the timeout split). In steady state (version ≤ backendLast), ensureBackendCoverage returns immediately and the point read gets the entire 10s uncontested. This narrow blast radius, combined with the fact that the "fix" (giving each RPC its own timeout) trades off against doubling worst-case caller latency to 20s, is why this is best classified as a minor robustness gap rather than a blocking issue.
Step-by-step proof. (1) A read for version = V arrives where V > backendLast and the 30s recheck interval has elapsed. (2) Get calls readContext(), getting a context with a 10s deadline T. (3) Get calls ensureBackendCoverage(ctx, V), which acquires backendMu and calls reader.LastVersion(ctx). Suppose the backend is in a state where each of the 4 sequential 16-worker waves takes ~1.5s due to a few slow/retrying buckets — this alone consumes ~6s of the 10s budget. (4) LastVersion still succeeds, returning a watermark ≥ V, so ensureBackendCoverage returns nil. (5) Control returns to Get, which now calls reader.Get(ctx, storeKey, key, V) — but ctx now has only ~4s left instead of a fresh 10s. (6) If the point-read RPC itself needs, say, 5s under the same partial degradation (a single-row lookup should normally be fast, but network/backend slowness affects it too), it now fails with context deadline exceeded even though a dedicated 10s window would have let it succeed.
Suggested fix. Derive a fresh, independent timeout for the LastVersion coverage check (e.g. its own shorter deadline) rather than reusing the same context/deadline that's about to be handed to the point read, so the two sequential RPCs don't compete for one shared budget.
There was a problem hiding this comment.
Part 2/2 wiring of the Bigtable historical read fallback: a FallbackStateStore that routes pruned point reads to Bigtable with an LRU cache, coverage gating, and per-store-key prune horizons, plus config/flags/README. The changed code is clean and well-tested; the main caveat is the documented point-read-only scope — advertising the coverage floor while iterators stay on the pruned local store can silently return incomplete range results — plus dependence on part-1 mechanisms that Codex flagged as risky.
Findings: 0 blocking | 5 non-blocking | 1 posted inline
Blockers
- None at the file/PR level.
Non-blocking
- REVIEW_GUIDELINES.md is empty/missing, so no repo-specific guidelines were applied. The Cursor second-opinion pass (cursor-review.md) was also empty — it produced no output.
- Codex P1 (consumer/bigtable.go:145 — version markers written independently by concurrent partition workers, so a higher LastVersion marker can appear before lower-version mutations) and Codex P1 (historical/bigtable.go:396 — reads ignore persisted store rename/upgrade rows) are in part-1 files NOT changed by this PR, so they're out of scope to fix here. However, this PR's coverage gate (ensureBackendCoverage relies on reader.LastVersion) and served reads depend on those mechanisms being correct; verify them before enabling the fallback in production, since a premature LastVersion could let a read at a not-yet-fully-ingested version pass the gate and return incomplete state.
- Minor: cacheValue skips caching when v.Bytes is nil (historical/store.go:236), so a backend 'found' result with a nil/empty value is re-fetched on every read rather than cached. Low impact, but worth a comment on the intended semantics of an existing-but-empty value.
- Minor test gap: metrics_test.go TestFallbackMetricsRecordNoPanic omits fallbackOutcomeBackendBehind from the exercised outcomes; consider including it for completeness.
- 1 suggestion(s)/nit(s) flagged inline on specific lines.
| // floor rather than the local horizon or every historical query is rejected | ||
| // before it reaches the store. Iterators do not use the fallback; range | ||
| // queries between the floor and the local horizon see only local data. | ||
| func (s *FallbackStateStore) GetEarliestVersion() int64 { |
There was a problem hiding this comment.
[suggestion] GetEarliestVersion advertises the coverage floor globally so RPC height gates admit pruned heights, but Iterator/ReverseIterator/RawIterate only ever delegate to the pruned primary. A query at a height in [coverageFloor, localEarliest) that passes the advertised-earliest gate and then iterates (e.g. a Cosmos gRPC prefix/state query, or any range scan) will silently return incomplete or empty data rather than erroring. This matches Codex's P2 and is acknowledged in the comment and README as a known limitation, but it's a real correctness footgun for iteration-based historical queries. Consider erroring (or otherwise signaling unsupported) on iteration below the local prune horizon so callers don't consume silently-partial results.
Superseded: latest AI review found no blocking issues.
Kbhat1
changed the title
Kbhat1
changed the title
Kbhat1
changed the title
1 participant
Description
Testing