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;