Skip to content

feat(runtime): injectable TaskExecutor threaded through the stream openers#191

Open
mfw78 wants to merge 1 commit into
refactor/chain-logs-nomenclaturefrom
feat/m0-task-executor
Open

feat(runtime): injectable TaskExecutor threaded through the stream openers#191
mfw78 wants to merge 1 commit into
refactor/chain-logs-nomenclaturefrom
feat/m0-task-executor

Conversation

@mfw78

@mfw78 mfw78 commented Jul 5, 2026

Copy link
Copy Markdown
Contributor

Summary

Closes #89.

This is part of the epic #79 train and merges in stack order (based on refactor/chain-logs-nomenclature, ahead of it by this one commit).

Reconnect tasks used to spawn into a bare JoinSet<()> created inside the generic launch path, with no injectable executor and no typed exit reason, and the two stream openers were declared async fn despite never awaiting anything before spawning.
This change introduces an object-safe TaskExecutor trait with a TokioExecutor default, threads it (plus a TaskSet that mirrors JoinSet::shutdown) through the openers in place of the JoinSet, and gives each reconnect task a typed TaskExit reason instead of returning ().

Changes

crates/nexum-runtime/src/runtime/task.rs is new and holds the task seam: the object-safe TaskExecutor trait (fn spawn(&self, TaskFuture) -> TaskHandle), the TokioExecutor default, a typed TaskHandle (abort plus join to a TaskExit), the TaskExit exit-reason enum with its first variant ReceiverGone, and TaskSet, which owns the handles and aborts-then-drains on shutdown() the same way JoinSet::shutdown did.

crates/nexum-runtime/src/runtime/mod.rs registers the new task module.

crates/nexum-runtime/src/runtime/event_loop.rs changes open_block_streams and open_chain_log_streams to take executor: &dyn TaskExecutor and tasks: &mut TaskSet instead of &mut JoinSet<()>, and drops their vestigial async since neither body awaits anything. run now takes a TaskSet rather than a JoinSet<()>. The two reconnecting_*_task functions return TaskExit, yielding TaskExit::ReceiverGone at the receiver-dropped exit path.

crates/nexum-runtime/src/bootstrap.rs builds a TokioExecutor and a TaskSet and passes them into the now-synchronous openers.

crates/nexum-runtime/src/supervisor/tests.rs updates its one call site from tokio::task::JoinSet::new() to TaskSet::new().

The base-shape assumption in the issue (a hoisted launch body at nexum-cli/src/launch.rs) holds, but on this base the reconnect-task lifecycle actually lives in nexum-runtime/src/bootstrap.rs, the generic launch path; nexum-cli/src/launch.rs is only the composition root that builds backends and delegates to bootstrap::run. The executor is threaded through bootstrap.rs, where the JoinSet was actually created, rather than forcing a CLI-side change that would not match the tree.

Test plan

  • cargo fmt --all --check
  • cargo clippy --workspace --all-targets --all-features -- -D warnings
  • cargo test --workspace --all-features --no-fail-fast (all module wasms built in-tree first so the e2e suite exercises them rather than skipping)
  • New unit tests pass: task::tests::tokio_executor_runs_the_future_to_its_exit_reason, task::tests::abort_stops_a_never_ending_task, task::tests::shutdown_drains_a_pending_task_set
  • Modified test passes: supervisor::tests::run_does_not_bail_when_both_stream_kinds_are_empty
  • Existing boot and dispatch end-to-end tests pass unchanged, confirming the reconnect-task rewiring did not regress the launch path: e2e_supervisor_boots_example_module, e2e_block_subscription_dispatched, restart_flaky_module_recovers_after_backoff

Notes for reviewer

The trait is object-safe (&dyn TaskExecutor, boxed TaskFuture) so the concrete runtime is chosen once at the launch root rather than baked into the openers; boxing cost is irrelevant given these tasks are only spawned a handful of times at boot.

TaskExit is a single-variant enum today because the reconnect loop is infinite and ReceiverGone is the sole ordinary exit; it is kept as an enum rather than a unit type so future exit reasons extend cleanly, and it is what makes TaskHandle and TaskSet::shutdown typed instead of ()-valued.

TaskSet::shutdown preserves the previous semantics exactly: abort all, then await each so the engine observes every task finishing before returning. Both the graceful-shutdown and stream-panic arms of run were already calling tasks.shutdown().await and are unchanged in behaviour.

Dropping async from the two openers is safe because neither body awaits internally; an async fn with no internal await point completes in a single synchronous poll, so no scheduling yield is lost.

Two follow-ups deliberately left out of scope. First, bootstrap still hard-picks TokioExecutor; threading the executor choice from the CLI launcher (the epic's separate imperative launcher) would let embedders supply a non-tokio executor without touching bootstrap. Second, TaskSet::shutdown currently discards each handle's TaskExit; a debug log line summarising how each reconnect task exited would help soak diagnosis but was left out to avoid adding shutdown-path log noise without an operator ask.

AI Assistance: Claude Code (Opus 4.8 implementation and adversarial review, Sonnet 5 PR authoring) used for the full change.

…eners

Replace the bare tokio JoinSet the launch path created for the reconnect
tasks with a TaskExecutor abstraction. The openers take a TaskExecutor and
hand back a typed TaskHandle per subscription, collected into a TaskSet the
event loop drains on shutdown. TokioExecutor is the default, spawning onto
the ambient runtime.

The reconnect tasks now return an explicit TaskExit reason (ReceiverGone)
rather than unit, documenting the sole ordinary exit. The stream openers
drop their vestigial async since they only spawn and never await.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant