fix: restart resumes the process loop and the pulse emitter

[jbiskur]

Problem

restart() calls stop(true), which clears running, empties the buffer and stops the pulse emitter. The fetch loop revives itself from restartTo:

this.running = true
return this.loop()

But the process loop is a while (this.running) loop and only start() called startProcessLoop(). A pump restarted while it was delivering keeps pulling events and never delivers or checkpoints again — and stops pulsing, so it looks dead to the control plane while its host still reports healthy.

This is the defect written up on 2026-04-14 in Usable fragment 28875215-003d-4770-9d44-8e3488d25246.

Production impact

It stalled two data-pathways pathways for 26 days (2026-08-02 → 2026-08-28):

Pathway	Tenant	Effect
417b8dd9-b6a3-4ce9-b254-6ea12ca97d57	memory-mesh	26-day backlog, drained in 2.4 h at 98% DB CPU
5a32bd1c-9d0c-4aef-a813-4fee60280f8e	flowcore-saas	nothing delivered, assignment reported active

A scoped replay asked for one flow type; the consumer restarted every source, and every restarted pump went quiet.

Change

• ensureProcessLoop() starts the delivery loop whenever the pump is running with a processor and no loop is live. The fetch loop calls it once per iteration, so a loop that exits for any reason comes back within one iteration.
• processLoopRunning guards against a second loop when the existing one is only parked in reserve() — two loops would race over the same buffer.
• A loop that exits while the pump is already running again (the restart landed mid-batch) restarts itself.
• The restart path calls pulseEmitter.start() again. It is already idempotent.

Tests

Three new cases in data-pump-restart.test.ts, all of which fail on the previous behaviour:

1. delivery continues after a restart() issued from inside the handler
2. no second process loop when one is only parked (an event is delivered once, not twice)
3. the pulse emitter is started again

Suite: 57 pass / 0 fail. tsc --noEmit and prettier --check clean.

🤖 Generated with Claude Code

[jbiskur]

Follow-up independent re-review — current verdict: APPROVE

Reviewed head: c5f5fff2ea1bbafbcfebc8417d1efae2aa7a1f3c
PR head verification: refs/pull/90/head resolved to that exact SHA. It is the direct child of the previously reviewed 76bf591651ca104c4ae96e93f1e31316c4653076.
Method: fresh detached checkout, full follow-up delta and surrounding lifecycle code inspected independently. No code changed. Do not treat the historical REQUEST CHANGES below as the current verdict.

Current verdict: APPROVE. All five earlier findings are resolved.

Resolution of the five earlier findings

1. HIGH pull-mode restart stall — resolved. Public and internal process waiters are now separated. stop(true) wakes stale generation-keyed process waiters without exposing the temporary running = false state to a public reserve() consumer. The new README-style pull-consumer test proves it remains parked across restart and receives replayed work.
2. MEDIUM stuck-handler stop() + start() wedge — resolved. Generation ownership replaces the global running flag; lifecycle changes synchronously invalidate and release old ownership, a new generation starts even while the old handler is unresolved, and stale continuations cannot acknowledge or clear newer ownership. The new stuck-handler test proves replacement delivery and checkpoint safety.
3. LOW retry-backoff bypass — resolved. A generation-scoped backoff marker now prevents ensureProcessLoop() and startProcessLoop() from starting early; lifecycle stop cancels and clears the timer. The new backoff test proves repeated ensure calls do not bypass the delay.
4. LOW missing duplicate-loop test — resolved. The explicit starts at most one process loop for a generation test is present and exercises repeated ensureProcessLoop() calls.
5. LOW inert stop(_isRestart) parameter — resolved. It is restored as stop(isRestart = false) and now controls whether the public waiter is notified. The generated declaration remains stop(isRestart?: boolean): void.

Exact verification evidence

• Local targeted restart suite: 17 passed / 0 failed, 50 assertions.
• Local full suite: 91 passed / 0 failed, 333 assertions.
• bunx tsc --noEmit: clean, exit 0.
• bunx prettier --check .: clean, exit 0.
• bun run build: successful; ESM and declarations generated (dist/mod.js 73.09 KB).
• git diff --check: clean.
• GitHub workflow 33518256856: build SUCCESS, integration-test SUCCESS.
• GitHub PR state when re-review began: head c5f5fff2ea1bbafbcfebc8417d1efae2aa7a1f3c, OPEN, MERGEABLE, review gate REVIEW_REQUIRED. It was then merged externally by jbiskur at 2026-09-01T14:22:28Z as 6acec00e6e90c382168798af99a059d61432f801; Allora made no merge call.

Residual risks — non-blocking and pre-existing

• A handler that hangs without a lifecycle change still has no timeout/watchdog.
• Concurrent public reserve() calls still share one waiter slot.
• Direct external use of stop(true) can leave a public waiter parked without a following restart; the internal-purpose semantics remain undocumented.
• restart(state, stopAt) still does not rebuild stopAtState, as already documented.

---

Historical review at 76bf591 — superseded by the APPROVE verdict above

Independent review — PR #90

Reviewed head: 76bf591651ca104c4ae96e93f1e31316c4653076 (verified against refs/pull/90/head at review time)
Method: fresh detached worktree at that exact commit. No prior conclusions reused. No code changed.

Verdict: REQUEST CHANGES. The push-mode fix is correct and well covered. One PR-introduced regression reproduces the same silent-stall class in the public pull-mode API, and a second wedges delivery after stop() + start(). Both are reproduced against the base commit for comparison.

Command results (worktree at 76bf591)

Command	Result
bun test test/tests/data-pump-restart.test.ts	13 pass / 0 fail, 40 expect()
bun test (full suite)	87 pass / 0 fail, 323 expect(), 8 files
bunx tsc --noEmit	clean, exit 0
bunx prettier --check .	clean, exit 0
bun run build (tsup esm + dts)	success, dist/mod.js 71.56 KB, dist/mod.d.ts 22.58 KB

What is correct

• The generation guard is sound. stop() increments processLoopGeneration synchronously, before any await, so a handler that resolves after a restart can never pass the check at data-pump.ts:681 and can never acknowledge a replayed copy of the same event ID. Reading restart() → stop() confirms no window exists.
• Push-mode mid-batch restart now resumes delivery. Confirmed independently.
• pulseEmitter.start() is idempotent (pulse.ts:56 returns early when interval or startTimeout is set), so the restart call is safe.
• restart() setting isLive = false is a real correctness gain. Without it the snapshot reports live while the pump replays history.
• The stale reOpen() timer scheduled before a restart is harmless. It filters on deliveryId (data-pump.ts:566), and replayed buffer items carry a new one.
• Tests use real TimeUuid.fromDate(...) values, and the checkpoint test asserts the durable setState payload, not a mock call count.
• One clarification on the PR description: the parked-in-reserve() case already recovered on the base commit, because the post-restart fetch wakes waiterEvents at data-pump.ts:374. The genuine defect this PR fixes is the mid-handler / mid-acknowledge case, where the loop had already left its while. That is the case the production stall matches.

---

Finding 1 — HIGH (blocker): pull-mode consumers stop consuming permanently after restart()

File: src/data-pump/data-pump.ts:322 (removal of the if (!isRestart) guard), with src/data-pump/data-pump.ts:445-452
Public contract affected: README.md:259-271

Triggering sequence

1. A pull-mode pump (no processor) runs the documented consumer loop:

   while (dataPump.isRunning) {
     const events = await dataPump.reserve(10)
     ...
   }

2. The buffer drains. The consumer parks inside reserve() → waitForEvents().
3. restart(state) is called. stop(true) now resolves waiterEvents unconditionally, and sets running = false.
4. The parked continuation runs on the next microtask. reserveInternal(10, undefined) has no generation to match, so only !this.running applies — it returns [].
5. running stays false for the whole restart block, which awaits dataSource.getTimeBuckets(true) — a forced, paged network fetch (data-source.ts:196-216), not a cache read. The microtask wins this race by a wide margin.
6. while (dataPump.isRunning) evaluates false. The consumer loop exits and never restarts.

Consequence: the fetch loop resumes, refills the buffer to bufferSize, then parks on buffer_full. isRunning is true, the pulse emitter was restarted, and the control plane sees a healthy pump. Nothing is delivered and nothing is checkpointed — the exact signature described in the PR body, moved from push mode to pull mode.

Reproduction (identical harness, the README loop verbatim, getTimeBuckets delayed 25 ms to model the network hop):

base 216ab80f8fa64a32769e193f590c972f4b96b2d8 : loopExited=false iterations=1 emptyReserves=0 isRunning=true
head 76bf591651ca104c4ae96e93f1e31316c4653076 : loopExited=true  iterations=1 emptyReserves=1 isRunning=true

On base the consumer stayed parked across the restart and survived, because stop(true) deliberately did not wake it. That guard was load-bearing for the public path.

The wake itself is necessary — without it a process loop parked in reserve() keeps processLoopRunning === true, and ensureProcessLoop() refuses to replace it. So the fix is not to restore the old guard, but to stop the public path from observing the mid-restart running === false. One option: when generation === undefined and restartTo is pending, re-park instead of returning [].

Please also add a pull-mode-across-restart test. The suite currently has none.

---

Finding 2 — MEDIUM: stop() then start() no longer resumes delivery while a handler is stuck

File: src/data-pump/data-pump.ts:640-642 (the processLoopRunning early return)

Triggering sequence

1. Push mode. The handler blocks on a call with no timeout — an HTTP delivery, a database write.
2. stop() is called. Generation increments, but the hung loop never reaches its guard, so processLoopRunning stays true forever.
3. start() is called. It reaches startProcessLoop(), which returns immediately on the guard. No loop is created.
4. ensureProcessLoop() from the fetch loop also returns early, for the same reason, on every iteration.

Consequence: the pump runs, pulls, buffers, and pulses healthy, and delivers nothing until the original handler resolves — which by assumption it never does. No log, no watchdog, no metric marks the wedge.

Reproduction (handler returns a never-resolving promise; stop(), then start()):

base 216ab80 : handlerCalls=2  bufferDepth=1  running=true   → delivery resumed
head 76bf591 : handlerCalls=1  bufferDepth=1  running=true   → delivery never resumed

I accept the trade being made: base recovered by spawning a second loop that races the first over one buffer, which is its own defect. But ensureProcessLoop()'s doc comment promises that "a delivery loop that exited for any reason … comes back within one fetch iteration", and that promise does not hold for a loop that never exits. At minimum this state deserves a log line or a metric, so it is not silent. A restart across a stuck handler behaves the same way (verified: handlerCalls=1, bufferDepth=1 on both base and head — no regression there, but no recovery either).

---

Finding 3 — LOW: ensureProcessLoop() defeats the process-loop backoff

File: src/data-pump/data-pump.ts:629-634 and 655-666

The .catch() handler sets processLoopRunning = false and then schedules a restart after exponential backoff. During that delay the fetch loop calls ensureProcessLoop(), sees processLoopRunning === false and running === true, and starts a replacement immediately. The backoff never applies, and the setTimeout restart later no-ops on the guard.

This is currently unreachable: processLoop()'s body is fully wrapped in try/catch, so the returned promise cannot reject. The backoff branch is effectively dead code. Flagging it so the interaction is not a surprise if that ever changes.

Finding 4 — LOW: the duplicate-loop test named in the PR description is not in this head

Commit 76bf591 replaced "does not start a second process loop when one is only parked" with the checkpoint test. The PR body still lists it as case 2. The duplicate-loop guard is now covered only indirectly, through the handler-call count in the checkpoint test.

Finding 5 — LOW: stop(_isRestart) keeps an inert public parameter

File: src/data-pump/data-pump.ts:311

The published type still advertises it — stop(isRestart?: boolean) on base becomes stop(_isRestart?: boolean) at head — but the argument now changes nothing. Not a type break; callers passing true still compile and now get plain stop() behaviour. Consider dropping the parameter in a major, or documenting that it is ignored.

---

Residual risks (not blockers, not introduced here)

• A hung handler wedges delivery in every mode, before and after this PR. There is no handler timeout and no stuck-loop detection. Given that this PR exists to remove one silent-stall mode, this is the next one worth closing.
• restart(state, stopAt) does not rebuild stopAtState. It is assigned only in start() (data-pump.ts:253). The restart path does not recompute it, so a new bound is ignored and an old bound persists. Pre-existing and already documented at README.md:792-795, but it matters for exactly the scoped-replay scenario in the PR description.
• Checkpoint ordering across a restart: an acknowledge() that has already passed the generation guard can complete its updateState() write after the restart repositioned bufferState, leaving a stale durable checkpoint. The window is one setState call and the next acknowledgement corrects it, so the practical impact is small.
• restart() stores the caller's state object by reference and later mutates bufferState.eventId (data-pump.ts:376, 421). Pre-existing aliasing.
• Replay observability integrates cleanly. The new early returns happen before observeHandler, so no empty batch is ever observed, and no histogram is left unrecorded.

Independent review. No files in the PR branch or the shared checkout were modified.