fix(api): retain flushed new-chain queue rows instead of deleting them

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

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,26 +88,59 @@ 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)
 	require.NoError(t, err)
 	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)
+}

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;