perf(durable-streams-rust): batch memory-mode meta sidecar flushes into a periodic sweep

[balegas]

Fixes #4691.

Problem

In --durability memory, every append fell back to StreamState::schedule_meta_flush: a per-stream tokio::spawn + 100 ms sleep + spawn_blocking running a full .meta sidecar rewrite (JSON serialize + File::create + rename, plus parent-dir inode contention). Whenever the per-stream inter-append gap is ≥ the 100 ms debounce (any rate ≤ ~10 ops/s/stream — the common case at high stream cardinality), every stream pays a timer task plus a full sidecar rewrite every 100 ms. Wal mode stopped paying this in #4675 (the checkpoint batches sidecar writes), so memory mode — supposedly the cheaper path — burned ~5x wal's CPU at identical load.

Fix

Give memory mode the same batched treatment (packages/durable-streams-rust):

• Store::mark_meta_dirty(st) — appends (memory branch) and TTL read touches now only CAS meta_dirty and, on the false→true edge, push the stream into a store-level meta_sweep set (deduped: at most one entry per stream per cycle). No timer task, no per-append write.
• Store::sweep_meta_once() — drains the set and writes every still-dirty sidecar in one spawn_blocking pass. It skips hard-deleted streams (Arc identity check against the streams map), so a pending flush can no longer resurrect an unlinked sidecar — a race the old debounce had.
• spawn_meta_sweeper (main.rs) — a single 1 s ticker drives the sweep in both durability modes (wal appends still flush via the checkpoint; only TTL read touches use the sweeper there).
• StreamState::schedule_meta_flush is deleted. Durable flush paths (close/delete/shutdown write_meta_sync(st, true) call sites) are unchanged.

The sidecar's producer/access state is documented as a non-durable, lagging flush; the lag bound moves from 100 ms to the 1 s sweep cadence — exactly the trade wal mode made in #4675.

Validation (local ds-bench, kind, server capped at 2 CPU)

ds-bench sustained suite (10 append/s per stream, 256 B payloads, 90 s window, 1 fleet pod), baseline image = main @ 640509c, fix image = this branch:

streams	config	cpu_mean before	cpu_mean after	p99 before (ms)	p99 after (ms)
10	memory	4.3	1.1	3.363	2.627
100	memory	31.3	5.6	9.455	6.191
150	memory	50.7	8.1	11.639	7.007
10	wal	2.6	2.2	16.607	8.455
100	wal	10.2	8.5	18.799	15.087
150	wal	12.6	11.1	23.951	13.087

Memory-mode CPU drops ~4–6x at fixed load and lands below wal at every stream count — the issue's expected outcome (memory does strictly less work per append). Throughput held at the offered rate in every cell (stable=True, zero errors); wal cells are unchanged within noise (its append path doesn't touch this code — only TTL read touches moved to the sweeper).

bench-latency (in-repo host harness, --quick, after the fix): memory-mode write ack mean 0.29–0.51 ms across 1–64 KiB payloads — within the ~0.2–0.5 ms band the issue expects; wal unchanged (2.5–3.3 ms, macOS F_FULLFSYNC dominated).

Tests

• store::meta_sweep_tests::mark_dedupes_and_sweep_flushes — CAS dedupe, sweep persists pending producer state, clears the flag, second sweep is a no-op.
• store::meta_sweep_tests::sweep_skips_hard_deleted_stream — a sweep after a hard delete does not resurrect the sidecar.
• handlers::memory_mode_tests::memory_append_defers_sidecar_to_store_sweep — a memory-mode append no longer flushes the sidecar within the old 100 ms debounce window; the batched sweep is what persists it.
• Full cargo test (105) and the protocol conformance suite (326 tests) pass; clippy clean.

🤖 Generated with Claude Code

https://claude.ai/code/session_01SmhCex16r9JUPxHunf7Roy

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
2296	1	2295	41

View the top 1 failed test(s) by shortest run time

test/runtime-dsl.test.ts > N: wake primitives verification > N4: ctx.agent.run receives the wake payload and performs a second run on child completion

Stack Traces | 30.1s run time

Error: Timeout (30000ms) waiting for completed runs=2 on /idle-wake-parent-n3/wake-agent-1
[
  {
    "args": {},
    "entityType": "idle-wake-parent-n3",
    "operation": "insert",
    "type": "entity_created"
  },
  {
    "from": "/principal/system%3Aruntime-dsl-test",
    "operation": "insert",
    "payload": "spawn wake-agent-child-1",
    "type": "inbox"
  },
  {
    "key": "run-0",
    "operation": "insert",
    "status": "started",
    "type": "run"
  },
  {
    "key": "step-0",
    "operation": "insert",
    "status": "started",
    "stepNumber": 1,
    "type": "step"
  },
  {
    "key": "msg-0",
    "operation": "insert",
    "status": "streaming",
    "type": "text"
  },
  {
    "delta": "spawned:wake-agent-child-1:wake.type=inbox",
    "key": "msg-0:0",
    "operation": "insert",
    "runId": "run-0",
    "textId": "msg-0",
    "type": "text_delta"
  },
  {
    "key": "msg-0",
    "operation": "update",
    "status": "completed",
    "type": "text"
  },
  {
    "finishReason": "stop",
    "key": "step-0",
    "operation": "update",
    "status": "completed",
    "stepNumber": 1,
    "type": "step"
  },
  {
    "finishReason": "stop",
    "key": "run-0",
    "operation": "update",
    "status": "completed",
    "type": "run"
  },
  {
    "key": "child:idle-wake-child-n3:wake-agent-child-1",
    "manifest": {
      "entityType": "idle-wake-child-n3",
      "entityUrl": "/idle-wake-child-n3/wake-agent-child-1",
      "id": "wake-agent-child-1",
      "key": "child:idle-wake-child-n3:wake-agent-child-1",
      "kind": "child",
      "wake": {
        "includeResponse": true,
        "on": "runFinished"
      }
    },
    "operation": "insert",
    "type": "manifest"
  }
]
 ❯ waitForHistory test/runtime-dsl.ts:666:11
 ❯ test/runtime-dsl.ts:789:35
 ❯ test/runtime-dsl.test.ts:6315:29

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}