Archive ephemeral oplogs from a sweep instead of a scheduled action

[kmatasfp]

Every agent invocation that ends in Idle | Failed | Exited registers a ScheduledAction::ArchiveOplog row, and that registration is a synchronous write to remote storage on the oplog commit path, taken while holding update_state_lock. A chaos run that injected 500ms of latency on the database backing the executor's scheduler storage drove ephemeral p50 from 180ms to 31s. Invocation throughput pins at pool_size / write_latency and no amount of executor concurrency moves it. The same fault against this branch leaves ephemeral at its baseline 170ms; see Verified on a cluster.

The row exists to answer one question: which agents have oplog entries sitting in the wrong layer. That answer is already in the storage, so this replaces the row with a scan.

What this does

Stops ephemeral agents registering the action. schedule_oplog_archive_if_needed returns early for AgentMode::Ephemeral. The write leaves the invocation path and the scheduler table stops accumulating a row per ephemeral invocation. Durable agents are untouched and still register as before.

Adds an on-executor sweep that covers what the guard stops doing. Without the registration, an ephemeral oplog is drained only by archive_ephemeral_oplog on teardown, and that is a detached tokio::spawn whose result is discarded and which is not in the executor's join set. A pod that crashes, shuts down, or simply loses that task strands the entries with nothing scheduled to move them. OplogSweeper finds them by paginating the oplog layer itself.

Adds IndexedStorage::scan_stable. A walk that survives the caller deleting what it walks, implemented across all five backends. See below for why scan could not be used.

The module is a work list, not a mover

MultiLayerOplog::archive already moves a prefix down one layer, handles the primary hop and the lower hops, dispatches on agent mode, and reports whether a layer below still holds entries. sweep_once is the body of the existing ArchiveOplog handler with the row replaced by a scan; it still calls open_oplog and try_archive_blocking, so the archiving semantics are the ones already in production.

That includes the exclusion. open_oplog builds a suspended Worker, which is what the scheduled action relies on to avoid racing a live writer, and the arithmetic favours keeping it: the scheduled action paid one construction per registered row, and for ephemeral agents every row passed its index gate, so one per invocation. The sweep pays one per agent that genuinely has a tail to move, which is rare because the size trigger at entry_count_limit already handles busy agents.

// Derives its routes from the layer stack lib.rs already built.
// Pure: no I/O, no task, no runtime.
pub fn over_layers(config, indexed_storage, archives, shards, components, worker_access) -> Arc<Self>;

// Ticks until the token is cancelled. Spawned into the executor's JoinSet.
pub async fn run(self: Arc<Self>, shutdown: CancellationToken);

One pass is sweep_once(&self, shutdown: &CancellationToken) -> SweepReport: deterministic, infallible, and the whole test surface.

Routes are derived from the layer stack rather than hand-assembled. One default method was added to OplogArchiveService:

fn scan_namespace(&self, _agent_mode: AgentMode) -> Option<IndexedStorageMetaNamespace> { None }

The compressed archive answers with its level, the blob archive keeps the None. A layer that cannot enumerate its own keys can receive an archive step but never be the source of one, so the bottom of any stack is a target only.

Everything that decides is a private free function with no self, no async and no storage: parse_agent_id, owns, assess, triage, tally, merge. The driver that talks to storage is thin, and ten of the module's thirty-three tests need no runtime at all.

A tick has two phases, and they must not interleave

The scan phase walks the namespace, decides every key it sees, and touches nothing. Only when the walk has finished does the archive phase run. That ordering is forced: an archive step ends in a drop_prefix that removes the key the scan was walking, so archiving mid-walk would delete keys out from under the cursor.

Deciding is cheapest first. Parsing the key and testing the shard need only the assignment already in memory. Residency comes next, also in memory, because on a busy executor most scanned keys belong to agents this pod is running and probing their index would be one storage read each per tick to learn what ActiveWorkers answers for free. Only what survives that gets a storage read, and the component lookup that resolves the target environment happens only for an agent about to be archived.

An agent is archived when its last oplog index is unchanged between the two scans that saw it. Since a scan resumes where it stopped, a key is visited once per pass rather than once per tick, so the gate is one interval only while the namespace fits inside one tick's budget, and one full pass over it otherwise.

Scanning a layer you are deleting from

IndexedStorage::scan cannot page this walk. Its cursor is a position on every backend that has one, an OFFSET in Postgres and SQLite and an iteration count in memory, so deleting a key behind the cursor shifts everything after it down and the next page steps over exactly that many keys nothing has looked at.

scan_stable resumes by seeking instead:

pub enum ScanResume {
    /// The last position a backend reached in whatever ordering it walks.
    Marker(String),
    /// The backend's own iteration cursor, for one that has no order to seek in at all.
    Cursor(ScanCursor),
}

Ordered backends return the last key they handed back. The multi-file SQLite backend returns the file it finished, because its keys are ordered only within a file. Redis returns its own SCAN cursor, which already tolerates deletion. The contract is that a key present for the whole walk is handed back at least once, and a key the caller deletes may or may not be.

Resume tokens live in an in-process map and are never persisted, so a token cannot outlive the backend that minted it.

What a tick costs

Bound	Default	Caps
page_size	128	Keys per scan call
max_concurrency	4	Agents archived at once, sharing the indexed-storage connection budget with the invocation path
max_archives_per_tick	256	Archive steps before the tick stops and keeps its cursor
max_scanned_per_tick	4096	Keys examined, so a tick cannot walk a namespace far larger than the work in it
max_tracked_agents	100000	Backstop on the tracking table
max_tick_duration	30s	Wall clock a tick may hold indexed-storage concurrency before stopping at its next boundary
max_backoff_intervals	8	Intervals the loop may wait after a tick that hit its deadline

A tick that hits a bound keeps its resume token and picks up from the same place next time, so work is deferred rather than dropped.

Ephemeral agent ids are unbounded, since an invocation with no phantom id gets a fresh Uuid::new_v4(), and an agent drained by its own teardown never appears under that id again. Tracking entries are therefore stamped with the scan pass that touched them and dropped when a pass completes without seeing them, so max_tracked_agents is a backstop rather than the mechanism.

Three metrics: oplog_sweep_outcome_total by route and outcome, oplog_sweep_tick_seconds, and oplog_sweep_truncated_total. In steady state the sweep does nothing: it scans the in-flight set, decides every one of them Resident in memory, and archives none of them. oplog_sweep_outcome_total{outcome="archived"} climbing on a healthy cluster is the signal that teardown drains are being lost.

The sweep stands down when the store is slow

The count budgets say how much a tick does. They do not say how long it holds anything, and under a degraded store those stop being the same quantity.

PostgresIndexedStorage puts one semaphore in front of its pool, max_concurrent_ops, and every caller shares it: the primary oplog on the durable commit path, the compressed archive an ephemeral agent writes through, and this sweep. golem-dev sets it to 64 against a 96-connection pool, so the gate binds before the pool does. A tick sized in operations is brief when an operation is milliseconds and runs for minutes when an operation is seconds. Without a time bound the sweep therefore stops being a periodic user of that semaphore and becomes a near-continuous one, at exactly the moment the invocation path can least afford it.

This is measured, not reasoned about. Chaos scenario S23 adds 500ms of one-way latency to the indexed-oplog cluster. It was run twice against the same image, the same driver commit and the same manifest, with one config flag between them:

	sweep off	sweep on
executor aborts	2	7
promise during the fault	2499x slower, still serving	0% of baseline, served nothing
scheduled actions fired more than once	11	19
exactly-once violations	0	0

Two caveats a reader should have. Each configuration was run once, so 2 aborts against 7 is directional rather than statistically settled. And the durable and scheduled latency percentiles look better in the sweep-on run purely because fewer operations survived to be measured, so those columns are not comparable between the two runs and are left out above.

Nothing was lost in either run — exactly-once held across 14k keys both times. The failure mode is availability, not correctness, and the double fires are downstream of the aborts rather than an independent defect.

So a tick now carries a deadline as well as its budgets, and the loop backs off while that deadline keeps being hit, doubling the wait to a cap and resetting on the first tick that finishes inside it. The sweep has no other way to know what the storage layer is doing; this is the one signal it can read about itself.

The deadline is enforced by cancelling the tick's own token, a child of the shutdown token. That is deliberate reuse rather than a new mechanism: every boundary in sweep_once that already tests for shutdown becomes a place the tick can stop for time too, each already marks the report truncated, and the invariant shutdown is written around — an archive step is never cut between its append below and its drop above — is inherited for free.

Deferring archiving costs nothing durable. The work list is the layer itself, so a skipped tick leaves the same agents to be found later; only the latency of moving a stranded oplog grows, against a default archive_interval of a day.

Both numbers are untuned and labelled as such in the config docs. 30s is half the default interval, so the duty cycle stays under half even when every operation is slow, before the backoff takes it lower.

Shutdown and failure

run is spawned into the executor's JoinSet, which the server binaries drain with join_next().await. Cancellation is observed between routes, between scan pages, before each component the archive phase has to resolve, and before each agent it reaches, but never inside one. So a shutdown never interrupts an archive step between its append to the layer below and its drop from the layer above; an agent already under way is finished, all of its layers, before the loop returns.

A sweep holding no shard assignment, or one whose shard count is still zero, does nothing and reports itself unassigned rather than routing an agent through a zero divisor.

A non-transient indexed-storage error inside a tick panics through retry_storage_op, and this workspace builds with panic = "abort", so it takes the process down. That is how every other oplog operation already behaves; the sweep is another caller of the same code, not a new failure mode.

Which agent modes are swept is not configurable

pub const SWEPT_MODES: &[AgentMode] = &[AgentMode::Ephemeral];

Ephemeral is not optional, because the guard took the per-invocation registration off the commit path and this sweep is now the only thing that moves an ephemeral oplog a crashed pod stranded.

Durable is absent rather than off, and there is deliberately no setting that turns it on. IndexedStorage::append is a plain INSERT against a table with PRIMARY KEY (namespace, key, id), a unique violation is not classified transient, and retry_storage_op turns that into a panic under panic = "abort". Re-appending a prefix into an indexed layer therefore takes the pod down, and an archive step interrupted between its append and its drop_prefix leaves exactly that prefix to be re-appended. Blob targets are safe, since their append is a put at a path keyed by the chunk's last index and a repeat rewrites identical bytes. With the default stack the ephemeral hop targets blob and the durable primary hop targets the indexed compressed layer.

There is a second gate on the same thing. lib.rs hands over_layers the archive stack alone, and the primary oplog service is not an OplogArchiveService, so level 0 is never a route source. A durable mode added to the list would sweep the compressed hops and leave the level-0 hop with ScheduledAction::ArchiveOplog.

Covering durable therefore needs the append to become an upsert and a layer passed in that can enumerate the primary's keys. Both are code changes, which is the point of keeping the list out of OplogSweepConfig. A test asserts no route the sweeper builds names a mode other than ephemeral.

The durable per-invocation registration stays exactly as it is.

enabled remains, as the operational lever that stops the whole sweep without a redeploy. It is read in two places, because the sweep and the registration it replaces are one switch: it stops the background loop, and it stops the guard in schedule_oplog_archive_if_needed from suppressing the ephemeral registration. So off is the behaviour that preceded this PR rather than a third state in which an ephemeral oplog has no mechanism behind it at all. An integration case asserts the rows come back.

Not fixed here

Moving a prefix reads the whole source layer into one Vec, written out twice, in EphemeralOplog::background_transfer and in BackgroundTransfer::run. The sweep calls whichever applies rather than adding a third copy, so an agent costs it exactly what archive_ephemeral_oplog already costs on teardown. What the sweep adds is a ceiling on how many run at once. Bounding the read itself means chunking both copies, or unifying them first, and belongs in its own change.

Archiving an agent builds a suspended Worker that nothing evicts, because stop_if_evictable matches only a running instance. Each archived agent therefore keeps an ActiveWorkers entry for the life of the pod. It is inherited from ScheduledAction::ArchiveOplog, which built the same worker far more often, and it is why the sweep drains an agent fully in one visit rather than leaving a hop for a later tick.

No single hop knows that an agent's whole stack has drained, so WorkerService::remove_cached_status, which the scheduled action calls when more == false, has no equivalent here. Ephemeral agents write no cached status, so this only matters once durable routes are enabled, and the code says so at the call site.

Testing

The guard is proven red then green by counting rows in the executor's own scheduler storage: 30 ephemeral invocations leave 31 rows on unmodified 1.5.x and 0 with the guard. A durable agent in the same component is the control, so an empty table cannot mean the guard stopped scheduling for every mode. A third case runs the same workload with enabled = false and asserts the rows return.

The sweep carries 37 tests at the module's interface, with no executor process. The pure functions are tested directly; above them, tests build the real layer stack over in-memory indexed and blob storage and assert on what moved between layers rather than on what the report claimed. Notable cases: an agent is archived only after its index survives a scan; a running agent and an agent on another shard are left alone; a completed pass drops the tracking entries for agents that left the layer; every budget truncates and resumes; a zero-shard assignment archives nothing; no route is ever built for a durable layer; cancelling the token stops the loop after it has archived. The backoff is a pure function tested directly, including its cap, its reset, and a zero cap read as one so the wait can never multiply to nothing and spin the loop.

scan_stable is pinned by a cross-backend test that deletes keys behind the cursor mid-walk and asserts every surviving key is still handed back, run against every indexed-storage backend, plus a test that the multi-file SQLite walk crosses its files a page at a time. The in-memory page selection was checked against the naive implementation over 200000 randomised single-page cases and 20000 full multi-page walks.

Both pre-existing archive tests in scalability.rs still pass. Full worker-executor unit suite is 568 passing, cargo fmt and cargo clippy --all-targets -- -D warnings are clean, and the generated config files are regenerated.

Verified on a cluster

The branch was deployed as a pre-release build and the two scenarios that found the problem were rerun against it, each on a freshly reset cluster. Same 500ms injected latency, same workload, same duration; only the platform build differs.

Ephemeral alone. This is the control that originally established the cause, by showing ephemeral slowed down with no other stream running to blame.

	before	after
baseline p50	180ms	170ms
fault p50	31,217ms	170ms
slowdown	173.4x	1.00x
throughput during fault	4,714	10,274

The delay is now invisible to ephemeral: p50 during the fault equals baseline, p90 is 200ms against 201ms, and throughput holds full rate instead of collapsing to 45%. The scenario's unexpected-slowdown: ephemeral finding cleared, leaving no findings at all.

All four streams. The full scenario runs durable, ephemeral, promise and scheduled together.

stream	before	after	before	after
ephemeral	46,037ms	13,844ms	235x	74x
durable	39,285ms	24,288ms	644x	392x
promise	86,398ms	56,260ms	939x	612x
scheduled	101,659ms	87,636ms	1640x	1413x

Throughput during the fault rose on every stream. That every stream improved, and not only the one this change touches, follows from the fault being queueing on a connection pool the four share: ephemeral ran the highest invocation rate and so was the largest single consumer of that pool, and removing its writes frees capacity for everything else on it.

Ephemeral's remaining 74x here is queueing behind the three streams that still register the action, not ephemeral reaching that storage. Running it alone gives 1.00x, which is what separates the two. Closing that gap means the durable write under Not fixed here.

Correctness was unaffected in both runs: every key exactly once with no findings, read-back consistent, every scheduled action fired exactly once, no executor restarts and no out-of-memory kills. One incidental improvement, scheduled registrations that were indeterminate but fired went from 104 to 0.

[vigoo]

Copying the agent review directly here, they seem to be valid:

• [P1] Redis duplicates can satisfy the quiet gate within one tick. Redis SCAN may return a key multiple times, but probe_agent compares only the index and ignores Seen.pass. A duplicate later in the same pass therefore changes Wait to Move. The later pending.dedup() is too late. This bypasses the intended interval and can race a previous shard owner still writing. Require the remembered sighting to come from an earlier pass.
• [P1] Deleted or undeployed components make stranded oplogs permanently unarchiveable. environment_of calls get_metadata(component_id, None), which only returns a currently deployed, non-deleted component. If teardown archival is lost and the component is deleted before the second sweep, every subsequent tick returns Unaddressable. The replaced scheduled action retained the OwnedAgentId and could open the worker using its pinned, deletion-tolerant revision.
• [P2] Multi-SQLite rescans and sorts every historical database file for every page. Each page calls namespace_db_files again before applying the resume marker. Files are never removed, and ephemeral IDs create one file per invocation. A full pass over FF files with page size PP therefore costs roughly O(F2log⁡F/P)O(F² \log F / P), with synchronous filesystem work on a Tokio thread. This defeats the sweep’s bounded-work guarantee.
• [P2] The archive step cap can permanently strand the remaining layers. When MAX_ARCHIVE_STEPS is reached with more == true, the memo is still forgotten and the operation is reported as archived. Production open_oplog has inserted a suspended active worker, so later sweeps classify it as resident and never continue. Any supported configuration with more than 16 movable hops triggers this deterministically.
• [P2] Tick budgets are actually per route. sweep_once gives every route fresh scan/archive counters, although the config documents them as per-tick limits. Configurations with multiple compressed source layers multiply database work by the route count.

[kmatasfp]

All five addressed in 75fa1c8. Each has a regression test that I checked goes red with the fix backed out.

[P1] Redis duplicates satisfying the quiet gate within one pass. Correct, and worse than the write-up in one respect: the two sightings need not even be in the same tick, since a pass survives truncation, so dedup was never going to catch it. The gate now compares the sighting's pass as well as its index:

fn assess(remembered: Option<Seen>, current: OplogIndex, pass: u64) -> Verdict {
    match remembered {
        Some(seen) if seen.index == current && seen.pass < pass => Verdict::Move,
        _ => Verdict::Wait,
    }
}

No latency cost: a pass visits each key once, so the sighting a healthy agent qualifies on already came from the previous pass. probe_agent still restamps to the current pass before returning Archive, so an archive that fails keeps the gate it passed.

[P1] Deleted components stranding oplogs permanently. environment_of now falls back to get_metadata(id, Some(ComponentRevision::INITIAL)). get_component_metadata is documented to return deleted components, and a component does not move between environments across revisions, so revision zero carries the same answer the current one would. Deployed is still tried first, since that is the lookup the rest of the executor keeps a cache warm for and most swept agents belong to components that are still there.

[P2] Multi-SQLite rescanning every file per page. Three changes: the listing is cached per meta-namespace prefix, read_dir moved to spawn_blocking, and the resume marker is found by partition_point rather than by walking every file before it. Staleness is not left to a TTL, because this process is the only writer to its own directory: storage_by_db_name drops the cached listings when it creates a file, tested inside the cache-miss path so an evicted connection for a file that already exists does not invalidate anything. The 10s TTL is only a backstop for something outside the process. multi_sqlite_scan_stable_sees_files_created_after_a_walk covers it.

[P2] The archive step cap stranding the remaining layers. The bound is now derived from the stack the sweeper was built over (archives.len() + 2) instead of a fixed 16, so a deep-but-valid stack can never reach it and the guard does only what it was for: stopping a layer that miscounts more. Reaching it now returns ArchiveFailed, keeps the tracking entry and warns, rather than forgetting the agent and reporting it archived.

One consequence worth flagging: Outcome::Archived { more: true } was reachable only from that cap, so with this it becomes unconstructible. I collapsed the variant to Outcome::Archived and dropped the drained counter, which would otherwise have been permanently equal to archived. That changes the oplog_sweep_outcome_total label set, which is only in a dev image so far. Say the word if you would rather keep the label.

[P2] Tick budgets being per route. sweep_once now builds one TickBudget and hands each route an even share of what is left rather than the whole of it. Sharing the remainder rather than a fixed slice is what stops that wasting budget: a route that finds nothing leaves its share to the routes behind it. The route order is load-bearing, lowest source level first so a tick never hands entries to a layer it is about to drain, so the shares move and the order does not. Config docs on both knobs now say they are tick budgets rather than per-route ones.