feat(ingest): rewrite backfill as a streaming fetch → process → flush pipeline

[aditya1702]

Stacked on protocol-migrate-windowed-coalescing (base of this PR), which is itself open as #622. Review/merge that first.

TL;DR

Rewrites historical-ledger backfill (internal/services/ingest_backfill.go) from a per-batch worker pool into a single streaming fetch → process → flush pipeline built on the optimizedStorageBackend fork introduced in #622. Net −1709 lines.

Area	Change	Why it matters
Streaming pipeline	One long-lived backend per gap → a single fetchLedgers goroutine → bounded rawCh → N process workers → bounded resultCh → one ordered flusher. Coordinated by errgroup.	Overlaps fetch/decode/DB-write instead of running them serially within each batch, and stops spinning up a fresh optimizedStorageBackend (10 decode workers + prefetch buffer) every ~250 ledgers — there's now exactly one per gap.
One parallelism axis	Each process worker uses a nil-pool serial indexer (new inline path in ProcessLedgerTransactions when the pool is nil) instead of N batch workers each fanning out onto a shared unbounded pond pool.	Removes the nested, uncapped goroutine fan-out that drove GC overhead past 58% in earlier profiling. Concurrency is now exactly --backfill-workers; decode parallelism stays in the fork's num-workers.
Single sequential consumer	fetchLedgers is the sole GetLedger caller, ascending, one backend per gap.	This is the only legal way to drive the fork (it rejects out-of-order sequences and forbids cross-goroutine sharing). The pipeline's shape is dictated by that contract, not just preference.
Progressive compression, simplified	The recompressor's watermark (completed[] / watermarkIdx / MarkDone) and its lower bound are gone. The ordered flusher emits a monotonic "safe-end" close time; the compressor compresses range_end < safeEnd, coalescing bursts via drainLatest.	Same progressive behavior (bounds peak uncompressed disk), but ordering lives in one place — the flusher — so the compressor is a stateless one-liner. Deletes a whole test file.
Fail-fast	A fetch/process/flush error cancels the pipeline and startBackfilling returns the error (previously it logged per-batch failures and returned nil).	Failures surface loudly. Flushed data stays durable (atomic txns, synchronous_commit=off); a rerun resumes via GetLedgerGaps.
Config	Drop --backfill-batch-size; rename --backfill-db-insert-batch-size → --backfill-flush-size.	One meaningful flush knob alongside --backfill-workers. Pre-release, so no compat shim.

Two correctness properties are load-bearing and were explicitly verified in review: (1) TimescaleDB range_end is exclusive, so compressing range_end < safeEnd never compresses a chunk before all its ledgers are durable; (2) the compressor's safe-end stays monotonic across gaps because calculateBackfillGaps returns gaps in ascending order.

---

Why

Backfill replays historical ledgers into TimescaleDB. The previous design layered three uncoordinated concurrency mechanisms — a pond batch pool, a fresh BufferedStorageBackend per 250-ledger batch, and an unbounded shared indexer pool — with fetch/decode/process/flush running serially inside each batch. That churns backends, oversubscribes goroutines, and leaves cores idle during every fetch stall and DB flush. This rewrite collapses it to one idiomatic pipeline with a single parallelism axis, reusing the decode pipeline the optimizedStorageBackend fork already provides.

Suggested review order

1. internal/services/backfill_pipeline.go — the pipeline: orderedBuffer, fetchLedgers, processWorker, runFlusher, runGapPipeline (the errgroup wiring + channel-close discipline).
2. internal/services/ingest_backfill.go — startBackfilling (validate → gaps → per-gap pipeline → compressor.Wait() on all paths); calculateBackfillGaps is unchanged.
3. internal/services/backfill_compressor.go — the stateless progressive compressor + drainLatest.
4. internal/indexer/indexer.go — the nil-pool serial path (live pooled path is byte-for-byte unchanged).
5. internal/services/ingest.go, internal/ingest/ingest.go, cmd/ingest.go — the processLedger split + config rename.

Design context: ~/.claude project spec/plan (out of repo).

Reviewer notes / before the in-cluster benchmark

• Deployment manifests (outside this repo): the backfill Job must drop --backfill-batch-size / --backfill-db-insert-batch-size and use --backfill-flush-size, or it errors on an unknown flag.
• First benchmark: pin --backfill-workers=8 — the NumCPU default may exceed the ~8-process GC sweet spot on a high-core node; co-tune with the fork's num-workers (decode) per the earlier ~16/8 split.
• Open question: the pre-existing --compression-max-chunks flag implies a TimescaleDB columnstore policy/job may already compress chunks in-cluster. If so, the app-level progressive compressor may partly overlap (safe — compress_chunk is idempotent on disjoint chunks — but possibly redundant). Worth confirming whether the app compressor is still wanted.

Concurrency notes

• Every blocking channel op selects on ctx.Done(), so first-error cancellation can't deadlock on a full rawCh or an unread resultCh.
• rawCh is closed by its sole sender (the fetcher); resultCh is closed by a single fan-in goroutine after wg.Wait() — no double-close, no send-on-closed on the error path.
• The compressor goroutine is joined via compressor.Wait() on both the success and gap-error paths.

Test plan

• go build ./... + go vet ./... — clean.
• ENABLE_INTEGRATION_TESTS=false go test -race on internal/services, internal/indexer, internal/ingest — pass (full services suite included, exercising the shared-processLedger live path).
• New unit tests: orderedBuffer reorder (out-of-order → contiguous runs), drainLatest coalescing, indexer serial==pooled equivalence, startBackfilling fail-fast validation.
• make check (lint / deadcode / shadow) — via CI.
• In-cluster backfill throughput benchmark — maintainer to run (see reviewer notes above).

🤖 Generated with Claude Code

[chatgpt-codex-connector]

Codex usage limits have been reached for code reviews. Please check with the admins of this repo to increase the limits by adding credits.
Credits must be used to enable repository wide code reviews.