From 1c976ba35faae1349d63690a773238b3372e85ad Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Wed, 12 Aug 2026 23:37:29 -0700 Subject: [PATCH] fix(api): retain flushed new-chain queue rows instead of deleting them The flusher deleted each row after forwarding it. That is unrecoverable, and the genesis migration chain is regenerated before it ships -- the validator key is baked into every block header, so it has to be one the bootstrap node holds, which means rebuilding the chain. Anything already forwarded and deleted would survive only on the chain being discarded. Mark rows instead. A forwarded row gets flushed_at set and stays put, so the queue is a durable log: repoint the flusher at the rebuilt chain, set NewChainFlushFromBlock to the new backfill's end height, and re-drive. Replay is idempotent -- the same signed transactions are rejected as duplicates by a chain that already has them and accepted by a rebuilt one. The two non-forwarding paths are marked rather than deleted for the same reason, and record why in skip_reason: 'backfilled' for rows the genesis backfill covers, 'corrupt' for payloads that fail to unmarshal (previously deleted outright, taking the evidence with them). trimBackfillRows now only touches pending rows, so re-running it with a higher flush-from block after a regeneration does not disturb the record of what was already sent. Reads are unaffected in the steady state: the flusher only ever selects pending rows, and a partial index on (id) WHERE flushed_at IS NULL keeps retained rows off the hot path as the table grows. Retention is now an explicit operator decision rather than a side effect of flushing. Co-Authored-By: Claude Opus 5 --- api/new_chain_flusher.go | 53 ++++--- api/new_chain_flusher_test.go | 133 +++++++++++++++--- .../0238_new_chain_queue_cursor.sql | 22 +++ 3 files changed, 174 insertions(+), 34 deletions(-) create mode 100644 ddl/migrations/0238_new_chain_queue_cursor.sql diff --git a/api/new_chain_flusher.go b/api/new_chain_flusher.go index 37b1de77..1ecc1647 100644 --- a/api/new_chain_flusher.go +++ b/api/new_chain_flusher.go @@ -19,11 +19,18 @@ import ( "google.golang.org/protobuf/proto" ) -// NewChainFlusher reads rows from new_chain_queue and forwards them to the new -// Core chain. On startup it deletes rows covered by the backfill (confirmed_block -// < cfg.NewChainFlushFromBlock), then sends the rest in id order, one at a time. -// Sequential processing is required to preserve transaction ordering across users -// (dev apps can act on behalf of other users, creating cross-user dependencies). +// NewChainFlusher reads pending rows from new_chain_queue and forwards them to +// the new Core chain. On startup it marks rows covered by the backfill +// (confirmed_block < cfg.NewChainFlushFromBlock) as skipped, then sends the rest +// in id order, one at a time. Sequential processing is required to preserve +// transaction ordering across users (dev apps can act on behalf of other users, +// creating cross-user dependencies). +// +// Rows are never deleted — a forwarded row gets flushed_at set and stays put. +// The genesis migration chain is regenerated before it ships, so a row deleted +// on success would survive only on the chain being discarded. Keeping them makes +// the queue a durable log: point the flusher at the rebuilt chain, set +// NewChainFlushFromBlock to the new backfill's end height, and re-drive. type NewChainFlusher struct { cfg *config.Config writePool *pgxpool.Pool @@ -130,7 +137,9 @@ type queueRow struct { func (f *NewChainFlusher) fetchBatch(ctx context.Context, limit int) ([]queueRow, error) { rows, err := f.writePool.Query(ctx, - `SELECT id, tx_data FROM new_chain_queue ORDER BY id LIMIT $1`, + `SELECT id, tx_data FROM new_chain_queue + WHERE flushed_at IS NULL + ORDER BY id LIMIT $1`, limit, ) if err != nil { @@ -152,9 +161,12 @@ func (f *NewChainFlusher) fetchBatch(ctx context.Context, limit int) ([]queueRow func (f *NewChainFlusher) flushRow(ctx context.Context, row queueRow) error { var tx v1.ManageEntityLegacy if err := proto.Unmarshal(row.txRaw, &tx); err != nil { - // Corrupt row — delete it and move on rather than retrying forever. - f.logger.Error("corrupt queue row, deleting", zap.Int64("id", row.id), zap.Error(err)) - _, _ = f.writePool.Exec(ctx, `DELETE FROM new_chain_queue WHERE id = $1`, row.id) + // Corrupt row — mark it skipped and move on rather than retrying forever. + // It stays in the table so the bad payload can still be inspected. + f.logger.Error("corrupt queue row, skipping", zap.Int64("id", row.id), zap.Error(err)) + _, _ = f.writePool.Exec(ctx, + `UPDATE new_chain_queue SET flushed_at = now(), skip_reason = 'corrupt' WHERE id = $1`, + row.id) return nil } @@ -191,26 +203,35 @@ func (f *NewChainFlusher) flushRow(ctx context.Context, row queueRow) error { } } - _, err := f.writePool.Exec(ctx, `DELETE FROM new_chain_queue WHERE id = $1`, row.id) + _, err := f.writePool.Exec(ctx, + `UPDATE new_chain_queue SET flushed_at = now() WHERE id = $1`, + row.id) return err } -// trimBackfillRows deletes all queue rows whose confirmed_block is before the -// configured flush-from block, i.e. rows already covered by the genesis backfill. -// Rows with a NULL confirmed_block are kept: NULL < $1 evaluates to NULL (falsy) in SQL. +// trimBackfillRows marks every pending row whose confirmed_block is before the +// configured flush-from block as skipped, i.e. already covered by the genesis +// backfill. Rows with a NULL confirmed_block are left pending: NULL < $1 +// evaluates to NULL (falsy) in SQL. +// +// Only pending rows are touched, so re-running with a higher flush-from block +// after a chain regeneration marks the newly-covered rows without disturbing +// the record of what was already sent. func (f *NewChainFlusher) trimBackfillRows(ctx context.Context) error { if f.cfg.NewChainFlushFromBlock <= 0 { return nil } tag, err := f.writePool.Exec(ctx, - `DELETE FROM new_chain_queue WHERE confirmed_block < $1`, + `UPDATE new_chain_queue + SET flushed_at = now(), skip_reason = 'backfilled' + WHERE confirmed_block < $1 AND flushed_at IS NULL`, f.cfg.NewChainFlushFromBlock, ) if err != nil { return err } - f.logger.Info("trimmed backfill-covered rows", - zap.Int64("deleted", tag.RowsAffected()), + f.logger.Info("marked backfill-covered rows as skipped", + zap.Int64("skipped", tag.RowsAffected()), zap.Int64("flush_from_block", f.cfg.NewChainFlushFromBlock), ) return nil diff --git a/api/new_chain_flusher_test.go b/api/new_chain_flusher_test.go index 674f01ce..c2f68299 100644 --- a/api/new_chain_flusher_test.go +++ b/api/new_chain_flusher_test.go @@ -44,13 +44,16 @@ func newTestFlusher(t *testing.T, cfg *config.Config) (*NewChainFlusher, *mockCo pool := database.CreateTestDatabase(t, "test_api") - // Create the new_chain_queue table (not in template DB yet). + // Fallback for a template DB predating the new_chain_queue migrations; + // a current template already has the table and this is a no-op. _, err := pool.Exec(context.Background(), ` CREATE TABLE IF NOT EXISTS new_chain_queue ( id bigserial PRIMARY KEY, created_at timestamptz NOT NULL DEFAULT now(), tx_data bytea NOT NULL, - confirmed_block bigint + confirmed_block bigint, + flushed_at timestamptz, + skip_reason text ) `) require.NoError(t, err) @@ -85,7 +88,19 @@ func insertQueueRow(t *testing.T, f *NewChainFlusher, tx *corev1.ManageEntityLeg require.NoError(t, err) } -func queueDepth(t *testing.T, f *NewChainFlusher) int { +// pendingDepth counts rows the flusher still has to send. Rows are never +// deleted, so this is the count of unflushed rows rather than the table size. +func pendingDepth(t *testing.T, f *NewChainFlusher) int { + t.Helper() + var n int + err := f.writePool.QueryRow(context.Background(), + `SELECT count(*) FROM new_chain_queue WHERE flushed_at IS NULL`).Scan(&n) + require.NoError(t, err) + return n +} + +// totalDepth counts every row, flushed or not. +func totalDepth(t *testing.T, f *NewChainFlusher) int { t.Helper() var n int err := f.writePool.QueryRow(context.Background(), `SELECT count(*) FROM new_chain_queue`).Scan(&n) @@ -93,18 +108,39 @@ func queueDepth(t *testing.T, f *NewChainFlusher) int { return n } +// skipReasons returns the skip_reason of every non-pending row, in id order, +// with NULL (i.e. genuinely forwarded) rendered as "sent". +func skipReasons(t *testing.T, f *NewChainFlusher) []string { + t.Helper() + rows, err := f.writePool.Query(context.Background(), + `SELECT coalesce(skip_reason, 'sent') FROM new_chain_queue + WHERE flushed_at IS NOT NULL ORDER BY id`) + require.NoError(t, err) + defer rows.Close() + var out []string + for rows.Next() { + var r string + require.NoError(t, rows.Scan(&r)) + out = append(out, r) + } + require.NoError(t, rows.Err()) + return out +} + // TestEnqueueForNewChain verifies that enqueueForNewChain inserts a row with the // correct tx_data and confirmed_block. func TestEnqueueForNewChain(t *testing.T) { app := emptyTestApp(t) - // Add new_chain_queue to the test DB. + // Fallback for a template DB predating the new_chain_queue migrations. _, err := app.writePool.Exec(context.Background(), ` CREATE TABLE IF NOT EXISTS new_chain_queue ( id bigserial PRIMARY KEY, created_at timestamptz NOT NULL DEFAULT now(), tx_data bytea NOT NULL, - confirmed_block bigint + confirmed_block bigint, + flushed_at timestamptz, + skip_reason text ) `) require.NoError(t, err) @@ -134,7 +170,8 @@ func TestEnqueueForNewChain(t *testing.T) { } // TestNewChainFlusherTrim verifies that rows with confirmed_block < FlushFromBlock -// are deleted on startup, and rows at or above the threshold are kept. +// are marked skipped on startup rather than deleted, and rows at or above the +// threshold stay pending. func TestNewChainFlusherTrim(t *testing.T) { cfg := &config.Config{NewChainFlushFromBlock: 100} f, _ := newTestFlusher(t, cfg) @@ -147,18 +184,20 @@ func TestNewChainFlusherTrim(t *testing.T) { insertQueueRow(t, f, sampleTx(2), &block99) // should be trimmed insertQueueRow(t, f, sampleTx(3), &block100) // kept (boundary) insertQueueRow(t, f, sampleTx(4), &block200) // kept - insertQueueRow(t, f, sampleTx(5), nil) // NULL confirmed_block — kept + insertQueueRow(t, f, sampleTx(5), nil) // NULL confirmed_block — kept - require.Equal(t, 5, queueDepth(t, f)) + require.Equal(t, 5, pendingDepth(t, f)) err := f.trimBackfillRows(context.Background()) require.NoError(t, err) - require.Equal(t, 3, queueDepth(t, f)) + require.Equal(t, 3, pendingDepth(t, f), "two backfill-covered rows should no longer be pending") + require.Equal(t, 5, totalDepth(t, f), "trim must not delete rows") + require.Equal(t, []string{"backfilled", "backfilled"}, skipReasons(t, f)) - // Verify the surviving entity IDs. + // Verify the entity IDs still pending. rows, err := f.writePool.Query(context.Background(), - `SELECT (tx_data) FROM new_chain_queue ORDER BY id`, + `SELECT (tx_data) FROM new_chain_queue WHERE flushed_at IS NULL ORDER BY id`, ) require.NoError(t, err) defer rows.Close() @@ -175,7 +214,7 @@ func TestNewChainFlusherTrim(t *testing.T) { } // TestNewChainFlusherSends verifies that the flusher forwards all queued rows to -// the new chain and deletes them on success. +// the new chain and marks them flushed — retaining them — on success. func TestNewChainFlusherSends(t *testing.T) { cfg := &config.Config{} // no trim f, mock := newTestFlusher(t, cfg) @@ -187,7 +226,7 @@ func TestNewChainFlusherSends(t *testing.T) { insertQueueRow(t, f, sampleTx(2), &block20) insertQueueRow(t, f, sampleTx(3), &block30) - require.Equal(t, 3, queueDepth(t, f)) + require.Equal(t, 3, pendingDepth(t, f)) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -199,9 +238,13 @@ func TestNewChainFlusherSends(t *testing.T) { }, 5*time.Second, 50*time.Millisecond, "expected 3 ForwardTransaction calls") require.Eventually(t, func() bool { - return queueDepth(t, f) == 0 + return pendingDepth(t, f) == 0 }, 5*time.Second, 50*time.Millisecond, "expected queue to drain") + // Forwarded rows are retained so they can be re-driven onto a rebuilt chain. + require.Equal(t, 3, totalDepth(t, f), "flushed rows must not be deleted") + require.Equal(t, []string{"sent", "sent", "sent"}, skipReasons(t, f)) + // Verify all three entity IDs were forwarded. receivedIDs := make([]int64, len(mock.received)) for i, me := range mock.received { @@ -216,9 +259,9 @@ func TestNewChainFlusherTrimThenSend(t *testing.T) { cfg := &config.Config{NewChainFlushFromBlock: 50} f, mock := newTestFlusher(t, cfg) - block10 := int64(10) // pre-backfill — trimmed - block49 := int64(49) // pre-backfill — trimmed - block50 := int64(50) // post-backfill — flushed + block10 := int64(10) // pre-backfill — trimmed + block49 := int64(49) // pre-backfill — trimmed + block50 := int64(50) // post-backfill — flushed block100 := int64(100) // post-backfill — flushed insertQueueRow(t, f, sampleTx(1), &block10) insertQueueRow(t, f, sampleTx(2), &block49) @@ -235,12 +278,66 @@ func TestNewChainFlusherTrimThenSend(t *testing.T) { }, 5*time.Second, 50*time.Millisecond, "expected 2 ForwardTransaction calls (post-trim)") require.Eventually(t, func() bool { - return queueDepth(t, f) == 0 + return pendingDepth(t, f) == 0 }, 5*time.Second, 50*time.Millisecond, "expected queue to drain") + // Two skipped by the trim, two genuinely sent — all four retained. + require.Equal(t, 4, totalDepth(t, f)) + require.Equal(t, []string{"backfilled", "backfilled", "sent", "sent"}, skipReasons(t, f)) + receivedIDs := make([]int64, len(mock.received)) for i, me := range mock.received { receivedIDs[i] = me.EntityId } require.ElementsMatch(t, []int64{3, 4}, receivedIDs) } + +// TestNewChainFlusherRedriveAfterRegeneration is the reason rows are retained +// rather than deleted. The genesis chain is regenerated before it ships, so +// everything already forwarded has to be replayable onto the rebuilt chain. +// +// Simulates that: drain the queue against chain v1, then clear flushed_at (what +// an operator does when repointing at the rebuilt chain) and confirm the same +// transactions are forwarded again. With delete-on-success there would be +// nothing left to re-drive. +func TestNewChainFlusherRedriveAfterRegeneration(t *testing.T) { + cfg := &config.Config{} + f, mock := newTestFlusher(t, cfg) + + block10 := int64(10) + block20 := int64(20) + insertQueueRow(t, f, sampleTx(1), &block10) + insertQueueRow(t, f, sampleTx(2), &block20) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + go f.Start(ctx) + + require.Eventually(t, func() bool { + return mock.calls.Load() == 2 + }, 5*time.Second, 50*time.Millisecond, "expected both rows forwarded to chain v1") + require.Eventually(t, func() bool { + return pendingDepth(t, f) == 0 + }, 5*time.Second, 50*time.Millisecond) + cancel() + + // The rows are still here — that is the point. + require.Equal(t, 2, totalDepth(t, f)) + + // Repoint at the rebuilt chain: everything becomes pending again. + _, err := f.writePool.Exec(context.Background(), + `UPDATE new_chain_queue SET flushed_at = NULL, skip_reason = NULL`) + require.NoError(t, err) + require.Equal(t, 2, pendingDepth(t, f)) + + ctx2, cancel2 := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel2() + go f.Start(ctx2) + + require.Eventually(t, func() bool { + return mock.calls.Load() == 4 + }, 5*time.Second, 50*time.Millisecond, "expected both rows forwarded again after re-drive") + require.Eventually(t, func() bool { + return pendingDepth(t, f) == 0 + }, 5*time.Second, 50*time.Millisecond) +} diff --git a/ddl/migrations/0238_new_chain_queue_cursor.sql b/ddl/migrations/0238_new_chain_queue_cursor.sql new file mode 100644 index 00000000..b136c958 --- /dev/null +++ b/ddl/migrations/0238_new_chain_queue_cursor.sql @@ -0,0 +1,22 @@ +BEGIN; + +-- Mark queue rows as flushed instead of deleting them. +-- +-- The genesis migration chain gets regenerated before it ships (the validator +-- key baked into every block header has to be one the bootstrap node holds), so +-- any row deleted after a successful forward would survive only on a chain that +-- is about to be discarded. Retaining the rows makes new_chain_queue a durable +-- log that can be re-driven onto the rebuilt chain. +ALTER TABLE new_chain_queue ADD COLUMN IF NOT EXISTS flushed_at timestamptz; +ALTER TABLE new_chain_queue ADD COLUMN IF NOT EXISTS skip_reason text; + +COMMENT ON COLUMN new_chain_queue.flushed_at IS 'When this row was forwarded to the new chain, or when it was marked skipped. NULL means pending.'; +COMMENT ON COLUMN new_chain_queue.skip_reason IS 'Set when flushed_at was recorded without actually forwarding: ''backfilled'' (covered by the genesis backfill) or ''corrupt'' (tx_data failed to unmarshal).'; + +-- The flusher only ever reads pending rows, so index those alone. The retained +-- flushed rows stay out of the index and off the hot path no matter how large +-- the table grows. +CREATE INDEX IF NOT EXISTS new_chain_queue_pending_idx + ON new_chain_queue (id) WHERE flushed_at IS NULL; + +COMMIT;