feat(api): add the indexer and flusher bounds the chain cutover needs

api/new_chain_flusher.go

@@ -128,11 +128,33 @@ type queueRow struct {
 	txRaw []byte
 }
 
+// fetchBatch returns the next rows to send, oldest first.
+//
+// NewChainFlushToBlock, when set, is a ceiling: rows confirmed above it are left
+// pending. It is the mirror of NewChainFlushFromBlock and exists for the indexer
+// cutover, where the old-chain indexer stops at a height L and everything
+// confirmed at or below L must be on the new chain before the new indexer starts
+// (ROLLOUT.md, Runbook step 12).
+//
+// A filter, not a stop. Enqueue is dispatched asynchronously, so confirmed_block
+// is only roughly ordered by id — halting at the first row above the ceiling
+// would strand a row below it, which would then flush after the boundary was
+// recorded and be indexed twice.
+//
+// Rows with a NULL confirmed_block are held while the ceiling is set: they
+// cannot be placed relative to L, and holding is the recoverable choice —
+// they flush once the ceiling is lifted. Drain or inspect them before the
+// cutover rather than discovering them during it.
 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`,
-		limit,
-	)
+	query := `SELECT id, tx_data FROM new_chain_queue ORDER BY id LIMIT $1`
+	args := []any{limit}
+	if f.cfg.NewChainFlushToBlock > 0 {
+		query = `SELECT id, tx_data FROM new_chain_queue
+		         WHERE confirmed_block IS NOT NULL AND confirmed_block <= $2
+		         ORDER BY id LIMIT $1`
+		args = append(args, f.cfg.NewChainFlushToBlock)
+	}
+	rows, err := f.writePool.Query(ctx, query, args...)
 	if err != nil {
 		return nil, err
 	}

api/new_chain_flusher_test.go

@@ -147,7 +147,7 @@ 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))
 
@@ -216,9 +216,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)
@@ -244,3 +244,90 @@ func TestNewChainFlusherTrimThenSend(t *testing.T) {
 	}
 	require.ElementsMatch(t, []int64{3, 4}, receivedIDs)
 }
+
+// The ceiling is the mirror of NewChainFlushFromBlock: rows confirmed above it
+// stay pending. It exists so everything confirmed at or below the indexer's stop
+// height L can be landed on the new chain before the new indexer starts, without
+// letting anything above L across the boundary first.
+func TestNewChainFlusherCeilingHoldsRowsAboveIt(t *testing.T) {
+	cfg := &config.Config{NewChainFlushToBlock: 100}
+	f, _ := newTestFlusher(t, cfg)
+
+	block50 := int64(50)
+	block100 := int64(100)
+	block101 := int64(101)
+	block200 := int64(200)
+	insertQueueRow(t, f, sampleTx(1), &block50)  // below ceiling — eligible
+	insertQueueRow(t, f, sampleTx(2), &block100) // at ceiling — eligible, inclusive
+	insertQueueRow(t, f, sampleTx(3), &block101) // above — held
+	insertQueueRow(t, f, sampleTx(4), &block200) // above — held
+
+	batch, err := f.fetchBatch(context.Background(), 100)
+	require.NoError(t, err)
+	require.Len(t, batch, 2, "only rows confirmed at or below the ceiling are eligible")
+	require.Equal(t, []int64{1, 2}, entityIDsOf(t, batch))
+}
+
+// A row below the ceiling sitting behind one above it must still be returned.
+// Enqueue is fire-and-forget, so confirmed_block is only roughly ordered by id;
+// a halt-on-first-above-ceiling would strand this row, and it would then flush
+// after the boundary was recorded and be indexed a second time.
+func TestNewChainFlusherCeilingFiltersRatherThanHalts(t *testing.T) {
+	cfg := &config.Config{NewChainFlushToBlock: 100}
+	f, _ := newTestFlusher(t, cfg)
+
+	above := int64(200)
+	below := int64(50)
+	insertQueueRow(t, f, sampleTx(1), &above) // lower id, above the ceiling
+	insertQueueRow(t, f, sampleTx(2), &below) // higher id, below it
+
+	batch, err := f.fetchBatch(context.Background(), 100)
+	require.NoError(t, err)
+	require.Equal(t, []int64{2}, entityIDsOf(t, batch),
+		"the row below the ceiling must be reachable past one above it")
+}
+
+// NULL confirmed_block cannot be placed relative to the ceiling, so it is held
+// while one is set and flushes once it is lifted. Holding is the recoverable
+// choice: sending it early could double-index, dropping it would lose the write.
+func TestNewChainFlusherCeilingHoldsNullConfirmedBlock(t *testing.T) {
+	cfg := &config.Config{NewChainFlushToBlock: 100}
+	f, _ := newTestFlusher(t, cfg)
+
+	insertQueueRow(t, f, sampleTx(1), nil)
+
+	batch, err := f.fetchBatch(context.Background(), 100)
+	require.NoError(t, err)
+	require.Empty(t, batch, "NULL confirmed_block is held while a ceiling is set")
+
+	// Lifting the ceiling releases it.
+	f.cfg.NewChainFlushToBlock = 0
+	batch, err = f.fetchBatch(context.Background(), 100)
+	require.NoError(t, err)
+	require.Equal(t, []int64{1}, entityIDsOf(t, batch))
+}
+
+// With no ceiling configured the flusher is unbounded, as in normal operation.
+func TestNewChainFlusherNoCeilingSendsEverything(t *testing.T) {
+	cfg := &config.Config{}
+	f, _ := newTestFlusher(t, cfg)
+
+	block50 := int64(50)
+	insertQueueRow(t, f, sampleTx(1), &block50)
+	insertQueueRow(t, f, sampleTx(2), nil)
+
+	batch, err := f.fetchBatch(context.Background(), 100)
+	require.NoError(t, err)
+	require.Equal(t, []int64{1, 2}, entityIDsOf(t, batch))
+}
+
+func entityIDsOf(t *testing.T, batch []queueRow) []int64 {
+	t.Helper()
+	var ids []int64
+	for _, r := range batch {
+		var me corev1.ManageEntityLegacy
+		require.NoError(t, proto.Unmarshal(r.txRaw, &me))
+		ids = append(ids, me.EntityId)
+	}
+	return ids
+}

config/config.go

@@ -93,7 +93,20 @@ type Config struct {
 	NewChainQueueEnabled       bool
 	NewChainFlushEnabled       bool
 	NewChainFlushFromBlock     int64
+	NewChainFlushToBlock       int64
 	NewChainInsecureSkipVerify bool
+
+	// Indexer cutover bounds. Both default to 0, meaning "unset" — the ETL
+	// treats a zero start as "resume from where you left off" and a zero end as
+	// "never stop", which is normal operation.
+	//
+	// They exist for the chain cutover, where the old-chain indexer must stop at
+	// a known height L and the new-chain indexer must begin at a known height,
+	// rather than resuming off MAX(block_height) — a query with no chain_id,
+	// which against a fresh chain resolves to the old chain's height and stalls
+	// silently. See cmd/genesis-writer/ROLLOUT.md, Runbook step 12.
+	EtlStartingBlockHeight int64
+	EtlEndingBlockHeight   int64
 }
 
 var Cfg = Config{
@@ -367,11 +380,26 @@ func init() {
 	Cfg.NewChainQueueEnabled = os.Getenv("newChainQueueEnabled") == "true"
 	Cfg.NewChainFlushEnabled = os.Getenv("newChainFlushEnabled") == "true"
 	Cfg.NewChainInsecureSkipVerify = os.Getenv("newChainInsecureSkipVerify") == "true"
-	if v := os.Getenv("newChainFlushFromBlock"); v != "" {
-		n, err := strconv.ParseInt(v, 10, 64)
-		if err != nil {
-			panic("Invalid newChainFlushFromBlock: " + err.Error())
-		}
-		Cfg.NewChainFlushFromBlock = n
+	Cfg.NewChainFlushFromBlock = mustParseInt64Env("newChainFlushFromBlock")
+	Cfg.NewChainFlushToBlock = mustParseInt64Env("newChainFlushToBlock")
+
+	// Indexer cutover bounds (see the struct fields).
+	Cfg.EtlStartingBlockHeight = mustParseInt64Env("etlStartingBlockHeight")
+	Cfg.EtlEndingBlockHeight = mustParseInt64Env("etlEndingBlockHeight")
+}
+
+// mustParseInt64Env reads an optional int64 env var, returning 0 when unset.
+// A malformed value panics rather than silently reading as 0: every caller here
+// is a cutover bound where 0 means "no bound", so a typo would quietly disable
+// the very limit it was meant to impose.
+func mustParseInt64Env(name string) int64 {
+	v := os.Getenv(name)
+	if v == "" {
+		return 0
+	}
+	n, err := strconv.ParseInt(v, 10, 64)
+	if err != nil {
+		panic("Invalid " + name + ": " + err.Error())
 	}
+	return n
 }

config/env_parse_test.go

@@ -0,0 +1,31 @@
+package config
+
+import "testing"
+
+// The cutover bounds all mean "no bound" at 0, so a malformed value must not
+// read as 0 -- that would silently disable the very limit it was meant to set,
+// and the failure would only surface as duplicated or missing rows much later.
+func TestMustParseInt64Env(t *testing.T) {
+	t.Run("unset is zero", func(t *testing.T) {
+		if got := mustParseInt64Env("definitelyNotSetAnywhere"); got != 0 {
+			t.Errorf("expected 0 for unset, got %d", got)
+		}
+	})
+
+	t.Run("parses a value", func(t *testing.T) {
+		t.Setenv("someCutoverBound", "24000000")
+		if got := mustParseInt64Env("someCutoverBound"); got != 24000000 {
+			t.Errorf("expected 24000000, got %d", got)
+		}
+	})
+
+	t.Run("panics on garbage rather than reading as zero", func(t *testing.T) {
+		t.Setenv("someCutoverBound", "24_000_000")
+		defer func() {
+			if recover() == nil {
+				t.Error("expected a panic; a typo must not silently disable the bound")
+			}
+		}()
+		mustParseInt64Env("someCutoverBound")
+	})
+}

indexer/indexer.go

@@ -94,6 +94,25 @@ func NewIndexer(cfg config.Config) *CoreIndexer {
 	etlIndexer.SetDBURL(cfg.WriteDbUrl)
 	etlIndexer.SetCheckReadiness(true)
 
+	// Chain cutover bounds. Unset (0) is normal operation: resume from the last
+	// indexed block, and never stop.
+	//
+	// The resume path is MAX(block_height) FROM etl_blocks, which carries no
+	// chain_id. Pointed at a new chain it resolves to the old chain's height and
+	// waits for a block that will not exist for years — a silent stall, not an
+	// error. There is a chain-aware fallback to core_indexed_blocks, but it only
+	// runs when etl_blocks is empty, so it never fires on a database that has
+	// indexed the old chain. An explicit start height is the way across.
+	if cfg.EtlStartingBlockHeight > 0 {
+		etlIndexer.SetStartingBlockHeight(cfg.EtlStartingBlockHeight)
+		logger.Info("etl: explicit starting block height",
+			zap.Int64("height", cfg.EtlStartingBlockHeight))
+	}
+	if cfg.EtlEndingBlockHeight > 0 {
+		etlIndexer.SetEndingBlockHeight(cfg.EtlEndingBlockHeight)
+		logger.Info("etl: will stop after block", zap.Int64("height", cfg.EtlEndingBlockHeight))
+	}
+
 	// When enabled, source blocks from the CoreService.StreamBlocks gRPC stream
 	// instead of polling GetBlocks. The ETL falls back to polling automatically
 	// if the endpoint doesn't support it, so this is safe to flip on per-env.