Correctness and deploy-safety fixes for the database write path - #709
Correctness and deploy-safety fixes for the database write path#709aditya1702 wants to merge 8 commits into
Conversation
22aeaaa to
d8e9f71
Compare
d8e9f71 to
505b369
Compare
505b369 to
1e837af
Compare
There was a problem hiding this comment.
Pull request overview
Improves correctness, resilience, and maintainability of the database persistence pipeline.
Changes:
- Preserves protocol classification across batches and rejects ledger zero.
- Makes commit barriers cancellation-safe and improves persistence metrics.
- Centralizes bulk-table and account-link COPY logic.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
internal/services/protocol_processor.go |
Clarifies protocol history transaction semantics. |
internal/services/ingest.go |
Tracks contract-to-WASM bindings. |
internal/services/ingest_test.go |
Adds persistence and classification regressions. |
internal/services/ingest_live.go |
Updates batching, commits, classification, and persistence. |
internal/services/ingest_backfill.go |
Reuses the authoritative bulk-table list. |
internal/metrics/ingestion.go |
Documents batch-level timing semantics. |
internal/ingest/timescaledb.go |
Derives hypertables from bulk-copy tables. |
internal/ingest/ingest.go |
Documents shutdown behavior. |
internal/data/transactions.go |
Uses the shared account-link COPY helper. |
internal/data/operations.go |
Uses the shared account-link COPY helper. |
internal/data/ingest_store.go |
Centralizes reconciliation table metadata. |
internal/data/ingest_store_test.go |
Reuses authoritative table metadata in tests. |
internal/data/accounts.go |
Adds the shared account-link COPY helper. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| // commit round-trip per sibling, the cursor still commits last, and | ||
| // re-reads converge. Hiding it would mean bounding every read by the | ||
| // committed cursor across the whole API surface. | ||
| commitCtx := context.WithoutCancel(ctx) |
| if len(m.classifiedWasms)+len(m.classifiedContracts) > maxClassificationSeenEntries { | ||
| clear(m.classifiedWasms) | ||
| clear(m.classifiedContracts) |
| for _, target := range data.BulkCopyTables { | ||
| assert.True(t, streamed[target.Table], "no persist sibling streams %s", target.Table) | ||
| } |
1e837af to
9c5d1e3
Compare
9c5d1e3 to
7d5c0b7
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated 2 comments.
Suppressed comments (2)
internal/services/ingest_test.go:1634
- This assertion checks only
BulkCopyTables ⊆ siblings, which is the reverse of the failure described above it. Adding a new COPY sibling while omitting it fromBulkCopyTablesleaves every current assertion true, so the crash-recovery regression is not detected. Pin the complete ordered sibling set, including the two non-COPY families.
for _, sibling := range siblings {
streamed[sibling.name] = true
}
for _, target := range data.BulkCopyTables {
assert.True(t, streamed[target.Table], "no persist sibling streams %s", target.Table)
internal/services/ingest_live.go:355
context.WithoutCancelalso removes any deadline, so a stalled commit can now block shutdown indefinitely while holding all eight persist connections. The advisory-lock cleanup in this file pairsWithoutCancelwith a finite timeout for exactly this failure mode. Use a bounded detached context sized to complete the barrier within the deployment termination grace period.
commitCtx := context.WithoutCancel(ctx)
for i, s := range siblings {
if commitErr := siblingTxs[i].Commit(commitCtx); commitErr != nil {
| // Merging the inputs is what makes a batch safe without any cut. A contract | ||
| // bound to a wasm uploaded by an earlier ledger of the same batch would, per | ||
| // ledger, have to resolve that wasm's verdict from protocol_wasms — a pool | ||
| // read that cannot see rows the batch has not committed. Supplied together, | ||
| // prepareClassificationPlan classifies the wasm from its buffered bytecode |
| // Test_getEffectiveProtocolContracts_NilClassificationKeepsCommitted pins the | ||
| // mid-batch semantics: ledgers riding behind a batch head run with no | ||
| // classification plan (classification == nil), and their buffered contracts | ||
| // are pure re-observations — every binding was already seen committed, the | ||
| // batch cut guarantees it. Committed membership must stand untouched; | ||
| // dropping a re-observed contract here silently discards that ledger's | ||
| // events for it. | ||
| // Test_getEffectiveProtocolContracts_ReObservationKeepsMembership pins the case |
7d5c0b7 to
8cbd794
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.
Suppressed comments (3)
internal/services/ingest_test.go:2601
- This comment names a test that does not exist and describes the removed nil-plan/batch-cut design; the following test passes a populated classification map and exercises the new merged-plan behavior. Remove this stale block so the regression coverage is not misrepresented.
// Test_getEffectiveProtocolContracts_NilClassificationKeepsCommitted pins the
// mid-batch semantics: ledgers riding behind a batch head run with no
// classification plan (classification == nil), and their buffered contracts
// are pure re-observations — every binding was already seen committed, the
// batch cut guarantees it. Committed membership must stand untouched;
internal/services/ingest_live.go:969
- This “safe without any cut” claim contradicts the PR's binding-aware cut requirement and is not correct for a contract rebound later in the batch. Because the merged contract map keeps only the final binding, the plan does not resolve the earlier WASM; staging the earlier ledger then removes that contract from protocol membership and can silently drop its events. Restore the batch cut on binding changes, or retain every intermediate binding in the classification plan.
// Merging the inputs is what makes a batch safe without any cut. A contract
// bound to a wasm uploaded by an earlier ledger of the same batch would, per
// ledger, have to resolve that wasm's verdict from protocol_wasms — a pool
// read that cannot see rows the batch has not committed. Supplied together,
// prepareClassificationPlan classifies the wasm from its buffered bytecode
internal/services/ingest_test.go:1634
- This assertion is one-way: if a bulk sibling is accidentally removed from
BulkCopyTables, the loop no longer checks it and the test still passes—the exact orphan-reconciliation failure this test claims to prevent. Compare the complete sibling-name set against the registry plus the two known balance siblings.
for _, target := range data.BulkCopyTables {
assert.True(t, streamed[target.Table], "no persist sibling streams %s", target.Table)
| commitCtx := context.WithoutCancel(ctx) | ||
| for i, s := range siblings { | ||
| if commitErr := siblingTxs[i].Commit(ctx); commitErr != nil { | ||
| if commitErr := siblingTxs[i].Commit(commitCtx); commitErr != nil { |
8cbd794 to
36982f7
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated no new comments.
Suppressed comments (3)
internal/services/ingest_test.go:1634
- This only checks
BulkCopyTables → siblings. Adding a COPY sibling but forgetting to add it toBulkCopyTables—the orphan/crash-loop case this test claims to prevent—still passes. Compare the exact first bulk-sibling list and assert that only the two documented balance siblings remain.
for _, target := range data.BulkCopyTables {
assert.True(t, streamed[target.Table], "no persist sibling streams %s", target.Table)
internal/services/ingest_test.go:2601
- This stale preamble names a test that does not exist and describes nil-classification/batch-cut semantics that the following test does not exercise—the test passes a non-nil batch-wide match map. Remove it so the regression test documents its actual coverage.
// Test_getEffectiveProtocolContracts_NilClassificationKeepsCommitted pins the
// mid-batch semantics: ledgers riding behind a batch head run with no
// classification plan (classification == nil), and their buffered contracts
// are pure re-observations — every binding was already seen committed, the
// batch cut guarantees it. Committed membership must stand untouched;
internal/services/ingest_live.go:973
- “Later ledgers win” is not safe for processing the earlier ledgers with the same final map. If committed contract C (A/W1) emits in ledger N and rebinds to B/W2 in N+1, the merged plan contains only W2; while processing N,
getEffectiveProtocolContractscannot re-add C under W1 and silently drops its A event. Split the batch when a binding changes, as the PR description specifies, or retain ledger-specific classification maps.
// Later ledgers win the merge, which is the correct end state for the batch:
// a contract rebound mid-batch ends up bound to its final wasm. Each ledger
36982f7 to
0b7673e
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.
Suppressed comments (2)
internal/services/ingest_test.go:2601
- This first comment block is stale: batched ledgers now share one non-nil batch plan, and there is no test with the name it cites. Remove it so the test documents only the current batch-wide classification behavior.
// Test_getEffectiveProtocolContracts_NilClassificationKeepsCommitted pins the
// mid-batch semantics: ledgers riding behind a batch head run with no
// classification plan (classification == nil), and their buffered contracts
// are pure re-observations — every binding was already seen committed, the
// batch cut guarantees it. Committed membership must stand untouched;
internal/services/ingest_live.go:962
- The new claim that every batch is safe without a cut contradicts the PR description's requirement to start a new batch when a contract switches wasm. The current last-binding-wins merge also loses an intermediate binding: if ledger 100 rebinds C to known W1 and ledger 101 rebinds it to W2, only W2 enters
plan.Matches, so staging ledger 100 can silently omit C's protocol events/state. Preserve every distinct binding needed by per-ledger staging or implement the described binding-aware batch cut, and change theLastBindingWinstest accordingly.
// Merging the inputs is what makes a batch safe without any cut. A contract
// bound to a wasm uploaded by an earlier ledger of the same batch would, per
// ledger, have to resolve that wasm's verdict from protocol_wasms — a pool
// read that cannot see rows the batch has not committed. Supplied together,
// prepareClassificationPlan classifies the wasm from its buffered bytecode
| for _, target := range BulkCopyTables { | ||
| var atCursor, aboveCursor int | ||
| boundary := toid.New(int32(cursorLedger+1), 0, 0).ToInt64() | ||
| require.NoError(t, dbConnectionPool.QueryRow(ctx, | ||
| fmt.Sprintf(`SELECT count(*) FILTER (WHERE %[1]s < $1), count(*) FILTER (WHERE %[1]s >= $1) FROM %[2]s`, column, table), | ||
| fmt.Sprintf(`SELECT count(*) FILTER (WHERE %[1]s < $1), count(*) FILTER (WHERE %[1]s >= $1) FROM %[2]s`, target.TOIDColumn, target.Table), | ||
| boundary).Scan(&atCursor, &aboveCursor)) | ||
| counts[table] = [2]int{atCursor, aboveCursor} | ||
| counts[target.Table] = [2]int{atCursor, aboveCursor} |
0b7673e to
f3b8ab5
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated no new comments.
Suppressed comments (3)
internal/services/ingest_live.go:353
context.WithoutCancelalso removes the caller's deadline, so any lock or network stall duringCommitcan block shutdown indefinitely. This is especially relevant at a commit barrier containing deferred constraint checks. Detach from the pipeline cancellation but apply an independent, bounded commit deadline; retain the existing partial-persist handling if that deadline expires after a sibling commits.
commitCtx := context.WithoutCancel(ctx)
internal/services/ingest_test.go:2601
- This comment names a test that does not exist and describes obsolete nil-classification/batch-cut semantics; the implementation now builds one non-nil plan for the whole batch. Remove this stale paragraph so the regression test documents the behavior it actually exercises.
// Test_getEffectiveProtocolContracts_NilClassificationKeepsCommitted pins the
// mid-batch semantics: ledgers riding behind a batch head run with no
// classification plan (classification == nil), and their buffered contracts
// are pure re-observations — every binding was already seen committed, the
// batch cut guarantees it. Committed membership must stand untouched;
internal/services/ingest_live.go:957
- “Later ledgers win” is unsafe for protocol processing because the last-write-wins contract merge also discards classifications needed by earlier ledgers. If ledger N observes C bound to an already-committed W1 and N+1 rebinds C to W2, only W2 reaches the plan; while staging N,
getEffectiveProtocolContractsremoves C and cannot re-add it because W1 is absent fromMatches, silently dropping N's protocol events/state. Preserve every binding's classification or cut the batch before a binding change, and add an end-to-end regression for events before the rebind.
// Later ledgers win the merge, which is the correct end state for the batch:
// a contract rebound mid-batch ends up bound to its final wasm. Each ledger
Sibling and coordinator commits ran on the cancellable pipeline context, so a SIGTERM (or another stage's failure cancelling the errgroup) landing after the first sibling committed aborted the remaining commits and wrapped the routine shutdown in a fatal ErrPartialPersist — forcing the next start through DeleteRowsAboveLedger and firing the db_persist error counter for a healthy rolling restart. The window widens linearly with batch size. The barrier now runs on context.WithoutCancel: a batch commits fully or not at all. Cancellation before the barrier still rolls the whole batch back, and the cancellation-counts-against-no-counter contract in the retry ladder's error switch holds again. Also updates the stale root-context comment in ingest.go that still described the pre-pipeline single-ledger atomicity.
Three deploy-safety fixes for the persist pipeline's schema footprint: - Forward migration for the state_changes index swap. 2025-06-10.4 was edited in place to trade idx_state_changes_operation_id and idx_state_changes_account_category for idx_state_changes_account_id, but sql-migrate keys applied migrations by filename, so environments already past that file would keep the old shape forever while fresh builds got the new one. The new migration converges them with idempotent statements that no-op on fresh databases. - The liquidity_pool_balances FK now applies as NOT VALID + VALIDATE under notransaction, with an orphan-cleanup DELETE first: the plain ADD CONSTRAINT validated every row under a write-blocking lock and a single pre-existing orphan (a reachable state under the old read-time join) would abort the migration mid-deploy. A DROP IF EXISTS guard keeps the unwrapped migration re-runnable after a partial failure. - Live ingestion fails fast when the pool cannot fit the persist barrier: it pins 9 connections (7 siblings + coordinator + advisory lock), and pgxpool.Acquire has no timeout, so a deployment pinning DB_MAX_CONNS at the pre-pipeline value of 8 or below wedged silently — the last Acquire waited forever on connections its own goroutine held.
insert_into_db (and its 0.6s/1.0s grading buckets) now observes one true wall-time value per persist commit instead of len(batch) amortized shares. Batching engages exactly when persist has fallen behind, so the dilution reported the overloaded condition as healthy — a 4s commit for 8 ledgers landed as eight 0.5s observations. The per-ledger total duration keeps amortized shares so ledger durations stay comparable. The commit barrier also documents the deliberate torn-read window between sibling commits (readers can briefly see a transaction without its participant links or state changes; the cursor still commits last).
DeleteRowsAboveLedger falls back to deleting with no partition-column bound when the cursor ledger left no transactions row — correct but a full scan of every chunk of the five bulk tables. Log the degenerate path so an operator can attribute a slow start to it.
TransactionModel.BatchCopyAccounts and OperationModel.BatchCopyAccounts were the same function twice — same row shape, same zero-timestamp guard, same metrics — differing only in table name, ID column, and parent type. One generic helper now serves both; the wrappers keep their public signatures.
The five bulk-COPY tables were enumerated independently in startup reconciliation, the hypertable settings pass, the backfill recompressor, and a test's own copy — and a table missing from DeleteRowsAboveLedger leaves orphans that crash-loop re-ingest on PK collisions. data.BulkCopyTables (table + TOID column) is now the single definition all of them consume.
…sist entry The queue between the process and persist stages carried each ledger's full decoded LedgerCloseMeta and its materialized transactions, pinning them for the pipeline depth though persist needs only the close time and the ContractData memo — both now derived at the end of the process pass. The memo keeps its laziness: extraction still runs only when a CAS-winning processor requires contract data. persistLedgerData materializes the buffer's transaction and operation slices once per ledger (two siblings consume each), rejects ledger sequence 0 at entry (the CAS chain and cursor guard both address ledgerSeq-1, which underflows unsigned), and its sibling set moves to a named constructor pinned by a test to cover data.BulkCopyTables exactly.
Same facts, easier shape: each block leads with the rule, then the why — the state_changes single-transaction rule, the commit barrier's three properties, the seen-set marking contract and its enrichment trade, the shutdown behavior split by pipeline position. Also detaches the seen-set cap's comment from markClassificationInputsSeen's doc block, which had fused into one comment attached to the wrong declaration.
f3b8ab5 to
9f06de9
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated no new comments.
Suppressed comments (3)
internal/services/ingest_test.go:1634
- This assertion is one-way: it verifies that every listed table has a sibling, but not that every bulk-COPY sibling appears in
BulkCopyTables. Adding a new bulk sibling without updating reconciliation—the exact crash-recovery regression this test is meant to prevent—still passes. Pin the documented ordered bulk prefix and sibling count instead.
for _, sibling := range siblings {
streamed[sibling.name] = true
}
for _, target := range data.BulkCopyTables {
assert.True(t, streamed[target.Table], "no persist sibling streams %s", target.Table)
internal/services/ingest_live.go:355
- The cancellation-detached commit barrier is a core correctness change, but no test exercises cancellation after sibling commits begin. A regression that passes the canceled context to a later commit would recreate
ErrPartialPersistduring normal shutdown; add a commit-barrier test with controllable transactions that cancels between commits and verifies every commit receives a live context.
commitCtx := context.WithoutCancel(ctx)
for i, s := range siblings {
if commitErr := siblingTxs[i].Commit(commitCtx); commitErr != nil {
internal/services/ingest_live.go:957
- “Later ledgers win” is not safe for each ledger’s protocol processing. The merge is keyed by contract ID, so a later rebind removes an earlier binding from the shared plan; when that earlier WASM was already committed (not uploaded in this batch), its hash is absent from
Matches, and replaying the earlier ledger drops that contract’s protocol events/state. Preserve every binding needed by per-ledger overlays, or restore a batch cut when a binding changes.
// Later ledgers win the merge, which is the correct end state for the batch:
// a contract rebound mid-batch ends up bound to its final wasm. Each ledger
Start here: the
08993822commit, if you're reviewing for deploy safety — it's the only part that can break an existing environment.Ten commits finishing off the write path.
Correctness
ErrPartialPersist. A batch now commits fully or not at all.ledgerSeq-1, which wraps around on an unsigned int.Deploy safety and metrics
2025-06-10.4before we edited it, plusNOT VALID+VALIDATEfor the pool foreign key so validating it doesn't block writes.insert_into_dbrecords one real timing per commit instead of splitting it across the ledgers in the batch, which made an overloaded write step look healthy.Cleanup
One shared COPY helper for the two account link tables, one shared list of bulk-COPY tables (a table missing from that list leaves rows behind that crash re-ingestion), a lighter handoff between processing and writing, and a comment pass.
Merging this finishes the chain. The result is exactly the same code as the old #684.
PR 6 of 6 replacing #684 · schema → fixes → pipeline → parallel writes → table groups → final fixes