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
Reviewed by Cursor Bugbot for commit bcde772. 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.14% 58.95% -1.19%
==========================================
Files 2303 2212 -91
Lines 192032 181473 -10559
==========================================
- Hits 115491 106995 -8496
+ Misses 66230 65075 -1155
+ Partials 10311 9403 -908
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.
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.
There are 4 total unresolved issues (including 2 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 bcde772. Configure here.
| m.sinkWriteLatency.Record(ctx, writeLatency.Seconds()) | ||
| if lag >= 0 { | ||
| m.kafkaLag.Record(ctx, lag) | ||
| } |
There was a problem hiding this comment.
Lag gauge races across workers
Medium Severity
consumer_kafka_lag is a single process-wide gauge updated from every worker after each batch. With concurrent workers, last writer wins, so a low-lag partition can overwrite a high-lag one and the scraped value no longer reflects true consumer lag.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit bcde772. 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 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.
| // 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.
1 participant
Description
Testing