feat(snapshot): four-phase timing (A/B/C/D) per table - #255
Merged
cnlangzi merged 5 commits intoMay 13, 2026
Conversation
- CaptureResult gains TableDone and Table fields to signal table completion - SnapshotCapturer.Fetch tracks Phase A (ReadMs) and Phase B (BuildMs) per batch - Runtime.Run tracks Phase C (TransformMs) and Phase D (WriteMs) per batch - FinalizeTable called from Runtime when TableDone==true to record C+D timing - TableProgress now includes ReadMs, BuildMs, TransformMs, WriteMs for bottleneck identification - SnapshotFinalizer interface allows Runtime to signal timing back to SnapshotCapturer Fixes #254
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThis PR instruments per-table timing for four snapshot phases: Read, Build, Transform, and Write. core.CaptureResult gains a Table field; Capturer gains RecordTiming(table, transformMs, writeMs). Runtime.Run measures per-batch Transform and SinkWrite durations and calls RecordTiming(result.Table, transformMs, writeMs). SnapshotCapturer records Read/Build timings in Fetch, exposes RecordTiming to accumulate Transform/Write, initializes/clears per-table accumulators on Restart, and includes all four timing fields in Progress(). The snapshot status API payload was updated to report row-based totals/read counts. 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 |
![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 found 1 issue, and left some high level feedback:
- The per-table TransformMs/WriteMs currently only record the timings from the final batch (FinalizeTable is called once with the last batch’s measurements), so if the intent is to capture total per-table C/D time you likely want to accumulate these across all batches for a table instead of overwriting with the last batch.
- In Runtime.Run you track batchTransformMs/batchWriteMs even when retries occur; if you intend to measure only plugin execution and not retry backoff/wait time, consider moving the timers inside the retry callback around the actual Transform/SinkWrite calls.
- FinalizeTable identifies tables by string equality on FullName; if tables can be renamed or represented in different canonical forms, it may be more robust to pass and track a stable table identifier (e.g., index or struct reference) instead of a string lookup on every finalization.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The per-table TransformMs/WriteMs currently only record the timings from the final batch (FinalizeTable is called once with the last batch’s measurements), so if the intent is to capture total per-table C/D time you likely want to accumulate these across all batches for a table instead of overwriting with the last batch.
- In Runtime.Run you track batchTransformMs/batchWriteMs even when retries occur; if you intend to measure only plugin execution and not retry backoff/wait time, consider moving the timers inside the retry callback around the actual Transform/SinkWrite calls.
- FinalizeTable identifies tables by string equality on FullName; if tables can be renamed or represented in different canonical forms, it may be more robust to pass and track a stable table identifier (e.g., index or struct reference) instead of a string lookup on every finalization.
## Individual Comments
### Comment 1
<location path="internal/core/runtime.go" line_range="122-123" />
<code_context>
// Convert capture.Change to core.Change
changes := r.convertChanges(result.Changes)
+ // Track per-batch timing for C (Transform) and D (SinkWrite)
+ var batchTransformMs, batchWriteMs int64
+
// Transform phase with retry (independent timing measurement)
</code_context>
<issue_to_address>
**issue (bug_risk):** Per-table C/D timing currently only captures the last batch, not the full table time.
Phases A/B correctly accumulate per-table time in `SnapshotCapturer.Fetch` (`+= fetchMs` / `+= buildMs`), but for C/D you only measure a single batch (`batchTransformMs` / `batchWriteMs`) and then overwrite `tableTransformMs[i]` / `tableWriteMs[i]` in `FinalizeTable`. Given the field comments (“time spent in Transform per table” / “time spent in SinkWrite per table”), this is misleading: it reports only the last batch, not total per-table time. Either accumulate C/D timings across batches (e.g., maintain per-table running totals or make `FinalizeTable` add instead of overwrite), or rename/reword the fields/comments if you only intend to record the final batch’s timings.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
- FinalizeTable now += accumulates rather than = overwrites - Runtime.Run calls FinalizeTable on every batch (not just TableDone) - This ensures C+D timing matches A+B behavior (all batches accumulate) - Add trailing newline to capturer.go for Go style conformance
![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 3 files
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/snapshot/capturer.go">
<violation number="1" location="internal/snapshot/capturer.go:293">
P1: Phase C/D table timing is underreported because only the last batch duration is persisted when finalizing a table.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
- Capturer interface now includes RecordTiming(table, transformMs, writeMs) - SnapshotCapturer implements RecordTiming (accumulates per batch) - CDC/Replay capturers implement RecordTiming as no-op - Removed standalone SnapshotFinalizer interface - Runtime.Run calls capturer.RecordTiming directly on every batch Simplifies the design by keeping all capturer operations in one interface.
![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 (1)
internal/core/capturer.go (1)
25-26: ⚡ Quick winClarify
CaptureResult.Tablesemantics to match runtime behavior.Line 26 says
Tableis only for the last batch, but runtime usesresult.Tablefor per-batch timing attribution. Please align the comment/contract (e.g., “current batch table;TableDonemarks final batch”) to avoid future regressions.🤖 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/core/capturer.go` around lines 25 - 26, Update the comment on the CaptureResult struct to reflect actual runtime semantics: change the Table field comment to say it represents the table for the current batch (not only the last batch) and clarify that TableDone is a boolean indicating that this batch is the final batch for that table; ensure the comment around CaptureResult.Table and CaptureResult.TableDone (symbols: CaptureResult.Table, CaptureResult.TableDone) clearly states "current batch table; TableDone marks final batch" so future readers align with runtime usage.
🤖 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/core/runtime.go`:
- Around line 154-155: Timing is currently dropped when result.Table is empty
because capturer.RecordTiming(result.Table, batchTransformMs, batchWriteMs) uses
only result.Table; change this to use a stable table attribution by falling back
to the last-seen table when result.Table == "" (e.g., maintain a local lastTable
variable in the surrounding scope or add a field on the capturer to store
lastTable), update lastTable whenever result.Table is non-empty, and call
capturer.RecordTiming(effectiveTable, batchTransformMs, batchWriteMs) where
effectiveTable = result.Table != "" ? result.Table : lastTable so intermediate
C/D timing is accumulated per-table.
---
Nitpick comments:
In `@internal/core/capturer.go`:
- Around line 25-26: Update the comment on the CaptureResult struct to reflect
actual runtime semantics: change the Table field comment to say it represents
the table for the current batch (not only the last batch) and clarify that
TableDone is a boolean indicating that this batch is the final batch for that
table; ensure the comment around CaptureResult.Table and CaptureResult.TableDone
(symbols: CaptureResult.Table, CaptureResult.TableDone) clearly states "current
batch table; TableDone marks final batch" so future readers align with runtime
usage.
🪄 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: de47391b-9ee4-4604-bd86-1749fc875e85
📒 Files selected for processing (5)
internal/cdc/capturer.gointernal/core/capturer.gointernal/core/runtime.gointernal/replay/capturer.gointernal/snapshot/capturer.go
✅ Files skipped from review due to trivial changes (1)
- internal/replay/capturer.go
🚧 Files skipped from review as they are similar to previous changes (1)
- internal/snapshot/capturer.go
Comment on lines
+154
to
+155
| // Record timing for every batch (accumulates C+D timing per table) | ||
| capturer.RecordTiming(result.Table, batchTransformMs, batchWriteMs) |
There was a problem hiding this comment.
Prevent dropping C/D timing when result.Table is empty.
At Line 155, timing is recorded every batch, but table attribution relies solely on result.Table. If that field is only populated on table-final batches, intermediate C/D timings will be lost from per-table totals.
Suggested fix
- // Record timing for every batch (accumulates C+D timing per table)
- capturer.RecordTiming(result.Table, batchTransformMs, batchWriteMs)
+ // Record timing for every batch (accumulates C+D timing per table)
+ tableForTiming := result.Table
+ if tableForTiming == "" && len(changes) > 0 {
+ tableForTiming = changes[0].Table
+ }
+ if tableForTiming != "" {
+ capturer.RecordTiming(tableForTiming, batchTransformMs, batchWriteMs)
+ }📝 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
| // Record timing for every batch (accumulates C+D timing per table) | |
| capturer.RecordTiming(result.Table, batchTransformMs, batchWriteMs) | |
| // Record timing for every batch (accumulates C+D timing per table) | |
| tableForTiming := result.Table | |
| if tableForTiming == "" && len(changes) > 0 { | |
| tableForTiming = changes[0].Table | |
| } | |
| if tableForTiming != "" { | |
| capturer.RecordTiming(tableForTiming, batchTransformMs, batchWriteMs) | |
| } |
🤖 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/core/runtime.go` around lines 154 - 155, Timing is currently dropped
when result.Table is empty because capturer.RecordTiming(result.Table,
batchTransformMs, batchWriteMs) uses only result.Table; change this to use a
stable table attribution by falling back to the last-seen table when
result.Table == "" (e.g., maintain a local lastTable variable in the surrounding
scope or add a field on the capturer to store lastTable), update lastTable
whenever result.Table is non-empty, and call
capturer.RecordTiming(effectiveTable, batchTransformMs, batchWriteMs) where
effectiveTable = result.Table != "" ? result.Table : lastTable so intermediate
C/D timing is accumulated per-table.
There was a problem hiding this comment.
1 issue found across 5 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/core/runtime.go">
<violation number="1" location="internal/core/runtime.go:155">
P2: Per-batch C/D timing is not actually accumulated because `result.Table` is empty on most snapshot batches, so `RecordTiming` skips those batches.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
| } | ||
|
|
||
| // Record timing for every batch (accumulates C+D timing per table) | ||
| capturer.RecordTiming(result.Table, batchTransformMs, batchWriteMs) |
There was a problem hiding this comment.
P2: Per-batch C/D timing is not actually accumulated because result.Table is empty on most snapshot batches, so RecordTiming skips those batches.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At internal/core/runtime.go, line 155:
<comment>Per-batch C/D timing is not actually accumulated because `result.Table` is empty on most snapshot batches, so `RecordTiming` skips those batches.</comment>
<file context>
@@ -151,12 +151,8 @@ func (r *Runtime) Run(ctx context.Context) error {
- }
- }
+ // Record timing for every batch (accumulates C+D timing per table)
+ capturer.RecordTiming(result.Table, batchTransformMs, batchWriteMs)
// Write batch log (after transformer, regardless of capturer type)
</file context>
Tip: Review your code locally with the cubic CLI to iterate faster.
added 2 commits
May 12, 2026 21:52
RecordTiming is now called every batch. SnapshotCapturer tracks table completion internally and accumulates timing per table.
UI no longer needs done/processed counts - real-time timing fields (ReadMs, BuildMs, TransformMs, WriteMs) are sufficient for monitoring.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
internal/snapshot/capturer.go (2)
415-429:⚠️ Potential issue | 🔴 Critical | ⚡ Quick winAvoid re-locking
c.muon the query error path.Line 426 calls
markError, but this block already holdsc.mufrom Line 415. BecausemarkErrorlocks the same mutex again, the first query failure deadlocksFetch.Suggested fix
c.mu.Lock() if c.stopped { c.mu.Unlock() if rows != nil { _ = rows.Close() } return &core.CaptureResult{NextCapturer: core.CapturerCDC} } if queryErr != nil { slog.Error("SnapshotCapturer: query failed", "table", c.currentTable, "offset", offset, "error", queryErr) - c.markError(fmt.Sprintf("table %s at offset %d: %v", c.currentTable, offset, queryErr)) + c.lastError = fmt.Sprintf("table %s at offset %d: %v", c.currentTable, offset, queryErr) + c.pending = false c.tableIndex++ c.rowOffset = 0 c.mu.Unlock() return &core.CaptureResult{NextCapturer: core.CapturerSnapshot} }🤖 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/snapshot/capturer.go` around lines 415 - 429, The query-error branch in Fetch currently holds c.mu and then calls c.markError which also attempts to lock c.mu, causing a deadlock; to fix, release c.mu before calling c.markError (or add a markErrorNoLock helper and call that while still holding the lock), ensuring you still update c.tableIndex and c.rowOffset under the lock if needed and close rows if non-nil; reference c.mu, Fetch, c.markError, c.tableIndex and c.rowOffset when making the change.
472-498:⚠️ Potential issue | 🟠 Major | ⚡ Quick winSet
TableDoneon the terminal batch.This result carries
Table, but never marks the last batch as complete. That breaks the Runtime-side table-finalization contract described in the PR, so per-table phase C/D aggregation won't reliably fire when a table finishes.Suggested fix
result := &core.CaptureResult{ Changes: changes, BatchID: batchCtx.BatchID, NextCapturer: core.CapturerSnapshot, Table: table.FullName(), + TableDone: isLastBatch, } return resultIf empty tables also need finalization, return a zero-change completion result for the
len(batchRows) == 0case instead of exiting without any table-complete signal.🤖 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/snapshot/capturer.go` around lines 472 - 498, The terminal snapshot batch must mark table completion: when isLastBatch is true set the returned core.CaptureResult.TableDone = true (for the result built in the non-empty branch where changes are created) and for the early-return case when len(batchRows) == 0 return a zero-change core.CaptureResult that includes Table: table.FullName(), NextCapturer: core.CapturerSnapshot, BatchID (use core.NewBatchContext()), and TableDone = true so empty tables also emit the table-complete signal; update the code paths around SnapshotCapturer, the len(batchRows) == 0 return, and the result construction to include TableDone accordingly.
🤖 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.
Outside diff comments:
In `@internal/snapshot/capturer.go`:
- Around line 415-429: The query-error branch in Fetch currently holds c.mu and
then calls c.markError which also attempts to lock c.mu, causing a deadlock; to
fix, release c.mu before calling c.markError (or add a markErrorNoLock helper
and call that while still holding the lock), ensuring you still update
c.tableIndex and c.rowOffset under the lock if needed and close rows if non-nil;
reference c.mu, Fetch, c.markError, c.tableIndex and c.rowOffset when making the
change.
- Around line 472-498: The terminal snapshot batch must mark table completion:
when isLastBatch is true set the returned core.CaptureResult.TableDone = true
(for the result built in the non-empty branch where changes are created) and for
the early-return case when len(batchRows) == 0 return a zero-change
core.CaptureResult that includes Table: table.FullName(), NextCapturer:
core.CapturerSnapshot, BatchID (use core.NewBatchContext()), and TableDone =
true so empty tables also emit the table-complete signal; update the code paths
around SnapshotCapturer, the len(batchRows) == 0 return, and the result
construction to include TableDone accordingly.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: e63c24bf-7a81-4b06-a7d8-3578c6a5d7bb
📒 Files selected for processing (2)
cmd/app/server.gointernal/snapshot/capturer.go
💤 Files with no reviewable changes (1)
- cmd/app/server.go
cnlangzi
deleted the
fix/snapshot-performance-monitoring-four-phase-timing
branch
May 13, 2026 02:03
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
1 participant
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
Implements per-table, per-phase timing (A/B/C/D) for snapshot processing to identify bottlenecks.
Changes
Four Phases
Verification
Fixes #254