feat: WIP Multistep scheduler by define-null · Pull Request #37 · subsquid/network-scheduler

define-null · 2026-06-12T22:58:09Z

What is this PR about?

Goal: make it possible to replace or correct data chunks in the network without breaking availability or consistency for portals reading them. Today the scheduler has no safe way to swap a chunk: a portal mid-query could see the old version vanish before the new one is ready. This branch builds the foundation for that — an MVCC (multi-version) chunk lifecycle — plus a smarter scheduling algorithm, and a large property-based test harness that validates both. Everything is behind the mvcc-chunks feature flag (off by default), so production builds are unaffected

MVCC chunk lifecycle (src/scheduler_storage/, migrations/)

A new storage layer that tracks every chunk through a versioned lifecycle instead of a single mutable assignment:

Two kinds of published state: the worker assignment (everything workers must hold) and portal assignments (point-in-time snapshots portals read from). Workers keep serving old snapshots for a grace window (the "M window") so slow portals never see data disappear under them.
A confirmation watermark: data is only dropped after a quorum of workers confirms it has applied the newer assignment. Until then, outgoing copies "drain" rather than vanish.
Chunk corrections: a 1-to-1 swap mechanism (register a replacement → it gets placed and confirmed → the swap fires atomically in a visibility cycle → the old chunk drains out). The old row is retained for audit.
Two interchangeable backends behind one trait: a fast in-memory implementation and a Postgres implementation (new migrations/, transactional cycles, advisory-lock guarded).

Multi-step (reconciliation) scheduling (src/multistep_scheduler.rs)

The current algorithm computes an ideal placement from scratch, blind to what workers already hold — which can demand more data movement than the fleet can absorb. The new algorithm reconciles instead: starting from the current placement, it produces a feasible step toward the ideal — held copies are free to keep, mandatory replication ("floor") copies preempt nice-to-have ("bonus") copies, and new chunks are placed all-or-nothing so nothing lands half-replicated. Standalone for now — not yet wired into the production path.

Simulation & property-based testing (src/multistep_scheduler/sim/)

A model-driven state-machine test harness that drives the full lifecycle the way the real network would — random walks of 100–300 steps mixing chunk additions, worker joins/departures/lagging fetches, clock jumps, corrections, and replication-factor changes — then checks safety/liveness oracles after every step (no portal ever routes to a worker that lacks the chunk; floors are eventually met; drains terminate). The same walks run against both the in-memory and Postgres backends (via testcontainers), with seed-based replay, captured regressions, statistics telemetry, and CI budget knobs (SIM_IN_MEMORY_CASES/SIM_PG_CASES, set to 16/2 on CI).

Supporting changes & docs

schedule_with_per_worker_allocations added to the existing scheduler so placement can account for per-worker occupied bytes; small extensions to weight.rs and test utilities; new reshuffling-cost scenario tests.
Six design docs under docs/ (mvcc-chunks.md, mvcc-corrections.md, mvcc-schema.md, capacity-aware-scheduling.md, chunk-reshuffling.md, mvcc-worker-mappings.md) — the durable references the code comments point to.

toschoosqd · 2026-06-16T10:53:42Z

    Ok(reliable)
 }

+pub fn schedule_with_per_worker_allocations(


schedule_with_per_worker_allocations appears to bypass the reliable-first/two-pass behavior from schedule(): it schedules all workers in one pass regardless of ignore_reliability. Since DefaultSchedulingAlgorithm uses this even with empty allocations, enabling MVCC default scheduling can assign chunks to unreliable workers where the current scheduler would keep them on reliable workers. Should this entry point mirror schedule()’s two-pass logic?

I have somewhat conflicting views regarding ignore_reliability, let's discuss.

…diagnosis (#37) * perf(pg-storage): instrument scheduler_storage phases for slow-cycle diagnosis Wrap every SQL-issuing phase of the Postgres SchedulerStorage in a PhaseTimer that records elapsed time, statements issued, and rows touched, emitting one structured `info` line per phase under a dedicated `pg_timing` tracing target (plus the existing EXEC_TIMES prometheus family). The statement count separates one slow query (statements=1) from an N+1 round-trip storm (statements ≈ chunks). Select just the breakdown on an otherwise-quiet run: RUST_LOG=error,pg_timing=info On the reshuffle-sim multistep path at ~95K chunks, this pinpoints five N+1 loops that dominate a cycle — insert_new_chunks, record_routing_diffs, commit_new_ideal, replay_confirmed_diffs, resolve_flip_flops (the last issues one round-trip per chunk for 0 rows) — while every set-based phase stays sub-second. No behavior change; instrumentation only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * build: bump sqlx 0.8.6 → 0.9.0 and Rust 1.89 → 1.94 sqlx 0.9.0 requires Rust 1.94 (rust-toolchain.toml). 0.9 gates dynamically built SQL behind the SqlSafeStr trait (only &'static str qualifies); the three sites that assemble SQL from in-crate fragments — the non-overlap probe and two test helpers — are audited-safe and wrapped in AssertSqlSafe. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Defnull <879658+define-null@users.noreply.github.com> Stacked PR **1 of 2**. This one adds only the diagnostic instrumentation; the set-based query rewrites it enabled are in the follow-up PR stacked on top. ## What A `PhaseTimer` (in `src/metrics.rs`) wraps every SQL-issuing phase of the Postgres `SchedulerStorage`. Per phase it records elapsed time, **statements issued**, and **rows touched**, emitting one structured `info` line under a dedicated `pg_timing` tracing target and feeding the existing `EXEC_TIMES` prometheus family. No behavior change. The statement count is the signal that separates *one slow query* (`statements=1`) from an *N+1 round-trip storm* (`statements ≈ chunk count`) — indistinguishable by time alone. ## Why The reshuffle-sim multistep run was slow at scale (~10M chunks) and `RUST_LOG=error` hid the existing `debug` timers (`EXEC_TIMES` is never dumped by the sim). This surfaces a per-phase breakdown on the slow run with no other noise: ``` RUST_LOG=error,pg_timing=info cargo run --release -p reshuffle-sim -- \ -c examples/testnet_config.yaml --chunks-per-step 1000 --steps 5 --scheduler multistep <input>.fb.1.gz ``` ## What it revealed (drives PR 2) At ~95K chunks, one cycle: `insert_new_chunks` 95,092 stmts/23.6s, `record_routing_diffs` 95,092/18.6s, `commit_new_ideal` 95,093/15.9s, `replay_confirmed_diffs` 95,093/14.3s, `resolve_flip_flops` 95,092/13.3s (0 rows) — five N+1 loops; every set-based phase sub-second. ## Scope Instruments `mod.rs` (6 trait methods + their write loops), `scheduling_cycle.rs` (all phases, incl. the new tombstoned-stale-mapping delete from #36), `visibility.rs`, and `nonoverlap.rs`. CPU-only phases timed with `statements=0`. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: claude <claude@example.com> Co-committed-by: claude <claude@example.com>

* perf(pg-storage): batch the N+1 loops into set-based statements Replace every per-row .execute() loop the instrumentation flagged with one set-based statement, eliminating the O(chunks) round-trips per cycle: - insert_new_chunks: one bulk INSERT…SELECT via jsonb_to_recordset. A unique violation aborts the batch and is reported generically as ChunkAlreadyExists (no per-row culprit lookup); an unknown dataset is detected by RETURNING-count mismatch. - update_worker_set: one batch upsert over UNNEST(peer_id[], version[]) RETURNING id,(xmax=0); added = freshly-inserted ids. - record_routing_diffs / commit_new_ideal: one jsonb_to_recordset INSERT / upsert (commit keeps its existing sweep-delete). - mint_superseded_stale_mappings / resolve_flip_flops: flatten to parallel bigint[] pairs, one INSERT…ON CONFLICT DO NOTHING / one DELETE…USING UNNEST. - replay_confirmed_diffs: fully server-side — DISTINCT ON (chunk_pk) … ORDER BY worker_assignment_id DESC for last-write-wins, one data-modifying CTE (upsert non-empty + delete empty); no rows round-trip to Rust. PhaseTimer instrumentation retained; statement counts now reflect the batched queries. Measured on the reshuffle-sim multistep path at ~95K chunks: the five N+1 phases drop from ~95K statements/~85s combined to 1-2 statements/~5s. pg correction tests pass; clippy clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * perf(pg-storage): drop jsonb_to_recordset for native UNNEST+array_agg commit_new_ideal and record_routing_diffs now flatten the placement to two parallel bigint[] params and re-aggregate server-side (array_agg ORDER BY worker_id, GROUP BY) via a LEFT JOIN over all touched chunk_pks with COALESCE to '{}' — instead of serializing rows to a JSON string and parsing it with jsonb_to_recordset. One/two statements, no JSON build/parse, and it scales to millions of elements in two array params (a single bigint[] is far under the 65535 bind-parameter cap). The LEFT-JOIN-over-all-pks form preserves the old behavior for empty/removed holder sets. Dropped the now-unused DiffRow struct. insert_new_chunks keeps jsonb (now commented): its 5 scalar columns plus an order-sensitive ragged files[] make a native reconstruction high-complexity for a one-time load path; COPY is the escalation if the 10M-baseline JSON allocation ever bites. Addresses review feedback on #38. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * perf(pg-storage): pre-allocate flat-array binds, skip no-op ideal updates - commit_new_ideal / resolve_flip_flops: pre-size the flattened (chunk_pk, worker_id) bind vectors with the exact pair count, avoiding repeated reallocation of the per-cycle placement arrays. - record_routing_diffs: drop the redundant Rust-side sort; the stored array is already ordered by array_agg(... ORDER BY worker_id). - commit_new_ideal upsert: add WHERE worker_ids IS DISTINCT FROM EXCLUDED so an unchanged chunk produces no dead tuple / WAL. Correct because both sides are canonically ordered, so an unchanged holder set compares equal. Also trim the branch's comments (drop scale figures and restated-code lines). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(pg-storage): batch insert_new_chunks; surface underlying DB errors A single all-rows INSERT…SELECT over jsonb_to_recordset fails at scale — the jsonb payload can exceed Postgres' 1 GB value limit, and one statement can trip a statement_timeout (observed: ~30s then a generic "database error"). Insert in 10k-row batches inside one transaction so neither the payload nor a single statement's runtime scales with the input; still atomic (any batch error rolls the load back). ChunkAlreadyExists / dataset-not-found semantics unchanged. Also render StorageError::Database with `{0:#}` so the underlying cause (timeout, value-size, …) is shown instead of just the outermost context. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(pg-storage): batch the full-ideal write phases to bound bind-array size record_routing_diffs, commit_new_ideal, resolve_flip_flops and mint_superseded_stale_mappings flattened the whole ideal into one statement's bind arrays, which at large placements exceeds Postgres' wire/value limit and drops the connection ("broken pipe"). Write in 10k-chunk batches instead. record_routing_diffs / commit_new_ideal window by whole chunks so a chunk's holder set never splits across batches (their array_agg / PK depend on it); commit keeps a single trailing sweep-delete. All batches run inside the cycle's transaction, so the phase stays atomic. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * perf(pg-storage): intern dataset names; drop the full chunks map Two per-cycle allocation cuts from the storage review: - chunk_from_row took a fresh Arc::new(dataset_name.clone()) per row — one String allocation for each chunk despite only a handful of distinct dataset names. It now takes a pre-interned Arc; a DatasetInterner shares one Arc per name across a ...--------- Co-authored-by: Defnull <879658+define-null@users.noreply.github.com> Stacked PR **2 of 2** — **base this on #37** (the instrumentation), so the diff here is only the SQL rewrites. Merge #37 first. ## What Replace every per-row `.execute()` loop the instrumentation (#37) flagged with one set-based statement, eliminating the O(chunks) round-trips per cycle: - **insert_new_chunks**: one bulk `INSERT…SELECT` via `jsonb_to_recordset`. A unique violation aborts the batch and is reported **generically** as `ChunkAlreadyExists` (no per-row culprit lookup — deliberate, to keep it one statement); an unknown dataset is detected by RETURNING-count mismatch. - **update_worker_set**: one batch upsert over `UNNEST(peer_id[], version[]) … RETURNING id,(xmax=0)`. - **record_routing_diffs / commit_new_ideal**: one `jsonb_to_recordset` INSERT / upsert (commit keeps its existing sweep-delete). - **mint_superseded_stale_mappings / resolve_flip_flops**: flatten to parallel `bigint[]` pairs; one `INSERT…ON CONFLICT DO NOTHING` / one `DELETE…USING UNNEST`. - **replay_confirmed_diffs**: fully server-side — `DISTINCT ON (chunk_pk) … ORDER BY worker_assignment_id DESC` for last-write-wins, one data-modifying CTE (upsert non-empty + delete empty); no rows round-trip to Rust. `PhaseTimer` instrumentation retained; statement counts now reflect the batched queries. ## Measured (reshuffle-sim multistep, ~95K chunks) Per-cycle, the five N+1 phases drop from ~95K statements/~85s combined to **1–2 statements/~5s**. End-to-end wall clock for `--chunks-per-step 1000 --steps 5`: **241s → 53s (~4.5×)** — and the gap widens at 10M scale and on a real networked DB (round-trip latency is exactly what batching removes; the bench used localhost `fsync=off`). ## Verification Full feature-gated suite **220 passed / 0 failed** (incl. the #36 overcommit regression test, which drives the batched `insert_new_chunks`/`update_worker_set`); clippy clean; sim runs end-to-end. Note: the sim's step-4/5 report differs run-to-run on **both** branches (pre-existing nondeterminism from unordered `SELECT`s feeding order-sensitive tie-breaks) — not introduced here. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: claude <claude@example.com> Co-committed-by: claude <claude@example.com>

Rebase of the NET-681 multi-step scheduling feature onto master @ e7aec30 (v3.11.0). Flattened from the incrementally-validated staging branch (defnull/net-681-multi-step-scheduling-2-squashed), which merged master forward one base commit at a time (#51 → #52 → #53 → #56). Includes: - Placement-aware multistep scheduler (Stage 1 ideal + Stage 2 reconcile), reusing the stored placement so only new data moves between cycles. - Single-step scheduler integrated onto master's replication back-off model (ensure_minimum / fill_to_cap; no panic-on-shortage). - reshuffle_sim unified across stateless + multistep paths, with a dedicated restricted-dataset model for version-restricted scheduling. Tree is byte-identical to the validated staging tip.

…(#59) * assert(weight): validate prepare() receives chunks sorted by (dataset, block) weight::prepare requires its input sorted by (dataset, block): chunk_by groups only consecutive same-dataset chunks and the segment cursor only advances. Violating it yields wrong weights and silently dropped chunks — a real risk since the Postgres fetch_active_chunks has no ORDER BY. Make the precondition explicit with a single debug_assert! using slice::is_sorted_by on a (dataset, block) tuple. The whole check (call + closure) lives inside debug_assert!, so it is compiled out of release builds — zero production cost, no extra state. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Defnull <879658+define-null@users.noreply.github.com> Reviewed-on: http://localhost:3000/defnull/network-scheduler/pulls/59 Co-authored-by: claude <claude@example.com> Co-committed-by: claude <claude@example.com>

toschoosqd · 2026-06-26T15:13:43Z

+        let worker_chunks = &mut self.worker_chunks;
+        let chunk_workers = &mut self.chunk_workers;
+
+        walk_ring(rings, chunk, tag, |worker| {


Are code and document out of sync? According to capacity-aware-scheduling.md:26, this should not happen: “When a new replica's home is full, leave it unplaced this cycle.”

They are, will fix.

toschoosqd · 2026-06-26T15:26:07Z

+        // Weight key is the chunk's own `(dataset, id)`, NOT the storage pk (an unrelated id space).
+        let dataset_key = |c: &Chunk| ((*c.dataset).clone(), (*c.id).clone());
+
+        // `prepare` may reorder / drop chunks, so key the result and realign below; it takes


Aren't we reintroducing chunks that were dropped by the weight strategy?

That's fixed in in the upcoming PR.

toschoosqd · 2026-06-30T08:48:26Z

    sorted_workers.extend(workers.iter().filter(|w| !w.reliable()));

+    // Every worker starts empty with the full per-worker budget.
+    let capacities = vec![config.worker_capacity; sorted_workers.len()];


The change looks innocent. But would it be possible to add a regression test to make sure the behaviour remains the same?

Can you elaborate what behaviour you would like to maintain? In the multi-step scheduler that wouldn't be the case.

toschoosqd · 2026-06-30T09:56:52Z

+        debug_assert!(
+            chunks.is_sorted_by(
+                |a, b| (&a.dataset, a.blocks.start()) <= (&b.dataset, b.blocks.start())
+            ),


Shouldn't we also add a test to make sure that this assertion still aligns with the assertion in controller.rs:203 in production (non-gated) code?

toschoosqd reviewed Jun 16, 2026

View reviewed changes

define-null force-pushed the defnull/net-681-multi-step-scheduling-2 branch from 33a8847 to e18ee8d Compare June 24, 2026 17:31

define-null force-pushed the defnull/net-681-multi-step-scheduling-2 branch from e18ee8d to 7b2adb1 Compare June 24, 2026 17:32

define-null marked this pull request as ready for review June 24, 2026 17:32

define-null requested a review from toschoosqd June 24, 2026 19:33

claude and others added 2 commits June 25, 2026 08:52

Fix ordering issue for Postgres backend

99ff300

define-null changed the title ~~feat: Multistep scheduler~~ feat: WIP Multistep scheduler Jun 25, 2026

toschoosqd reviewed Jun 26, 2026

View reviewed changes

toschoosqd reviewed Jun 30, 2026

View reviewed changes

define-null merged commit f6f599e into master Jul 8, 2026
3 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat: WIP Multistep scheduler#37

feat: WIP Multistep scheduler#37
define-null merged 3 commits into
masterfrom
defnull/net-681-multi-step-scheduling-2

define-null commented Jun 12, 2026 •

edited

Loading

Uh oh!

Uh oh!

toschoosqd Jun 16, 2026

Uh oh!

define-null Jun 16, 2026

Uh oh!

Uh oh!

toschoosqd Jun 26, 2026

Uh oh!

define-null Jun 29, 2026

Uh oh!

toschoosqd Jun 26, 2026

Uh oh!

define-null Jun 29, 2026

Uh oh!

toschoosqd Jun 30, 2026

Uh oh!

define-null Jun 30, 2026

Uh oh!

toschoosqd Jun 30, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

define-null commented Jun 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is this PR about?

MVCC chunk lifecycle (src/scheduler_storage/, migrations/)

Multi-step (reconciliation) scheduling (src/multistep_scheduler.rs)

Simulation & property-based testing (src/multistep_scheduler/sim/)

Supporting changes & docs

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

define-null commented Jun 12, 2026 •

edited

Loading