fix(cdc): handle empty LSN/SeqVal in native ID dedup - #251
Merged
Conversation
added 3 commits
May 12, 2026 12:53
Replace content-based FNV-1a hash with native CDC unique identifier for row deduplication. - Remove ComputeChangeID (FNV-1a hash over txid, table, data, lsn, operation) - Add ComputeNativeID using __ + __ + __ tuple - The native CDC key is guaranteed unique by SQL Server CDC engine, eliminating potential hash collisions from content-based approach - Add SeqVal field to Change struct to capture sequence value within transaction
- ComputeNativeID: return empty string when LSN or SeqVal is nil/empty - GetChanges: log warning on seqval extraction failure instead of silent ignore - Add tests for empty input handling and duplicate ID dedup behavior - Integration test: use ComputeNativeID instead of manual hex formatting
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've left some high level feedback:
- Removing the table+primary-key-based ID recomputation in
convertToCoreChangesmeans that when__$seqvalis absent you now end up with an empty ID fromComputeNativeID, which can collapse many distinct rows onto the same ID; consider keeping a deterministic non-empty fallback (e.g., table+PK) instead of returning an empty string in these cases. - Both
ComputeNativeIDandGetChangesnow emitslog.Warnin fairly low-level paths, which may be hit very frequently in production; consider downgrading to debug/info or adding some rate limiting/aggregation to avoid log noise under normal operation.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Removing the table+primary-key-based ID recomputation in `convertToCoreChanges` means that when `__$seqval` is absent you now end up with an empty ID from `ComputeNativeID`, which can collapse many distinct rows onto the same ID; consider keeping a deterministic non-empty fallback (e.g., table+PK) instead of returning an empty string in these cases.
- Both `ComputeNativeID` and `GetChanges` now emit `slog.Warn` in fairly low-level paths, which may be hit very frequently in production; consider downgrading to debug/info or adding some rate limiting/aggregation to avoid log noise under normal operation.Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
WalkthroughFetch no longer computes or stores a native ID or seqval; convertToCoreChanges now derives core.Change.ID from LSN, table, table keys, and operation. ComputeNativeID returns an empty string and logs a warning when LSN or seqval are empty; tests assert exact hex:hex:op formatting and empty behavior. GetChanges logs warnings on __$seqval extraction failures and only sets SeqVal when a non-nil value is returned. The core Change struct removed SeqVal. Store documents the native ID format and a new test verifies duplicate IDs are ignored. Integration test ID formatting simplified to hex(LSN):table:op. Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
tests/integration/integration_test.go (1)
206-211: ⚡ Quick winUse real
c.SeqValin integration ID generation for production parity.The synthetic seqval fallback can hide behavior when seqval is missing. For this integration flow, using the raw CDC value keeps test semantics aligned with runtime behavior.
Proposed adjustment
- // Generate ID from LSN+SeqVal+Op (native CDC unique key) - var seqval []byte - if len(c.SeqVal) > 0 { - seqval = c.SeqVal - } else { - seqval = []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, byte(i + 1)} - } coreChanges[i] = core.Change{ Table: c.Table, TransactionID: c.TransactionID, LSN: c.LSN, Operation: core.Operation(c.Operation), Data: c.Data, CommitTime: c.CommitTime, - ID: cdc.ComputeNativeID(c.LSN, seqval, c.Operation), + ID: cdc.ComputeNativeID(c.LSN, c.SeqVal, c.Operation), }Also applies to: 219-219
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/integration_test.go` around lines 206 - 211, The test currently fabricates a synthetic seqval when c.SeqVal is empty; change the integration ID generation to use the raw CDC value directly by assigning seqval = c.SeqVal (no synthetic fallback), so tests mirror production behavior; update the other occurrence (the similar assignment near the block at the second location) to the same pattern and ensure any downstream code that expects a non-nil slice handles an empty c.SeqVal appropriately.internal/store/sqlite/store_test.go (1)
557-558: ⚡ Quick winTest name claims log verification but no log assertion exists.
TestStore_Write_DuplicateID_IsIgnoredAndLoggedcurrently verifies dedup behavior only. Either add log capture/assertion or rename to avoid implying log coverage.Minimal rename option
-// TestStore_Write_DuplicateID_IsIgnoredAndLogged verifies duplicate IDs are deduped -func TestStore_Write_DuplicateID_IsIgnoredAndLogged(t *testing.T) { +// TestStore_Write_DuplicateID_IsIgnored verifies duplicate IDs are deduped +func TestStore_Write_DuplicateID_IsIgnored(t *testing.T) {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/store/sqlite/store_test.go` around lines 557 - 558, The test named TestStore_Write_DuplicateID_IsIgnoredAndLogged misleads: it asserts deduplication but does not capture or assert any logs; update the test to either (A) capture the logger output and add assertions that the expected "duplicate" log message is emitted (instrument the same logger used by the store in the test and assert on its entries) or (B) rename the test to TestStore_Write_DuplicateID_IsIgnored to accurately reflect behavior; modify the test function name (and any references) or add log-capture/assertion code targeting TestStore_Write_DuplicateID_IsIgnoredAndLogged as appropriate.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/cdc/query.go`:
- Around line 218-220: GetChanges currently logs and continues when s.Value()
for __$seqval fails, which allows an empty SeqVal and produces empty native IDs;
change the behavior in the block where s.Value() is called so that instead of
slog.Warn(...) you return a wrapped error (use fmt.Errorf("GetChanges: failed to
extract __$seqval: %w", err)) from GetChanges to fail fast and let the poll
cycle retry, ensuring SeqVal and native ID correctness.
---
Nitpick comments:
In `@internal/store/sqlite/store_test.go`:
- Around line 557-558: The test named
TestStore_Write_DuplicateID_IsIgnoredAndLogged misleads: it asserts
deduplication but does not capture or assert any logs; update the test to either
(A) capture the logger output and add assertions that the expected "duplicate"
log message is emitted (instrument the same logger used by the store in the test
and assert on its entries) or (B) rename the test to
TestStore_Write_DuplicateID_IsIgnored to accurately reflect behavior; modify the
test function name (and any references) or add log-capture/assertion code
targeting TestStore_Write_DuplicateID_IsIgnoredAndLogged as appropriate.
In `@tests/integration/integration_test.go`:
- Around line 206-211: The test currently fabricates a synthetic seqval when
c.SeqVal is empty; change the integration ID generation to use the raw CDC value
directly by assigning seqval = c.SeqVal (no synthetic fallback), so tests mirror
production behavior; update the other occurrence (the similar assignment near
the block at the second location) to the same pattern and ensure any downstream
code that expects a non-nil slice handles an empty c.SeqVal appropriately.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 8eb81a68-3b5e-415a-b6a0-621d8affd2df
📒 Files selected for processing (5)
internal/cdc/capturer.gointernal/cdc/capturer_test.gointernal/cdc/query.gointernal/store/sqlite/store_test.gotests/integration/integration_test.go
Comment on lines
+218
to
+220
| if val, err := s.Value(); err != nil { | ||
| slog.Warn("GetChanges: failed to extract __$seqval", "error", err) | ||
| } else if val != nil { |
There was a problem hiding this comment.
Fail fast on __$seqval extraction errors to protect native-ID correctness.
On Line 218, the code logs and continues when seqval extraction fails. That lets rows proceed with empty SeqVal, which later produces an empty native ID and weakens dedup guarantees. Return a wrapped error here so the poll cycle retries safely.
Proposed fix
- if val, err := s.Value(); err != nil {
- slog.Warn("GetChanges: failed to extract __$seqval", "error", err)
+ if val, err := s.Value(); err != nil {
+ slog.Warn("GetChanges: failed to extract __$seqval", "capture_instance", captureInstance, "error", err)
+ return nil, fmt.Errorf("extract __$seqval for %s: %w", captureInstance, err)
} else if val != nil {As per coding guidelines, "Errors should be propagated upward and not silently ignored" and "Use fmt.Errorf with %w verb to wrap errors with context instead of losing the original error".
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if val, err := s.Value(); err != nil { | |
| slog.Warn("GetChanges: failed to extract __$seqval", "error", err) | |
| } else if val != nil { | |
| if val, err := s.Value(); err != nil { | |
| slog.Warn("GetChanges: failed to extract __$seqval", "capture_instance", captureInstance, "error", err) | |
| return nil, fmt.Errorf("extract __$seqval for %s: %w", captureInstance, err) | |
| } else if val != nil { |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@internal/cdc/query.go` around lines 218 - 220, GetChanges currently logs and
continues when s.Value() for __$seqval fails, which allows an empty SeqVal and
produces empty native IDs; change the behavior in the block where s.Value() is
called so that instead of slog.Warn(...) you return a wrapped error (use
fmt.Errorf("GetChanges: failed to extract __$seqval: %w", err)) from GetChanges
to fail fast and let the poll cycle retry, ensuring SeqVal and native ID
correctness.
…nge ID - Remove SeqVal from Change struct and CDC query - Remove ComputeNativeID (replaced by inline hex encoding) - New ID format: hex(LSN):table:pk_values:op (no seqval dependency) - net_changes doesn't return seqval anyway, so seqval was always empty - ID now uses table+pkeys for row uniqueness within a transaction
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
internal/cdc/capturer.go (1)
374-384:⚠️ Potential issue | 🟠 Major | ⚡ Quick winGuard
changeIDconstruction whenlsnis empty to avoid malformed dedup keys.On Line 374, an empty
lsnproduces a non-empty but malformed ID (e.g.,:table:...), which can silently degrade dedup behavior.Suggested fix
+ if len(lsn) == 0 { + slog.Warn("convertToCoreChanges: missing LSN, skipping change", + "table", table, + "transaction_id", txID, + "operation", opVal, + ) + continue + } changeID := hex.EncodeToString(lsn) + ":" + table + ":" + tableKeysStr + ":" + strconv.Itoa(opVal)As per coding guidelines, "Errors should be propagated upward and not silently ignored".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/cdc/capturer.go` around lines 374 - 384, The changeID construction uses hex.EncodeToString(lsn) unguarded so an empty lsn yields a malformed ID; update the code around changeID (the hex.EncodeToString(lsn) + ":" + table + ":" + tableKeysStr + ":" + strconv.Itoa(opVal) construction in the function that appends to changes) to check for an empty or nil lsn before building changeID, return or propagate an error (do not silently continue) when lsn is missing, and only append the core.Change when a valid lsn is present so dedup keys remain well-formed; ensure the error is returned to the caller rather than swallowed.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tests/integration/integration_test.go`:
- Around line 207-208: The test builds changeID as
hex.EncodeToString(c.LSN)+":"+c.Table+":"+strconv.Itoa(int(c.Operation)) which
doesn't match runtime core.Change.ID format; update the construction of changeID
in tests/integration/integration_test.go to include table keys so it becomes
hex.EncodeToString(c.LSN) + ":" + c.Table + ":" + <tableKeysField> + ":" +
strconv.Itoa(int(c.Operation)) (use the actual field on the mock change, e.g.,
c.TableKeys or c.Keys, while keeping hex.EncodeToString(c.LSN), c.Table and
c.Operation unchanged) so the test ID shape matches production.
---
Outside diff comments:
In `@internal/cdc/capturer.go`:
- Around line 374-384: The changeID construction uses hex.EncodeToString(lsn)
unguarded so an empty lsn yields a malformed ID; update the code around changeID
(the hex.EncodeToString(lsn) + ":" + table + ":" + tableKeysStr + ":" +
strconv.Itoa(opVal) construction in the function that appends to changes) to
check for an empty or nil lsn before building changeID, return or propagate an
error (do not silently continue) when lsn is missing, and only append the
core.Change when a valid lsn is present so dedup keys remain well-formed; ensure
the error is returned to the caller rather than swallowed.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: b2002fe2-4f5b-4423-8edf-134968ab0110
📒 Files selected for processing (7)
internal/cdc/capturer.gointernal/cdc/capturer_test.gointernal/cdc/query.gointernal/core/transaction.gointernal/store/sqlite/store.gointernal/store/sqlite/store_test.gotests/integration/integration_test.go
💤 Files with no reviewable changes (1)
- internal/cdc/capturer_test.go
✅ Files skipped from review due to trivial changes (1)
- internal/store/sqlite/store.go
🚧 Files skipped from review as they are similar to previous changes (1)
- internal/store/sqlite/store_test.go
Comment on lines
+207
to
+208
| // ID format: hex(LSN):table:op (no PK available in mock) | ||
| changeID := hex.EncodeToString(c.LSN) + ":" + c.Table + ":" + strconv.Itoa(int(c.Operation)) |
There was a problem hiding this comment.
Align test ID format with runtime core.Change.ID composition.
Line 208 builds hex(LSN):table:op, but runtime currently composes hex(LSN):table:tableKeys:op. This test won’t validate the same ID shape used in production.
Suggested fix
- changeID := hex.EncodeToString(c.LSN) + ":" + c.Table + ":" + strconv.Itoa(int(c.Operation))
+ // Keep tableKeys slot for parity with capturer format (empty in this mock path)
+ changeID := hex.EncodeToString(c.LSN) + ":" + c.Table + "::" + strconv.Itoa(int(c.Operation))📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // ID format: hex(LSN):table:op (no PK available in mock) | |
| changeID := hex.EncodeToString(c.LSN) + ":" + c.Table + ":" + strconv.Itoa(int(c.Operation)) | |
| // ID format: hex(LSN):table:op (no PK available in mock) | |
| // Keep tableKeys slot for parity with capturer format (empty in this mock path) | |
| changeID := hex.EncodeToString(c.LSN) + ":" + c.Table + "::" + strconv.Itoa(int(c.Operation)) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/integration/integration_test.go` around lines 207 - 208, The test
builds changeID as
hex.EncodeToString(c.LSN)+":"+c.Table+":"+strconv.Itoa(int(c.Operation)) which
doesn't match runtime core.Change.ID format; update the construction of changeID
in tests/integration/integration_test.go to include table keys so it becomes
hex.EncodeToString(c.LSN) + ":" + c.Table + ":" + <tableKeysField> + ":" +
strconv.Itoa(int(c.Operation)) (use the actual field on the mock change, e.g.,
c.TableKeys or c.Keys, while keeping hex.EncodeToString(c.LSN), c.Table and
c.Operation unchanged) so the test ID shape matches production.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 7 files (changes from recent commits).
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="internal/cdc/capturer.go">
<violation number="1" location="internal/cdc/capturer.go:374">
P1: ID generation can collide when `tableKeysStr` is empty, causing valid CDC rows to be dropped by `INSERT OR IGNORE` dedup.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
| if tableKeysStr != "" { | ||
| id = hex.EncodeToString(lsn) + ":" + table + ":" + tableKeysStr + ":" + strconv.Itoa(opVal) | ||
| } | ||
| changeID := hex.EncodeToString(lsn) + ":" + table + ":" + tableKeysStr + ":" + strconv.Itoa(opVal) |
There was a problem hiding this comment.
P1: ID generation can collide when tableKeysStr is empty, causing valid CDC rows to be dropped by INSERT OR IGNORE dedup.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At internal/cdc/capturer.go, line 374:
<comment>ID generation can collide when `tableKeysStr` is empty, causing valid CDC rows to be dropped by `INSERT OR IGNORE` dedup.</comment>
<file context>
@@ -377,14 +371,16 @@ func (c *ChangeCapturer) convertToCoreChanges(captureChanges []core.CaptureChang
}
}
+ changeID := hex.EncodeToString(lsn) + ":" + table + ":" + tableKeysStr + ":" + strconv.Itoa(opVal)
+
changes = append(changes, core.Change{
</file context>
Suggested change
| changeID := hex.EncodeToString(lsn) + ":" + table + ":" + tableKeysStr + ":" + strconv.Itoa(opVal) | |
| changeID := hex.EncodeToString(lsn) + ":" + table + ":" + tableKeysStr + ":" + strconv.Itoa(opVal) | |
| if tableKeysStr == "" { | |
| changeID = core.ComputeChangeID(txID, table, data, lsn, core.Operation(opVal)) | |
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
ComputeNativeID: return empty string when LSN or SeqVal is nil/empty, with warn logGetChanges: log warning on seqval extraction failure instead of silent ignoreComputeNativeIDinstead of manual hex formattingChanges
internal/cdc/capturer.gointernal/cdc/capturer_test.gointernal/cdc/query.gointernal/store/sqlite/store_test.gotests/integration/integration_test.go