Part 1: Bigtable client and Kafka ingestion#3776
Conversation
…rt 1) Part 1 of the Bigtable historical MVCC state offload: the Bigtable data client (hand-rolled over the repo's pinned gRPC v1.57, with bounded read retries, keepalive, and billing-dimension cost metrics), the frozen row key encoding (m|shard|store|key|inverted-version mutation rows and v|bucket|inverted-version version markers), the point-read Reader, and the Kafka consumer that ingests the changelog stream into Bigtable with per-partition ordering, version dedupe, locality-chunked bulk writes, and offset commits only after durable sink writes. Connects to Google Cloud Managed Kafka via TLS + SASL/PLAIN. Part 2 wires the node-side pruned-read fallback on top of this. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
PR SummaryMedium Risk Overview A new Kafka transport is factored into The Reviewed by Cursor Bugbot for commit 03c410b. Bugbot is set up for automated code reviews on this repo. Configure here. |
|
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 @@
## main #3776 +/- ##
==========================================
- Coverage 60.11% 58.93% -1.18%
==========================================
Files 2302 2215 -87
Lines 192098 181401 -10697
==========================================
- Hits 115479 106911 -8568
+ Misses 66291 65092 -1199
+ Partials 10328 9398 -930
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
![seidroid[bot]](https://avatars.githubusercontent.com/in/2745135?s=60&v=4){kind=small}
There was a problem hiding this comment.
Part 1 of the Bigtable historical-state offload adds a well-tested, self-contained gRPC Bigtable client and a Kafka ingestion consumer with no node-behavior changes. No hard blockers, but there is a credential-exposure hardening gap (SASL/PLAIN accepted without TLS) plus two latent-correctness/observability issues (non-contiguous LastVersion, unaggregated lag gauge) worth addressing before the part-2 read path lands.
Findings: 0 blocking | 5 non-blocking | 2 posted inline
Blockers
- None at the file/PR level.
Non-blocking
REVIEW_GUIDELINES.md(from the base branch) andcursor-review.mdare both empty, so no repo-specific guidelines and no Cursor second-opinion pass were available for this synthesis.LastVersion/BigtableLastVersion(historical/bigtable.go) returns the max version marker across buckets. Because Kafka is hash-partitioned and consumed concurrently with per-partition ordering, version N's marker can be durable while lower versions on a slower partition are not yet written — so this value overstates contiguous history. Not consumed in this PR (read fallback is part 2), but if part 2 uses it to gate historical reads/pruning it needs a contiguous watermark, not a max. (Codex High — agreed, downgraded to non-blocker since it's latent in this part.)- General: the mutation-row and version-marker cell timestamps are derived from the block version (
BigtableTimestamp(version)), which is intentional for idempotent redelivery — worth a brief note/test that operators must not enable Bigtable GC-by-age on the column family, or old-height cells could be reaped. - 2 suggestion(s)/nit(s) flagged inline on specific lines.
| switch strings.ToLower(cfg.SASLMechanism) { | ||
| case "", kafkaOptionNone: | ||
| return nil | ||
| case "plain": |
There was a problem hiding this comment.
[suggestion] SASL/PLAIN sends the username and password to the broker in cleartext, but this plain case only validates that credentials are present — it never requires TLS. The aws-msk-iam case below correctly rejects !TLSEnabled; the plain case should do the same so a misconfigured producer/consumer can't leak the service-account credential over an unencrypted connection. The shipped example config and README both set TLSEnabled: true, so this is a defense-in-depth guard, but it's cheap to add:
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")
}(Codex High.)
| } | ||
| m.sinkWriteLatency.Record(ctx, writeLatency.Seconds()) | ||
| if lag >= 0 { | ||
| m.kafkaLag.Record(ctx, lag) |
There was a problem hiding this comment.
[suggestion] kafkaLag is a single unlabeled gauge, and recordBatch is called concurrently by every worker goroutine (one per shard/partition-group). Each worker overwrites the gauge with only its latest batch's lag, so it's last-writer-wins: a low-lag partition can mask another partition that is far behind, and the exported value flaps between workers. Consider a per-partition attribute on the gauge, or tracking a process-wide max/sum, so the metric reflects the worst-case backlog rather than whichever worker recorded last. (Codex Medium.)
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
| } | ||
| if bytes > 0 { | ||
| m.bytesWritten.Add(ctx, bytes, attrs) | ||
| } |
There was a problem hiding this comment.
Hot-path metric attrs reallocated
Medium Severity
recordRead and recordWrite build metric.WithAttributes(attribute.String("table", table)) on every RPC. The table name is fixed at client construction, so this allocates on the ingestion and read hot paths. Existing SeiDB metrics (for example CacheMetrics) precompute the attribute option once at init and reuse it.
Triggered by learned rule: OTel metrics: guard attribute cardinality and use native types
Reviewed by Cursor Bugbot for commit 0c5d3dd. Configure here.
| grpc.WithKeepaliveParams(keepalive.ClientParameters{ | ||
| Time: 30 * time.Second, | ||
| Timeout: 10 * time.Second, | ||
| }), |
There was a problem hiding this comment.
Idle keepalive pings disabled
Low Severity
The client sets gRPC keepalive Time/Timeout specifically to detect silently dropped Bigtable connections, but omits PermitWithoutStream: true. Without that flag, keepalive pings are only sent while a stream is active, so idle connections between consumer batches can still die quietly until the next RPC fails.
Reviewed by Cursor Bugbot for commit 0c5d3dd. Configure here.
Kbhat1
changed the title
Kbhat1
changed the title
Kbhat1
changed the title
Kbhat1
changed the title
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
|
|
||
| switch strings.ToLower(c.SASLMechanism) { | ||
| 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.
🔴 ValidateSASL's "plain" case in sei-db/state_db/ss/offload/kafka.go only checks that Username/Password are non-empty and never requires cfg.TLSEnabled, unlike the aws-msk-iam case a few lines below which does. This lets an operator configure SASLMechanism="plain" with TLSEnabled=false, causing NewKafkaStream/NewKafkaReader to dial a plaintext TCP connection while sending SASL/PLAIN credentials (e.g. a GCP service-account key, per the consumer README) in the clear, since PLAIN is base64-framed but not encrypted. Fix by requiring TLSEnabled in the "plain" branch, mirroring the aws-msk-iam check.
Extended reasoning...
This PR adds a new "plain" case to the shared ValidateSASL function in sei-db/state_db/ss/offload/kafka.go to support Google Cloud Managed Kafka authentication. The new case validates only that Username and Password are non-empty:
case "plain":
if cfg.Username == "" || cfg.Password == "" {
return fmt.Errorf("kafka username and password are required for sasl plain")
}The very next case in the same switch statement, "aws-msk-iam", explicitly requires TLS:
case "aws-msk-iam":
if !cfg.TLSEnabled {
return fmt.Errorf("kafka tls must be enabled for aws-msk-iam")
}SASL/PLAIN transmits the username and password base64-encoded in the SASL exchange, which is a framing, not an encryption. Anyone who can observe the TCP connection can trivially decode the base64 payload and recover the credentials. TLS is what actually protects the credentials on the wire for PLAIN auth — this is precisely why the codebase already enforces TLSEnabled for aws-msk-iam right below it. The PR does not carry that same discipline over to the newly-added plain mechanism.
Code path: Both NewKafkaStream (producer, existing) and NewKafkaReader (new consumer, sei-db/state_db/ss/offload/consumer/kafka.go) gate TLS strictly on cfg.TLSEnabled:
if cfg.TLSEnabled {
dialer.TLS = &tls.Config{MinVersion: tls.VersionTLS12}
}ValidateSASL runs before this and does not block a config where SASLMechanism == "plain" and TLSEnabled == false. So such a config passes validation, and the dialer/mechanism proceed to dial plaintext TCP while still attaching plain.Mechanism{Username, Password} to the connection — sending the credentials over an unencrypted channel.
Step-by-step proof:
- Operator sets
Kafka.TLSEnabled = false(default value if omitted) andKafka.SASLMechanism = "plain"withUsername/Passwordpopulated (e.g. following the README's Google Cloud Managed Kafka example, wherePasswordis a base64-encoded GCP service-account key JSON). KafkaReaderConfig.Validate()callsoffload.ValidateSASL(c.saslConfig()), which only checksUsername != "" && Password != ""for"plain"and returnsnil— no error.NewKafkaReader(orNewKafkaStream) proceeds: sincecfg.TLSEnabledisfalse,dialer.TLSis leftnil, so the connection to the Kafka brokers is a plaintext TCP socket.NewSASLMechanismstill returnsplain.Mechanism{Username, Password}and attaches it to the dialer/transport, so the SASL PLAIN handshake — carrying the base64-framed (not encrypted) credentials — is sent over that plaintext socket.- Any on-path network observer (e.g. anyone with visibility into the VPC/hosting network) can capture the TCP stream and base64-decode the SASL PLAIN frame to recover the GCP service-account credentials.
This is confirmed directly by the PR's own test, TestKafkaReaderConfigValidateSASL in sei-db/state_db/ss/offload/consumer/kafka_test.go: it exercises plain with Username/Password set and never sets TLSEnabled, and asserts cfg.Validate() returns no error — i.e., the test locks in the gap rather than catching it.
Fix: add a TLS check in the "plain" branch, e.g.:
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 mirrors the existing aws-msk-iam guard exactly and is a one-line addition. The shipped example config (config/example-bigtable.json) already sets TLSEnabled: true, so the fix does not affect the documented happy path — it only closes a misconfiguration gap that would otherwise leak credentials.
| "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 scope prefix
Low Severity
Prometheus series use short names like consumer_records_processed_total and consumer_kafka_lag. The meter seidb_offload_consumer does not become a series prefix, so these names are underscoped relative to repo convention (seiwal_*, pebblecache_*, bigtable_*) and the OTel naming rule.
Triggered by learned rule: OTel Prometheus export: namespace prefix and unit auto-suffix naming
Reviewed by Cursor Bugbot for commit bcde772. Configure here.
There was a problem hiding this comment.
A well-structured, well-tested Part 1 that adds a self-contained Bigtable client and a Kafka→Bigtable ingestion consumer with no node-behavior change. No blocking issues; a few robustness/design notes are worth addressing before Part 2 wires in the read fallback.
Findings: 0 blocking | 7 non-blocking | 3 posted inline
Blockers
- None at the file/PR level.
Non-blocking
- Cursor's review file (cursor-review.md) was empty — that pass produced no output, so its perspective is not reflected here.
- Merged from Codex:
LastVersionreturns the max version marker, but because versions are ingested concurrently across partition-sharded workers, a higher version can be marked durable before lower ones finish.LastVersionis therefore a high-water-mark, not a 'complete up to V' guarantee. Harmless in this PR (unused by node code) but the contract should be documented/tightened before Part 2's read fallback relies on it to decide the local-SS retention boundary — otherwise pruned-but-not-yet-ingested versions could read as absent. - Merged from Codex: decoded changelog entries are never structurally validated. A malformed/corrupt message with a negative
VersionreachesbigtableInvertedVersion→uint64FromNonNegativeInt64, which panics; an unrecovered panic in a worker goroutine crashes the whole consumer. Since offsets commit only after a successful write, a single poison message becomes a restart/crash loop. The input stream is the node's own trusted changelog so this is unlikely in practice, but converting these encode-time panics into per-message errors (poison-message handling / dead-letter or skip-with-log) would make ingestion robust.compactMutationsalso assumes non-nilChangesets/Pairselements. - The example config documents storing a base64 service-account key as the Kafka
Passwordin a plaintext JSON file. Consider recommending a secrets manager / env indirection in the README rather than a key committed to an operator's config file. - 3 suggestion(s)/nit(s) flagged inline on specific lines.
| return bigtableValueFromRow(row, r.family) | ||
| } | ||
|
|
||
| func BigtableLastVersion(ctx context.Context, readRows BigtableReadRowsFunc) (int64, error) { |
There was a problem hiding this comment.
[suggestion] LastVersion returns max(bucketVersion) across all buckets. Because consecutive versions are ingested concurrently by partition-sharded workers, a higher version's marker can be written before a lower version finishes, so this value does not guarantee every version below it is present. That's fine here (nothing consumes it yet), but Part 2's read fallback will presumably use it to decide which versions are safely offloaded — please document that this is a high-water-mark, not a contiguity guarantee, or track a distinct 'contiguously-complete' watermark before relying on it for pruning decisions.
| records := make([]Record, 0, len(msgs)) | ||
| var firstVersion, lastVersion int64 | ||
| for i, msg := range msgs { | ||
| entry, err := DecodeEntry(msg.Value) |
There was a problem hiding this comment.
[suggestion] Decoded entries are used without structural validation. A corrupt message with a negative Version flows into BigtableMutationRowKey → bigtableInvertedVersion → uint64FromNonNegativeInt64, which panics; an unrecovered panic in the worker goroutine crashes the process. Because offsets are committed only after a successful write, the same poison message is re-fetched on restart → crash loop. Consider validating the entry here (non-negative version, non-nil changesets/pairs) and returning it as a bounded/skippable error instead of letting encoding panic.
There was a problem hiding this comment.
+1, If a Kafka message contains invalid protobuf, DecodeEntry returns an error, processBatch returns the error, the worker kills the errgroup, Run returns, and main calls log.Fatalf. On restart, the consumer re-fetches the same uncommitted message and crashes again — infinite loop. There's no dead-letter mechanism, no skip-after-N-failures, no offset advance for undecodable messages.
| // followed by a tombstone leaves both cells on the row. | ||
| func (s *bigtableSink) mutationRow(version int64, storeName string, pair *proto.KVPair) historical.BigtableRowMutation { | ||
| ts := historical.BigtableTimestamp(version) | ||
| deleted := pair.Delete || pair.Value == nil |
There was a problem hiding this comment.
[nit] deleted := pair.Delete || pair.Value == nil conflates a live write of a nil value with a tombstone, so a genuine nil-value write is stored (and later read) as absent. bigtableValueFromRow mirrors this (value == nil → ErrNotFound), so it's self-consistent, but worth a comment confirming SS never emits a live nil-value pair — otherwise such writes silently disappear.
| } | ||
| } | ||
|
|
||
| func sleepWithContext(ctx context.Context, d time.Duration) error { |
There was a problem hiding this comment.
Identical implementations in both consumer/consumer.go and historical/bigtable.go. Extract to a shared util package.
| "Username": "kafka-client@my-gcp-project.iam.gserviceaccount.com", | ||
| "Password": "<base64-encoded service account key JSON>" |
There was a problem hiding this comment.
Kafka SASL passwords live in a plain JSON file on disk. No secret manager integration, no envelope encryption. For a production system handling chain state, this is a security concern. At minimum, support reading credentials from environment variables or a secrets manager.
|
|
||
| const ( | ||
| DefaultBigtableFamily = "state" | ||
| DefaultBigtableShards = 256 |
There was a problem hiding this comment.
The shard value is baked into every mutation row key: FNV32a(storeName + \0 + key) % Shards. If an operator changes Shards from 256 to 512, every Get and Has query computes a different shard for the same (storeName, key), so it looks up the wrong row prefix and returns ErrNotFound. All previously ingested data becomes unreachable with no error. This needs to either be persisted in a metadata row and validated on startup, or enforced as immutable in the config.
| if len(errs) != len(rows) { | ||
| return fmt.Errorf("bigtable returned %d mutation results for %d rows", len(errs), len(rows)) | ||
| } | ||
| for i, rowErr := range errs { |
There was a problem hiding this comment.
If 50 out of 1024 rows fail, you see one error and no indication of how widespread the failure was. This makes it much harder to distinguish "one hot tablet is overloaded" from "everything is broken."
Consider Include the failure count:
return fmt.Errorf("%d of %d rows failed, first: row %q: %w", failCount, len(rows), rows[firstIdx].RowKey, errs[firstIdx])
| func (c *Consumer) writeBatchWithRetry(ctx context.Context, records []Record) error { | ||
| backoff := sinkBaseBackoff | ||
| var lastErr error | ||
| for attempt := 1; attempt <= sinkMaxAttempts; attempt++ { |
There was a problem hiding this comment.
writeBatchWithRetry retries the entire batch even after partial success, If 1023 out of 1024 mutation rows succeed and 1 fails, the retry re-sends all 1024. Since SetCell is idempotent at the same timestamp this is correct, but it amplifies write cost and latency proportional to batch size. A targeted retry of only failed rows would be more efficient — ApplyBulk already returns per-row errors.
| records := make([]Record, 0, len(msgs)) | ||
| var firstVersion, lastVersion int64 | ||
| for i, msg := range msgs { | ||
| entry, err := DecodeEntry(msg.Value) |
There was a problem hiding this comment.
+1, If a Kafka message contains invalid protobuf, DecodeEntry returns an error, processBatch returns the error, the worker kills the errgroup, Run returns, and main calls log.Fatalf. On restart, the consumer re-fetches the same uncommitted message and crashes again — infinite loop. There's no dead-letter mechanism, no skip-after-N-failures, no offset advance for undecodable messages.
| if err := c.Kafka.Validate(); err != nil { | ||
| return fmt.Errorf("kafka: %w", err) | ||
| } | ||
| bigtable := c.Bigtable |
There was a problem hiding this comment.
BigtableConfig is a struct (not a pointer), so bigtable := c.Bigtable creates a copy.
ApplyDefaults() fills in Family = "state" and Shards = 256 on the copy. Then Validate() passes because the copy has those defaults. But the original c.Bigtable is untouched — c.Bigtable.Family is still "".
It happens to work in practice because NewBigtableSink calls cfg.ApplyDefaults() again before using it. But if any code between LoadConfig and NewBigtableSink reads cfg.Bigtable.Family (e.g., for logging or metrics tagging), it sees an empty string despite validation having passed.
The fix is simple — apply defaults on the original, not a copy.
| for i, msg := range msgs { | ||
| entry, err := DecodeEntry(msg.Value) | ||
| if err != nil { | ||
| return fmt.Errorf("decode message at offset %d: %w", msg.Offset, err) |
There was a problem hiding this comment.
Poison messages crash the consumer
High Severity
DecodeEntry failures return from processBatch, kill the errgroup, and main calls log.Fatalf without committing offsets. On restart the same uncommitted message is fetched again, causing a crash loop. A negative Version also panics in uint64FromNonNegativeInt64 with the same outcome. There is no dead-letter, skip-after-N, or offset advance for poison payloads.
Additional Locations (2)
Reviewed by Cursor Bugbot for commit b7b8327. Configure here.
| } | ||
| if c.Shards == 0 { | ||
| c.Shards = DefaultBigtableShards | ||
| } |
There was a problem hiding this comment.
Shard count silently orphans data
High Severity
Shards is part of every mutation row key via FNV32a(storeName + \0 + key) % Shards, but it is only a free config field with a default. Changing Shards after ingestion makes Get/Has compute a different shard prefix and return ErrNotFound, so previously written history becomes unreachable with no error. Nothing persists or validates the shard count against the table.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit b7b8327. Configure here.
| // "aws-msk-iam". | ||
| SASLMechanism string | ||
| Username string | ||
| Password string |
There was a problem hiding this comment.
Plaintext Kafka credentials on disk
Medium Severity
SASL Username/Password are loaded only from a plain JSON config file. For Managed Kafka this password is a base64 service-account key. There is no env-var or secret-manager path, so production deployments are pushed toward storing long-lived credentials on disk in cleartext.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit b7b8327. Configure here.
There was a problem hiding this comment.
Part 1 of the Bigtable historical-offload work: a self-contained Bigtable gRPC client, MVCC row-key encoding, and a Kafka→Bigtable ingestion consumer with per-partition ordering, redelivery dedupe, and offset-after-write semantics. Code is well-tested and cleanly structured, but the shared SASL validation now accepts SASL/PLAIN with no TLS requirement, allowing Kafka credentials to be sent in cleartext.
Findings: 1 blocking | 3 non-blocking | 1 posted inline
Blockers
- None at the file/PR level.
- 1 blocking issue(s) flagged inline on specific lines.
Non-blocking
BigtableLastVersion(sei-db/state_db/ss/offload/historical/bigtable.go:412) returns the maximum version marker across buckets. Because the consumer shards by partition and writes concurrently, and version markers are keyed only by version (no partition), a higher version can be marked durable before a lower one's mutation rows land.LastVersioncan therefore advertise a max that is ahead of a contiguous, fully-written prefix. No node-behavior impact in this PR (the read fallback lands in part 2), but before part 2 uses this as a retention/availability boundary, consider tracking a contiguous low-watermark rather than the max, or document that LastVersion is not a completeness guarantee. (Raised by Codex.)- The Cursor second-opinion review file (
cursor-review.md) was empty — that pass produced no output. - Consumer/consumer.go retries
sink.WriteBatchup to 5 attempts with backoff and, on final failure, returns an error that abortsRun; this is intentional (offsets are only committed after a durable write) but means a persistently failing sink halts the whole consumer. Confirm that's the desired operational behavior (crash-and-restart) vs. per-shard isolation.
| switch strings.ToLower(cfg.SASLMechanism) { | ||
| case "", kafkaOptionNone: | ||
| return nil | ||
| case "plain": |
There was a problem hiding this comment.
[blocker] SASL/PLAIN is accepted here with no TLS requirement. SASL/PLAIN transmits the username and password effectively in cleartext, so a plain config with TLSEnabled: false leaks Kafka credentials on the wire. This is now reachable on the producer path too (NewKafkaStream → KafkaConfig.Validate → ValidateSASL), not just the new consumer. The aws-msk-iam case already requires TLS — apply the same guard here:
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")
}(Flagged by Codex as P1.)
| func (c BigtableConfig) Configured() bool { | ||
| return strings.TrimSpace(c.ProjectID) != "" || | ||
| strings.TrimSpace(c.InstanceID) != "" || | ||
| strings.TrimSpace(c.Table) != "" |
There was a problem hiding this comment.
Setting just Project without Instance or Table returns true, and the code proceeds to build a Bigtable reader with empty connection parameters. This should be AND (all required fields present)
| @@ -0,0 +1,126 @@ | |||
| package consumer | |||
There was a problem hiding this comment.
Would recommend to put Kafka consumer and publisher together into a separate package, not under ss. Something like sei-db/queue/kafka
| @@ -0,0 +1,233 @@ | |||
| package consumer | |||
There was a problem hiding this comment.
Would recommend bundle all BigTable libraries together into a DB library, similar to littDB, and move them into sei-db/db_engine/bigtable/
Configured() ORed the three connection fields, so a config with only ProjectID set read as configured and then failed later when the client opened. Require ProjectID, InstanceID, and Table to all be non-blank so partially filled configs are treated as not configured. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Pull the Kafka plumbing out of the state-store offload package into a standalone queue transport library: - writer construction (WriterConfig, SASL validation and mechanisms, the AWS MSK IAM signer) moves from offload, - reader construction (ReaderConfig) moves from offload/consumer, - the generic batching consume loop (Consumer, Options, Sink) and its throughput/lag metrics move from offload/consumer. The loop is now payload-agnostic: Sink.WriteBatch receives raw Kafka messages and owns decoding. The Bigtable sink decodes changelog entries itself, so offload/consumer keeps only the changelog-to-Bigtable application. offload.KafkaConfig stays as a type alias so existing callers keep compiling. Meter and metric names are unchanged. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Bundle the Bigtable pieces into a standalone DB library alongside the other db_engine backends: the gRPC data-plane client (ReadRows chunk assembly, MutateRows accounting, bounded retries), the frozen row-key and cell encoding, Config, and the per-RPC cost metrics on the seidb_bigtable meter. The now-redundant Bigtable prefix is dropped from names (BigtableClient -> Client, BigtableMutationRowKey -> MutationRowKey, ...); encoding behavior and metric names are unchanged and the encoding tests move with the code with assertions untouched. offload/historical keeps the Reader contract and a slim Bigtable-backed reader adapter (Get/Has/LastVersion plus mutation-row interpretation); the offload consumer sink now builds row mutations against the new package directly. 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 2 potential issues.
There are 8 total unresolved issues (including 6 from previous reviews).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 03c410b. Configure here.
| for _, msg := range msgs { | ||
| entry, err := DecodeEntry(msg.Value) | ||
| if err != nil { | ||
| return fmt.Errorf("decode message at offset %d: %w", msg.Offset, err) |
There was a problem hiding this comment.
Poison-pill decode crash loop
High Severity
An undecodable Kafka payload makes DecodeEntry fail, WriteBatch fail, and after retries the worker tears down Run. main then log.Fatalfs. On restart the same uncommitted offset is fetched again, so one bad message can stall ingestion indefinitely with no dead-letter or skip path.
Additional Locations (2)
Reviewed by Cursor Bugbot for commit 03c410b. Configure here.
| _, _ = h.Write([]byte{0}) | ||
| _, _ = h.Write(key) | ||
| return uint16FromBoundedUint32(h.Sum32() % uint32FromBoundedInt(shards)) | ||
| } |
There was a problem hiding this comment.
Shards change orphans all data
High Severity
Row keys embed FNV32a(storeName + NUL + key) % Shards, but Shards is only a config default with no persisted metadata or startup check. Changing Shards remaps every key prefix, so Get/Has miss existing rows and return not-found while the data still sits in Bigtable.
Additional Locations (2)
Reviewed by Cursor Bugbot for commit 03c410b. Configure here.
There was a problem hiding this comment.
Part 1 adds a self-contained Bigtable data-plane client and a Kafka-to-Bigtable ingestion consumer with strong unit-test coverage and zero node-behavior change. No blockers, but the consumer's poison-message handling (unvalidated version panic, byte-unbounded batches) can crash-loop ingestion and is worth hardening.
Findings: 0 blocking | 5 non-blocking | 2 posted inline
Blockers
- None at the file/PR level.
Non-blocking
- Poison-message handling: because Kafka offsets commit only after a successful write, any record that fails deterministically (undecodable payload, invalid version) is re-fetched forever and crash-loops the consumer, permanently halting ingestion. Consider dead-lettering or skip-with-metric for records that fail after N attempts, and distinguish permanent decode/validation failures from transient sink errors (writeBatchWithRetry currently burns all 5 retries on a deterministic decode error).
- Cursor second-opinion review (cursor-review.md) was empty — that pass produced no output.
- Positive: the new packages carry thorough table-driven unit tests (row-key ordering, split-cell assembly, retry semantics, compaction/dedup, chunk locality, concurrency, metrics nil-safety), and metrics are cleanly nil-safe.
- 2 suggestion(s)/nit(s) flagged inline on specific lines.
| func (s *bigtableSink) mutationRow(version int64, storeName string, pair *proto.KVPair) bigtable.RowMutation { | ||
| ts := bigtable.Timestamp(version) | ||
| deleted := pair.Delete || pair.Value == nil | ||
| rowKey := bigtable.MutationRowKey(storeName, pair.Key, version, s.shards) |
There was a problem hiding this comment.
[suggestion] A negative entry.Version panics rather than erroring: MutationRowKey → invertedVersion → uint64FromNonNegativeInt64 calls panic() for value < 0 (db_engine/bigtable/bigtable.go:721-727). DecodeEntry does no semantic validation, so a single Kafka record with a negative version panics WriteBatch, and since offsets are only committed after a successful write, the record is re-fetched on restart and crash-loops the consumer. Validate entry.Version >= 0 when decoding/compacting and return a controlled error instead of letting it reach the panic path.
| } | ||
|
|
||
| func (s *bigtableSink) applyRecordRowMutations(ctx context.Context, rows []bigtable.RowMutation) error { | ||
| chunks := bigtableRowMutationChunks(rows, s.bulkChunkRows) |
There was a problem hiding this comment.
[suggestion] bigtableRowMutationChunks bounds a chunk only by row count (defaultBigtableMutationChunkRows = 1024), never by serialized size. Bigtable's MutateRows caps the request at ~20 MiB; a 1024-row chunk of large changelog values (e.g. big contract storage/code) can exceed that and fail with a non-retryable error, which writeBatchWithRetry then retries to exhaustion and halts ingestion. Consider also splitting chunks when accumulated mutationSize crosses a byte threshold.
Superseded: latest AI review found no blocking issues.
| func (s *bigtableSink) WriteBatch(ctx context.Context, msgs []kafkago.Message) error { | ||
| records := make([]Record, 0, len(msgs)) | ||
| for _, msg := range msgs { | ||
| entry, err := DecodeEntry(msg.Value) | ||
| if err != nil { | ||
| return fmt.Errorf("decode message at offset %d: %w", msg.Offset, err) | ||
| } |
There was a problem hiding this comment.
🔴 A single malformed/corrupt Kafka message permanently wedges this consumer: bigtableSink.WriteBatch returns a decode error that writeBatchWithRetry retries as if transient, then propagates through processBatch/Consumer.Run to main.go's log.Fatalf, but since offsets are only committed after a successful write, the same poison message is re-fetched and the process crashes again on every restart. Worse, a decoded entry with a negative Version skips the error path entirely and panics unrecovered inside the worker goroutine via uint64FromNonNegativeInt64, crashing the process outright with the same uncommitted-offset outcome. Recommend validating decoded entries (non-negative version, well-formed changesets) and routing invalid payloads to a bounded/skippable or dead-letter path instead of retrying/panicking indefinitely.
Extended reasoning...
The bug: bigtableSink.WriteBatch (sei-db/state_db/ss/offload/consumer/bigtable.go:63-69) calls DecodeEntry(msg.Value) for every message in the batch and returns a wrapped error the first time decoding fails:
entry, err := DecodeEntry(msg.Value)
if err != nil {
return fmt.Errorf("decode message at offset %d: %w", msg.Offset, err)
}This is a permanent error — the same bytes will never decode successfully — but nothing downstream treats it that way.
The path to an unrecoverable loop: Consumer.writeBatchWithRetry (sei-db/queue/kafka/consumer.go) only short-circuits on isCancellation(err) (context cancel/deadline); a decode error doesn't match that, so it gets retried sinkMaxAttempts (5) times with backoff, re-decoding and failing identically each time, before finally being returned. processBatch wraps and returns it; workerLoop returns it because it isn't a cancellation, killing the errgroup; Consumer.Run's g.Wait() returns the error; and cmd/historical-offload-consumer/main.go does if err := c.Run(ctx); err != nil { log.Fatalf(...) }, which os.Exit(1)s. Crucially, processBatch only calls reader.CommitMessages after writeBatchWithRetry succeeds, so the poison message's offset is never committed. On restart, the kafka-go consumer group resumes from the last committed offset, re-fetches the exact same message, and crashes again — an infinite loop with no dead-letter path, no skip-after-N-failures, and no offset-advance mechanism for undecodable payloads.
The panic variant is worse: once an entry does decode but carries a negative Version (e.g. from a corrupted/truncated field, or any producer bug/schema mismatch), it bypasses the error-return path entirely. WriteBatch -> writeRecordRows -> appendRecordRowMutations -> mutationRow -> bigtable.MutationRowKey -> mutationRowKeyBytes -> invertedVersion -> uint64FromNonNegativeInt64, which does panic(fmt.Sprintf(\"bigtable version %d is negative\", value)) for value < 0. This executes synchronously inside a worker goroutine spawned by errgroup.Go, and nothing in that call chain has a recover(), so the whole process crashes immediately — again before any offset commit, producing the identical restart-refetch-crash cycle via a panic instead of a clean error return.
Why nothing today prevents this: writeBatchWithRetry is generic retry logic with no notion of permanent-vs-transient errors (it only recognizes context cancellation as non-retryable), and WriteBatch/mutationRow never validate the decoded ChangelogEntry before using it structurally. The offset-commit-after-write invariant, which is otherwise a correct at-least-once design, is exactly what turns a single bad payload into a permanent outage rather than a one-time skip.
Concrete walkthrough:
- A Kafka message arrives with a corrupted protobuf payload (bit flip in transit, truncation, wrong topic/schema) at offset N.
WriteBatchcallsDecodeEntry, which fails and returns an error.writeBatchWithRetryretries 5 times (each attempt re-decodes and fails identically), then returns the error up throughprocessBatch,workerLoop, and theerrgroup.Consumer.Runreturns the error;main.gocallslog.Fatalf, exiting the process. Offset N was never committed.- The process supervisor (systemd/k8s) restarts the binary. The kafka-go consumer group resumes at the last committed offset, which is still before N.
FetchMessagereturns the same poison message at offset N again, and steps 2-4 repeat forever, with no operator signal beyond repeated crash logs, until someone manually seeks the consumer group past offset N.
Fix direction: validate decoded entries where they're first inspected (non-negative Version, well-formed changesets/pairs), and treat both undecodable payloads and structurally invalid entries as a bounded/skippable failure — e.g. log-and-skip with a counter/alert, or write to a dead-letter topic/table — rather than retrying indefinitely or letting an encoding helper panic. This is a real availability gap: it requires a corrupted/malformed message to trigger (the in-repo producer NewKafkaStream always marshals a valid, non-negative-version ChangelogEntry, so it won't fire in normal operation), and it only affects the standalone historical-offload-consumer binary with zero impact on the sei node/consensus — but when it does trigger, the pipeline is permanently wedged until a human manually advances the Kafka offset, which is a concrete operational failure rather than a cosmetic issue.
| func (m *consumerMetrics) recordBatch(ctx context.Context, records int64, lag int64, writeLatency time.Duration) { | ||
| if m == nil { | ||
| return | ||
| } | ||
| if records > 0 { | ||
| m.recordsProcessed.Add(ctx, records) | ||
| } | ||
| m.sinkWriteLatency.Record(ctx, writeLatency.Seconds()) | ||
| if lag >= 0 { | ||
| m.kafkaLag.Record(ctx, lag) | ||
| } |
There was a problem hiding this comment.
🟡 consumer_kafka_lag is a single unlabeled Int64Gauge, but Consumer.Run in sei-db/queue/kafka/consumer.go spawns one worker goroutine per shard, each of which calls recordBatch after every batch it writes with no partition/shard attribute. Since these workers run concurrently, whichever finishes last overwrites the gauge process-wide, so a caught-up partition can mask a partition that is far behind. Consider adding a per-partition/shard attribute, or tracking a process-wide max across shards.
Extended reasoning...
consumerMetrics.kafkaLag (sei-db/queue/kafka/metrics.go:40-44) is created as a single OTel Int64Gauge with no attributes:
kafkaLag, _ := meter.Int64Gauge(
"consumer_kafka_lag",
...
)recordBatch (metrics.go:63-65) records into it unconditionally with no attribute set:
if lag >= 0 {
m.kafkaLag.Record(ctx, lag)
}Consumer.Run (consumer.go) spawns c.workers goroutines (16 by default) via workerLoop, one per partition shard (shardFor(partition, workers)). Each worker independently calls processBatch -> c.metrics.recordBatch after every batch it writes and commits. Because every worker records to the same gauge under the same (empty) attribute set, OTel's synchronous-gauge semantics mean the exported value is simply "the last Record call wins" — there is no aggregation across concurrent recorders for a synchronous gauge.
The struct comment says "One consumer runs per process, so no attributes are needed," but that reasoning only accounts for there being one consumer instance per process — it misses that a single consumer instance runs many concurrent per-shard workers, each of which is an independent recorder into the same series.
Concrete walkthrough:
- Partition A is 5,000 messages behind (
lag = 5000); partition B is fully caught up (lag = 0). - Worker A finishes its batch and calls
kafkaLag.Record(ctx, 5000). - A few milliseconds later, worker B (a different partition, different goroutine) finishes its own unrelated batch and calls
kafkaLag.Record(ctx, 0)— same gauge, same attribute set. - When Prometheus scrapes
consumer_kafka_lagbetween these two events, it reads whatever value was written most recently — in this example0, even though partition A is still 5,000 messages behind. - Because batch completion order across 16 concurrent workers is driven by goroutine scheduling and per-partition throughput rather than any deterministic ordering, the exported metric flaps unpredictably and does not reliably reflect the true worst-case consumer lag. An operator relying on this metric for alerting could miss a partition that is genuinely falling far behind, because a healthy partition's
Record(0)call happened to land last before the scrape.
This is purely an observability-accuracy issue: it does not affect message ordering, offset-commit correctness, or data durability in the ingestion pipeline itself — the sink still writes and commits every message correctly regardless of what the gauge reports.
Fix: either add a per-partition (or per-shard) attribute to the gauge so Prometheus exposes one time series per partition/shard (letting max() be computed in the query layer), or maintain a process-wide running max across shards (e.g. via a small mutex-guarded or atomic max tracker) and record only that value into the gauge.
| }) | ||
| } | ||
| entries = append(entries, entry) | ||
| } | ||
| stream, err := c.data.MutateRows(ctx, &bigtablepb.MutateRowsRequest{ | ||
| TableName: c.tableName, | ||
| AppProfileId: c.appProfile, | ||
| Entries: entries, | ||
| }) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| rowErrs := make([]error, len(rows)) | ||
| seen := make([]bool, len(rows)) | ||
| for { |
There was a problem hiding this comment.
🟡 Bigtable's Shards config value is baked directly into every mutation row key (shardOf(storeName, key, shards) = FNV32a(storeName+\0+key) % shards in mutationRowPrefixBytes), but that shard count is never persisted anywhere or validated against what earlier data was written with. If an operator changes Shards between the ingestion consumer and a later historical reader (or between ingestion runs), Get/Has/MutationRowRange will compute a different shard prefix for the same key and silently return ErrNotFound/false for data that is still physically present, with no error surfaced anywhere.
Extended reasoning...
What the bug is. Config.Shards (sei-db/db_engine/bigtable/bigtable.go) is hashed into the second/third bytes of every mutation row key via shardOf(storeName, key, shards) = FNV32a(storeName + \0 + key) % shards, used inside mutationRowPrefixBytes (lines ~302-316), which in turn backs MutationRowKey and MutationRowRange. Both the write path (bigtableSink.shards = cfg.Shards in consumer/bigtable.go) and the new read path (bigtableReader.shards = cfg.Shards in historical/bigtable.go, used by Get/Has via MutationRowRange) derive this shard count purely from whatever value is in the live process config at startup. Nothing writes the shard count that was actually used to build existing rows into a metadata row, and nothing checks a later run's Shards against it.
Code path that triggers it. Config.Validate() only range-checks 1 <= Shards <= 65535 — it has no notion of 'the value this table was already written with.' ApplyDefaults() fills in DefaultShards = 256 when unset, so an operator who omits Shards from one config and later adds an explicit value (or changes 256 -> 512 for better tablet spread) changes the shard byte computed for every (storeName, key) pair. Get/Has then call bigtable.MutationRowRange(storeName, key, targetVersion, r.shards), which hashes with the new shards value and scans a row-key prefix that the old data was never written under.
Why nothing prevents it. The shard count is effectively part of the on-disk key-encoding format (like the row-key layout itself), but it is treated as an ordinary, freely-editable, defaulted config knob with no persistence or immutability enforcement. There is no metadata row recording the shard count used at ingest time, and no startup check comparing the configured value against it.
Impact. The failure is completely silent: Get returns ErrNotFound and Has returns false for keys that are still present in Bigtable under the old shard prefix — indistinguishable from genuinely missing data, with no error, warning, or metric indicating a shard mismatch occurred. This is a silent, undetected data-unreachability bug equivalent to data loss from the caller's perspective.
Step-by-step proof.
- Operator runs the ingestion consumer with
Shards: 256(or omitted, defaulting to 256). Entrybank/k1 = v1at version 40 is written to row keym | shardOf("bank", "k1", 256) | "bank" | "k1" | inverted(40). - Operator later reconfigures
Shards: 512(e.g. to spread hot tablets), either for a fresh ingestion run or for the historical reader process that serves reads. - A read for
bank/k1at version >= 40 callsMutationRowRange("bank", []byte("k1"), targetVersion, 512), which computesshardOf("bank", "k1", 512)— a different value thanshardOf("bank", "k1", 256)in virtually all cases (different modulus over the same FNV32a hash). ReadRowsscans the prefix for the new shard, finds no row there (the real row lives under the old shard's prefix), andGetreturnsErrNotFound/Hasreturnsfalse— even thoughbank/k1's data is untouched and fully present in the table.
Fix. Either (a) persist the shard count actually used at ingest time in a metadata row and validate it against the configured value on startup (failing loudly on mismatch), or (b) explicitly document/enforce Shards as immutable once any data has been ingested for a table.
On severity: verifiers split between normal and nit on this finding. The case for nit is that the code is internally correct for any single fixed Shards value, the failure requires an operator to actively change a documented config field (not something that happens on ordinary use or by merging this PR as-is), the shipped example config pins Shards: 256, and per the PR description this is Part 1 (client + ingestion) — the node-side read fallback that would actually expose orphaned data to a live caller lands in Part 2, so nothing observes this failure mode today. Given that framing, I'm filing this as a hardening/defense-in-depth gap rather than a blocking defect, while flagging that reviewer yzang2019 explicitly asked for it to be persisted+validated or enforced immutable before this ships as production ingestion — worth prioritizing before Part 2 wires up live reads.
🔬 also observed by yzang2019
| func (c *Config) Validate() error { | ||
| if err := c.Kafka.Validate(); err != nil { | ||
| return fmt.Errorf("kafka: %w", err) | ||
| } | ||
| bigtable := c.Bigtable | ||
| bigtable.ApplyDefaults() | ||
| if err := bigtable.Validate(); err != nil { | ||
| return fmt.Errorf("bigtable: %w", err) | ||
| } |
There was a problem hiding this comment.
🟡 In Config.Validate(), bigtable := c.Bigtable copies the BigtableConfig value before calling ApplyDefaults()/Validate() on it, so validation succeeds using the defaulted Family/Shards while the original c.Bigtable (what LoadConfig returns to callers) still has Family=="" and Shards==0. This currently causes no observable failure since NewBigtableSink (and NewBigtableReader) call ApplyDefaults() again before use, but it's a latent footgun: a green Validate() doesn't guarantee the struct it validated is the one callers actually hold. Fix by calling c.Bigtable.ApplyDefaults() on the original field (or writing the copy back) before validating.
Extended reasoning...
What the bug is: Config.Validate() in sei-db/state_db/ss/offload/consumer/config.go does:
bigtable := c.Bigtable // value copy — BigtableConfig (aka bigtable.Config) is a struct, not a pointer
bigtable.ApplyDefaults() // mutates the copy only
if err := bigtable.Validate(); err != nil { ... } // validates the copyApplyDefaults() fills in Family = "state" and Shards = 256 on the local copy bigtable, and Validate() is called on that same copy. The original c.Bigtable — the field that LoadConfig actually returns to callers — is never touched. So after a successful LoadConfig, cfg.Bigtable.Family == "" and cfg.Bigtable.Shards == 0 whenever the JSON config omitted them, even though Validate() reported success.
Code path: LoadConfig → cfg.applyDefaults() (only touches MaxBatchRecords/BatchMaxWaitMS, not Bigtable) → cfg.Validate() (defaults a throwaway copy of Bigtable) → returns cfg with an un-defaulted Bigtable field.
Why existing code doesn't prevent it: it happens to work today purely by coincidence of call order. main.go passes cfg.Bigtable straight into consumer.NewBigtableSink, which itself calls cfg.ApplyDefaults() again before Validate()/NewClient() (see bigtable.go: func NewBigtableSink(cfg BigtableConfig) { cfg.ApplyDefaults(); ... }). historical.NewBigtableReader does the same. Neither of those paths ever reads cfg.Bigtable.Family/Shards between LoadConfig returning and the second ApplyDefaults() call, so the un-defaulted zero values are never observed in the current codebase.
Impact: none today — this is a purely latent inconsistency, not a reachable failure. It becomes a real bug the moment any code reads cfg.Bigtable.Family/Shards between LoadConfig and sink/reader construction (e.g. logging the effective config, tagging metrics with the family/shard count, or the Part 2 read-path reader construction this consumer's README describes as still to come) — that code would see Family==""/Shards==0 despite Validate() having reported success, which is a surprising and easy-to-miss trap for future maintainers.
Proof (step-by-step): 1) Write a config JSON where Bigtable omits Family/Shards (like the shipped example-bigtable.json doesn't, but a minimal user config could). 2) Call cfg, err := consumer.LoadConfig(path) — err is nil because Config.Validate()'s local bigtable copy got defaulted to Family="state", Shards=256 and passed its own Validate(). 3) Inspect cfg.Bigtable.Family and cfg.Bigtable.Shards — they are still "" and 0, contradicting the fact that Validate() just succeeded. 4) Today, main.go immediately calls consumer.NewBigtableSink(cfg.Bigtable), which calls ApplyDefaults() a second time and masks the problem — but any code inserted between steps 2 and 4 that reads cfg.Bigtable.Family/Shards directly would observe the zero values.
Fix: apply defaults to c.Bigtable itself before validating, e.g. c.Bigtable.ApplyDefaults(); if err := c.Bigtable.Validate(); err != nil { return fmt.Errorf("bigtable: %w", err) }, removing the throwaway copy.
Severity: all verifiers agree the code path described is factually accurate, but also agree there is no concrete, reachable failure in this PR as merged — the only two current consumers of cfg.Bigtable re-apply defaults themselves. This is a one-line correctness/maintainability fix worth making, but not something that should block merge.
2 participants
Description
Testing