From c4571e284490624612344692ad5065dd020938c4 Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Wed, 26 Aug 2026 03:00:31 -0700 Subject: [PATCH] feat(api): add the indexer and flusher bounds the chain cutover needs Three config values, all defaulting to 0 meaning unbounded, so nothing changes in normal operation. etlStartingBlockHeight / etlEndingBlockHeight are passed to the ETL's existing SetStartingBlockHeight / SetEndingBlockHeight, which indexer.go never called. Without them the indexer only knows how to resume, and the resume query -- MAX(block_height) FROM etl_blocks -- 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, silently. 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. newChainFlushToBlock is a ceiling on the flusher, the mirror of the existing newChainFlushFromBlock. The cutover stops the old-chain indexer at a height L and needs everything confirmed at or below L to be on the new chain -- and nothing above L across that line -- before the new indexer starts. The ceiling filters rather than halts. Enqueue is dispatched asynchronously, so confirmed_block is only roughly ordered by id; stopping at the first row above the ceiling would strand one below it, which would then flush after the boundary was recorded and be indexed twice. Rows with a NULL confirmed_block cannot be placed relative to L and are held until the ceiling lifts. Tests cover the ceiling boundary (inclusive), that it filters past an out-of-order row, that NULLs are held and then released, and that an unset ceiling is unbounded -- confirmed failing without the change. A config test covers the parse helper, since a malformed bound reading as 0 would silently disable the limit it was meant to impose. --- api/new_chain_flusher.go | 30 +++++++++-- api/new_chain_flusher_test.go | 95 +++++++++++++++++++++++++++++++++-- config/config.go | 40 ++++++++++++--- config/env_parse_test.go | 31 ++++++++++++ indexer/indexer.go | 19 +++++++ 5 files changed, 201 insertions(+), 14 deletions(-) create mode 100644 config/env_parse_test.go diff --git a/api/new_chain_flusher.go b/api/new_chain_flusher.go index 37b1de77..2adca2be 100644 --- a/api/new_chain_flusher.go +++ b/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 } diff --git a/api/new_chain_flusher_test.go b/api/new_chain_flusher_test.go index 674f01ce..d574244d 100644 --- a/api/new_chain_flusher_test.go +++ b/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 +} diff --git a/config/config.go b/config/config.go index ed1a8439..5d55e422 100644 --- a/config/config.go +++ b/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 } diff --git a/config/env_parse_test.go b/config/env_parse_test.go new file mode 100644 index 00000000..a2b6ff0c --- /dev/null +++ b/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") + }) +} diff --git a/indexer/indexer.go b/indexer/indexer.go index cca9f8d7..21e6f5c0 100644 --- a/indexer/indexer.go +++ b/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.