Gate the background ohlcv task on IBD completion

On master the synchronous backfill always ran and blocked on
initial_sync_complete, so the background task spawned afterwards could not start
until IBD had finished. Stage 1 wrapped that backfill in `if !ohlcv_wiped`, which
removed the barrier for the background task as a side effect.

The guard fires more often than a version bump suggests: a fresh database has no
version key, so `stored (None) != Some(OHLCV_VERSION)` and migrate_if_stale
reports a wipe. Every clean resync therefore skipped the IBD-gated path entirely.

Left ungated, the task sweeps from the first confirmed trade to `now - 3h` while
indexing is still years behind. It materialises nothing, re-runs ~700 empty
24-hour batches every 600s, competes with block indexing for the cauldron write
lock, and advances materialized_end to roughly now over an empty table -- after
which candlesticks() takes the ohlcv fast path against nothing instead of falling
back to the raw path.

Waiting inside the spawned task rather than before the spawn keeps startup
non-blocking.

Tests: 239 passing

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
jakobsn 2026-07-29 19:49:29 +02:00
parent 8e45d5f216
commit d961a5f762

View file

@ -515,18 +515,33 @@ async fn launch() -> _ {
} }
// Background task: incrementally materialise new 1-hour OHLCV buckets as blocks arrive. // Background task: incrementally materialise new 1-hour OHLCV buckets as blocks arrive.
// The full historical backfill above already ran; this task only handles the tail. // After a version wipe this is also what repopulates history, since the synchronous
// backfill above is skipped in that case.
// Only processes buckets older than 3 hours (well beyond BCH reorg depth). // Only processes buckets older than 3 hours (well beyond BCH reorg depth).
{ {
let ohlcv_write = dbpool.cauldron_w.clone(); let ohlcv_write = dbpool.cauldron_w.clone();
let ohlcv_read = dbpool.cauldron_r.clone(); let ohlcv_read = dbpool.cauldron_r.clone();
let ohlcv_state_bg = ohlcv_state.clone(); let ohlcv_state_bg = ohlcv_state.clone();
let ohlcv_ibd = ibd_state.clone();
tokio::spawn(async move { tokio::spawn(async move {
// Batch size: 1 day per SQL call to keep each write short. // Batch size: 1 day per SQL call to keep each write short.
const BATCH_SECS: i64 = 24 * 3600; const BATCH_SECS: i64 = 24 * 3600;
// Safety margin: only materialise buckets older than this many seconds. // Safety margin: only materialise buckets older than this many seconds.
const SAFETY_SECS: i64 = 3 * 3600; const SAFETY_SECS: i64 = 3 * 3600;
// Wait for IBD before materialising anything.
//
// The synchronous backfill above waits too, but it is skipped whenever
// `ohlcv_wiped` is set — which includes every fresh database, since a missing
// version key reads as stale. Without this the task would sweep from the
// first trade all the way to `now - 3h` while indexing is still years behind,
// writing nothing, contending with block writes for the cauldron write lock,
// and advancing `materialized_end` to roughly now against an empty table — at
// which point `candlesticks()` would take the fast path over nothing.
while !ohlcv_ibd.initial_sync_complete.load(Ordering::Relaxed) {
tokio::time::sleep(Duration::from_secs(1)).await;
}
loop { loop {
let now = crate::timeutil::time_now(); let now = crate::timeutil::time_now();
// Floor to 1-hour boundary, 3 hours ago. // Floor to 1-hour boundary, 3 hours ago.