fix(scheduler): emit to subscribers asynchronously

[iamquang95]

Fix #535

Changes

The slot and duty subscriber loops in scheduler.rs awaited each handler inline before receiving the next broadcast event. A slow or hung handler stalled the loop until it lagged, at which point the subscriber terminated (the loop breaks on RecvError::Lagged) — cascading into duty failures. Surfaced during simnet smoke debugging.

This spawns the handler per event and loops straight back to rx.recv(), so a slow subscriber never blocks emission of later slots/duties.

• subscribe_slot / subscribe_duty: spawn a detached task per event instead of awaiting inline.
• label → Arc<str> so it clones cheaply into each task; duty error log gains %duty.
• Regression tests: a later slot/duty is still delivered to a subscriber whose earlier handler hangs (fails on the inline-await version, passes after).

Charon references

Charon emits to subscribers asynchronously so a slow subscriber never blocks emission.

• Slot subscribers — goroutine per subscriber: core/scheduler/scheduler.go:157-167 (go func(sub){ ... }(sub); matches the "Emit scheduled slot event" log).
• Duty trigger — goroutine per duty: core/scheduler/scheduler.go:240-265 (log at :263; matches the "Trigger duty subscriber error" log).

Not a 1:1 port: Charon runs one goroutine per duty calling all subscribers sequentially, but each Charon subscriber is itself retry-wrapped fire-and-forget (go retryer.DoAsync(...); return nil, core/retry.go), so those calls return instantly. pluto has no such retry-wrapper, so it puts the detachment in the scheduler instead — one task per (subscriber, event). Different structure, same guarantee: a slow handler never blocks emission.

Note

Changes the concurrency contract — subscribers are now invoked concurrently, and tasks are detached fire-and-forget. Safe standalone in healthy operation (instances decide fast, tasks are short-lived),
but best landed with/after the consensus deadline-cancellation change, which caps stuck-instance task lifetime.

[iamquang95]

@claude review this PR

[claude]

Claude finished @iamquang95's task in 2m 39s —— View job

---

Review: fix(scheduler): emit to subscribers asynchronously

• Read the PR diff and surrounding scheduler code
• Compare against Charon Go reference
• Analyze correctness, concurrency, and style
• Post review

The change is correct and well-motivated, and it aligns pluto's emission model more closely with Charon. Nicely scoped with genuine regression tests. Below are the tradeoffs worth being explicit about and a couple of small nits.

---

Correctness ✅

The core change — receive the event, spawn a detached per-event task, loop straight back to rx.recv() — does exactly what the PR claims: a slow/hung handler no longer parks the receive loop.

• let fut = f(&slot) is created before slot is moved into the inner task. Since the trait bound is F: Fn(&Slot) -> Fut with Fut a fixed type parameter (independent of the input lifetime), the returned future cannot borrow slot/set, so moving slot (for slot = %slot.slot) and dropping set after f(&duty, &set) are both sound. Same for the duty arm. ✔
• Matches Charon: emitCoreSlot spawns a goroutine per subscriber per slot (scheduler.go:157-167) and duty triggering spawns per duty (scheduler.go:240-265). Charon therefore also has no per-subscriber in-order guarantee across slots — so this change doesn't regress relative to Charon; it converges toward it. ✔
• Regression tests are solid: unbounded_channel + HashSet + timeout, with an explicit comment that inter-task scheduling order isn't fixed, so they don't rely on ordering. They genuinely fail on the inline-await version (loop parks on slot 0, slot 1 never received → 2s timeout). ✔

Main tradeoff to make explicit ⚠️ — loss of the lag fail-safe

Before this change, a persistently slow handler eventually triggered RecvError::Lagged, which terminated the subscription — an (implicit) backpressure fail-safe that bounded the damage. After this change the receive loop never blocks, so for the slow-handler scenario the Lagged branch becomes effectively dead code: instead of the subscription dying, handler tasks now accumulate without bound (one per slot every ~12s, one per duty) for as long as handlers stay stuck. A permanently hung handler is now a permanent task leak rather than a terminated subscriber.

This is the concurrency-contract shift the PR already flags, and the mitigation (consensus deadline-cancellation capping stuck-task lifetime) is the right dependency. Two concrete follow-ups:

1. The // NOTE: A lagging subscriber requires further analysis ... terminate the subscription. comment (scheduler.rs:138-139, 181) is now stale/misleading: after this change, lag no longer signals a slow handler — it can only fire under runtime/scheduler saturation (the receive-loop task itself being starved). Worth updating so the next reader doesn't reason about the old model. Fix this →
2. Consider whether unbounded per-subscriber task growth needs a bound (e.g. a JoinSet/semaphore, or at minimum a metric/warn) independent of the deadline-cancellation work, so a bug in any handler can't silently pile up tasks. Not blocking given the noted dependency — just flagging it shouldn't be forgotten once deadline-cancellation lands.

Ordering / concurrency contract 📝

Handlers for consecutive events on a single subscriber can now run concurrently and complete out of order. Any subscriber that implicitly assumes sequential, in-order processing would break. Today the only in-tree callers are the tests, so current risk is low, and this matches Charon's model — but it's a real contract change for future subscribers (fetcher → consensus, etc.), so the "best landed with/after consensus deadline-cancellation" note is well placed.

Interaction with the existing abort TODO 📝

The // TODO: return a handle so clients can .abort() now spans two task layers: aborting the outer receive-loop task would leave already-spawned per-event handler tasks running. Whoever implements that handle will need to track/abort the inner tasks too (a JoinSet would cover both this and the unbounded-growth concern above).

Nits

• Arc::from(label.as_ref()) allocates a fresh Arc<str> once per subscription, then clones are cheap refcount bumps — correct and idiomatic. ✔
• %duty added to the duty error log is a nice touch and mirrors Charon logging z.U64("slot", ...) (duty's Display includes the slot). ✔

Verdict

Approve with notes. The change is correct, matches Charon, and is well-tested. The one thing I'd ask before/at merge is updating the now-stale Lagged comments (nit #1); the unbounded-task-growth and ordering concerns are real but already acknowledged and appropriately gated on the consensus deadline-cancellation change.
• feat/scheduler-loop