Streaming daemon: Phase 1 layer 2 — primitives (#815)

[chowbao]

Part of #815 — Phase 1 (Backfill), layer 2 of 3. Stacked on PR1A (streaming-phase1-foundations).

The primitives — single-pass cold materialization for all three data types:

• processChunk — materializes a chunk's ledgers (.pack), events (cold segment), and tx-hash (sorted .bin) artifacts for every Kind, with per-kind idempotency
• the cold backfillSource (frozen .pack → bulk backend)
• buildTxhashIndex — k-way merge of the in-window .bin runs into a sorted .idx (byte-format-identical to the single-index build)
• adds the ingest.RunColdChunk single-chunk cold primitive

Verification: go build + go vet + go test -short green on ./cmd/stellar-rpc/internal/fullhistory/... (cgo RocksDB toolchain). Note: the full cmd/stellar-rpc binary link requires the Rust libpreflight/libxdr2json (CI make build-libs); go vet ./cmd/stellar-rpc/ type-checks the entrypoint locally. golangci-lint runs in CI.

Stack: streaming-phase1-primitives → streaming-phase1-foundations

[tamirms]

Could this snapshot and the newParent plumbing into BarrierNewFile just go away? It exists to skip the grandparent fsync when the bucket already existed — but fsyncing an unchanged directory is nearly free (no dirty metadata to flush). If BarrierNewFile always did file → parent → grandparent, you'd delete this loop, the newParent parameter, and the newIndexDir branch in txindex.go, for one fewer before-the-write step to get subtly wrong. (geometry's DeepestExistingDir/FsyncNewDirs already do "fsync a created dir chain" if you'd rather keep the optimization but not hand-roll it.)

[chowbao]

Done in 7a51cf81. BarrierNewFile now always fsyncs file → parent → grandparent (the newParent bool is gone), so the bucketExisted snapshot loop in processChunk and the newIndexDir branch in buildTxhashIndex both disappear — and with the stat gone the index dir is just an idempotent os.MkdirAll. Always-fsync-grandparent is a strict superset of the old durability (an unchanged dir has no dirty metadata to flush, so it's nearly free), so it's never less safe.

I kept the hand-rolled two-level barrier rather than reaching for DeepestExistingDir/FsyncNewDirs: after the startup root creation, only a single bucket/index dir is ever created on demand, so file → parent → grandparent is exactly the depth needed — those helpers are for the multi-level root chain at startup.

[tamirms]

This mark → write → barrier → flip skeleton is the crash-safety invariant, and it's spelled out twice (here and in buildTxhashIndex), with a third copy in the queued live-ingestion PR — openHotTierForChunk is PutHotTransient → mkdir/open → fsync → FlipHotReady. Three sites share this exact order. A small oneWrite(mark, create, barrier, flip) helper would keep the order in one place so none of the three can drift; the states and the create step differ per site, the order doesn't. Worth establishing here and adopting in the hot-tier open.

[chowbao]

Done in 7a51cf81. Added oneWrite(mark, create, barrier, flip func() error) error and adopted it in both processChunk and buildTxhashIndex, so the mark → create → barrier → flip order is defined once and the sites only supply the per-step closures.

I scoped it to the backfill package for now since both current callers live here; when #820 adds the hot-tier openHotTierForChunk (PutHotTransient → mkdir/open → fsync → FlipHotReady) it can promote/relocate oneWrite and adopt it as the third caller. Folding the ordering out also dropped both functions under the cyclop threshold (13 < 15), so I removed their //nolint:cyclop directives.

[chowbao]

Done now instead — promoted oneWrite to catalog.OneWrite in catalog_protocol.go, right under the protocol header it implements (the file already documents the same mark→create→barrier→flip steps). processChunk and buildTxhashIndex now call it across the package boundary.

It's a zero-dependency pure function and catalog never imports backfill, so there's no import cycle; #820's openHotTierForChunk adopts it as the third caller by import alone, with no later relocation. Pushed in 7e56b926.

[tamirms]

github.com/cenkalti/backoff/v4 is already a direct dependency — this hand-rolled poll loop could use it and shed the timer/deadline/select bookkeeping. backoff.Retry with a ConstantBackOff (the interval), WithContext (cancellation), and WithMaxElapsedTime (the timeout) covers all of it; wrap the tip-query error in backoff.Permanent to keep its current fatal-on-error behavior. The executor's own retry logic (#819) could share the same primitive too.

[chowbao]

Done in 7a51cf81. WaitForCoverage now drives backoff.Retry: the tip-query error is wrapped in backoff.Permanent (stays fatal), and the low-tip case returns the ErrBackendCoverageTimeout-wrapped error — which Retry surfaces once MaxElapsedTime stops the loop, so callers still classify the timeout via errors.Is.

Two deviations from the literal recipe, both deliberate:

• WithMaxElapsedTime is an ExponentialBackOffOpts, not a wrapper around ConstantBackOff, so the constant poll is expressed as NewExponentialBackOff with WithMultiplier(1) + WithRandomizationFactor(0) (+ WithInitialInterval/WithMaxInterval = the interval).
• I kept a per-call context.WithDeadline inside the operation: the backoff bounds total retry time but not a single in-flight Tip call, so dropping it would regress the hung-query bound added in 808eb622 (and its TestPollingBackendWaiter_TipBoundedByDeadline).

Added TestPollingBackendWaiter_TimeoutReturnsSentinel to cover the timeout→sentinel path. Agreed the #819 executor retry can share this primitive.

[tamirms]

go.mod is on 1.26, and the catalog already iterates this way (for e, err := range c.store.PrefixScan(...)), so this callback is a bit against the grain. As an iter.Seq2[chunk.ID, geometry.State] it'd read for cid, state := range cat.TxHashStates(lo, hi) { … }, and the two callers (txhashBinInputs, demotedTxhashRefs) could use a plain for with break/continue instead of returning sentinel errors from a closure.

[chowbao]

Done in 7a51cf81. forEachChunkTxHashState is now txHashStates(...) iter.Seq2[…, error], and txhashBinInputs/demotedTxhashRefs use a plain for cs, err := range … with normal error handling and continue instead of returning sentinel errors from a closure.

One adjustment from the sketch: since catalog.State can return an error, I followed the codebase's iter.Seq2[T, error] scan convention (the same shape as PrefixScan/IterateLedgers) rather than iter.Seq2[chunk.ID, geometry.State] — yielding a small chunkTxHashState{Chunk, State} value paired with the error. I left it as a backfill free function (not a cat.TxHashStates method) to avoid growing the catalog API for a backfill-local helper, but happy to move it to a method if you'd prefer that form.

[tamirms]

Side note, visible here but really a #824 / geometry-package thing: the fsync helpers BarrierNewFile & co. don't belong in geometry. BarrierNewFile, FsyncDir, FsyncParentDirs, DeepestExistingDir, FsyncNewDirs, DeleteFileIfExists, RmdirIfEmpty are generic os-level durability ops with zero geometry or domain references — they only landed in geometry because they shared paths.go with the path bijection in the monolith. geometry should be naming + math (identity and location, no I/O); crash-safe file ordering is a different concern that'd sit better in a small durable/fsync leaf package imported by catalog, backfill, and the hot-tier code. Worth folding into the #824 split.

[chowbao]

Added to #824

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 4f85de7dfd

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[tamirms]

Two remaining items in the backfill/ingest area — both more structural than the last round, and both cheap to do now precisely because none of this is wired into the daemon yet (ProcessConfig/processChunk/RunColdChunk have no production constructor; the daemon still runs the legacy ingest.BackfillMeta path). Fixing them before the streaming layers stack on top is much cheaper than after.

1. The ledger-source boundary between backfill and ingest

The problem. ingest.ChunkSource exposes only OpenStream(chunkID) → LedgerStream, which discards the one thing the bulk source can actually report: how far it currently reaches. To get that back, backfill bolts on a parallel apparatus — BackendWaiter + pollingBackendWaiter + an injected Tip func — to poll a "latest ledger" the source already knows. Separately, ingest still carries the dead multi-chunk batch path (RunCold → runOneChunkCold → buildColdIngesters) next to the live single-chunk RunColdChunk, which duplicates it almost line-for-line. These are two symptoms of one thing: the source abstraction is drawn in the wrong place.

What the SDK already gives us. ledgerbackend.LedgerStream (RawLedgers(ctx, range) iter.Seq2[[]byte,error]) is a universal streaming currency everything already speaks — the local pack (packStream already implements it) and the external backends (NewBufferedStorageStream, NewCaptiveCoreStream) all return it. So ChunkSource/OpenStream is a redundant wrapper over it.

The tip is a separate value — and one trap worth flagging: the SDK's LedgerBackend.GetLatestLedgerSequence is not the tip we want. It's prepared-relative — on BufferedStorageBackend it errors before PrepareRange and otherwise returns the buffer's latest, not the source's frontier. The frontier tip is an independent value per backend: datastore.FindLatestLedgerSequence for the lake, and the history archive's GetRootHAS().CurrentLedger for captive core (whose own GetLatestLedgerSequence likewise needs a running, prepared core, so it's no use here).

Proposed shape. The materializer becomes source-blind — it takes the raw ledger iterator and nothing else, and is faked in tests with a literal slice iterator:

func WriteColdChunk(ctx, logger, chunkID chunk.ID,
	raw iter.Seq2[[]byte, error], dirs ColdDirs, sink MetricSink, cfg Config) error

backfill owns the external-source seam, one interface:

// An external ledger source (BSB now, captive core later) we fetch from and may
// have to wait for. The local frozen .pack is NOT a Backend — it's complete, so
// it's a bare ledgerbackend.LedgerStream with nothing to wait on.
//   bsbSource     (now)   Tip = datastore.FindLatestLedgerSequence
//   captiveSource (later) Tip = historyarchive GetRootHAS().CurrentLedger
type Backend interface {
	ledgerbackend.LedgerStream
	Tip(ctx context.Context) (uint32, error)
}
type ProcessConfig struct { /* … */ Backend Backend }   // was: Backend ChunkSource + BackendWaiter + Tip func

This Backend is a new interface and deliberately not an evolution of today's BackendWaiter — that interface, its pollingBackendWaiter implementation, and NewPollingBackendWaiter are removed entirely. The coverage-polling they did becomes just the Tip method above plus the waitForCoverage function below; nothing in backfill implements BackendWaiter anymore.

The one impl wired now:

type bsbSource struct {
	ledgerbackend.LedgerStream          // NewBufferedStorageStream
	ds datastore.DataStore              // long-lived; tip only; daemon-Closed
}
func (s *bsbSource) Tip(ctx context.Context) (uint32, error) { return datastore.FindLatestLedgerSequence(ctx, s.ds) }

Coverage-waiting collapses to one waitForCoverage function that polls Backend.Tip. This is what removes BackendWaiter, pollingBackendWaiter, and NewPollingBackendWaiter:

func waitForCoverage(ctx, b Backend, target uint32, interval, timeout time.Duration) error // polls b.Tip

The local pack moves to its codec owner as a bare stream, constructed inside backfillSource (not injected) when a chunk's ledgers are already frozen locally and we're deriving the other artifacts from them:

func NewPackStream(path string) ledgerbackend.LedgerStream   // the existing packStream, relocated to pkg/stores/ledger

so the call site is uniform — both arms are just a LedgerStream:

src, err := backfillSource(ctx, chunkID, artifacts, cfg)   // pack OR cfg.Backend
raw := src.RawLedgers(ctx, ledgerbackend.BoundedRange(chunkID.FirstLedger(), chunkID.LastLedger()))
ingest.WriteColdChunk(ctx, cfg.Logger, chunkID, raw, dirs, cfg.Sink, ingestConfigFor(artifacts))

Captive core is deferred, but the path stays open. No need to implement it now — the Backend interface is the seam. It's one new struct (embed NewCaptiveCoreStream + a Tip from the history archive) plus one daemon factory case, with zero changes to ProcessConfig, backfillSource, waitForCoverage, or WriteColdChunk. Captive's tip comes from historyarchive.NewArchivePool(cfg.HistoryArchiveURLs, …).GetRootHAS().CurrentLedger. Listing the intended impls in the Backend doc comment keeps the intent visible without a stub.

Net: delete ingest/source.go whole (ChunkSource, ChunkSourceFunc, packSource, dataStoreSource, NewDataStoreSource); the dead RunCold/runOneChunkCold/buildColdIngesters; and backfill's BackendWaiter/pollingBackendWaiter/NewPollingBackendWaiter/Tip. Rename buildColdIngestersIn → buildColdIngesters and RunColdChunk → WriteColdChunk. Roughly −7 types and −3 constructors for +1 interface.

Two things to keep honest in the code: the ingest package still imports ledgerbackend for RunHot, so only the materializer function is source-blind, not the package; and since source resolution (pack-stat, coverage wait) now runs in backfill before WriteColdChunk, a pack-missing or coverage-timeout failure is metered there rather than as an ingest ColdChunkTotal attempt — worth a line in the doc.

2. catalog.OneWrite — revert the two backfill sites to straight-line

First, apologies for the round-trip: I asked for the oneWrite wrapper last round, and having gone through both call sites properly since, straight-line reads better and I'd suggest reverting it. Sorry for sending you back and forth on it.

Why the wrapper doesn't pay off: every step wraps its error with per-site context (chunkID/artifacts in processChunk; w/lo/hi/cov/idxPath in buildTxhashIndex), so the closures can't collapse to one-liners — they stay multi-line and carry the closure wrapper on top. And buildTxhashIndex pays extra: because the flip needs the cov the mark returns, it hoists var cov …; var idxPath string above the call and does var merr error; cov, merr = … (no :=). Straight-line, those are ordinary downward locals:

// one-write:mark — returns the coverage the commit (flip) needs.
cov, err := cat.MarkTxHashIndexFreezing(w, lo, hi)
if err != nil { return fmt.Errorf("buildTxhashIndex mark freezing %s: %w", geometry.TxHashIndexKey(w, lo, hi), err) }
idxPath := layout.TxHashIndexFilePath(cov)
// one-write:create  (os.MkdirAll + txhash.BuildColdIndex(ctx, inputs, idxPath, …))
// one-write:barrier (geometry.BarrierNewFile(idxPath))
// one-write:flip — CommitTxHashIndex(cov) promotes cov, demotes predecessor, demotes window .bin on terminal.
if err := cat.CommitTxHashIndex(cov); err != nil { return fmt.Errorf("buildTxhashIndex commit %s [%s,%s]: %w", w, lo, hi, err) }
return nil

Nothing is lost on safety: the wrapper only sequences four calls — it never checks their contents (a wrong fsync path, the wrong catalog method, a dropped step all pass it), so it isn't what guarantees crash-safety. That guarantee is catalog/crashsafety_test.go, which drives the protocol to each crash point and asserts the surviving state is recoverable. A // one-write:mark|create|barrier|flip label per step keeps the four beats visible and greppable. processChunk converts the same way.

So: delete catalog.OneWrite and inline both sites.

[chowbao]

Both done in 096463ef.

1. Source boundary.

• Deleted ingest/source.go (ChunkSource/packSource/dataStoreSource) and the dead RunCold path.
• Materializer is now source-blind: WriteColdChunk(ctx, logger, chunkID, raw iter.Seq2[[]byte, error], dirs, sink, cfg).
• buildColdIngestersIn → buildColdIngesters; packStream → ledger.NewPackStream(path).
• backfill owns the seam: Backend{ ledgerbackend.LedgerStream; Tip } (bsbSource now, captive later), BackendWaiter/polling collapsed into waitForCoverage, ProcessConfig down to one Backend, backfillSource returns a bare LedgerStream.

2. catalog.OneWrite reverted — inlined straight-line at both sites with // one-write: labels.

[tamirms]

Approving. The source-boundary redesign and the OneWrite revert both landed cleanly in 096463ef, and the rest of the primitives layer matches the design — the resolver/executor and daemon wiring are correctly left to their later layers, and the empty-window (#826) and geometry/fsync (#824) items are tracked. Build and unit tests are green; the two red checks are the unrelated XDR Go/Rust rev mismatch.

A few optional cosmetic nits, none blocking:

• process.go:93 — the one-write comment is garbled (a dangling "The" and a doubled // //); txindex.go has the clean version to copy.
• geometry/paths.go:109 — the TxHashRawRoot doc still references RunCold, which this PR removed; the TestRunCold_* names in ingest_test.go are similarly stale now that they call WriteColdChunk.
• ingest/service.go — the ColdService struct comment says the timer starts "from the Finalize/Close call," but it actually starts at construction (the constructor comment is the accurate one).