feat(runtime): injectable TaskExecutor threaded through the stream openers#191
Open
mfw78 wants to merge 1 commit into
Open
feat(runtime): injectable TaskExecutor threaded through the stream openers#191mfw78 wants to merge 1 commit into
mfw78 wants to merge 1 commit into
Conversation
…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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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 declaredasync fndespite never awaiting anything before spawning.This change introduces an object-safe
TaskExecutortrait with aTokioExecutordefault, threads it (plus aTaskSetthat mirrorsJoinSet::shutdown) through the openers in place of theJoinSet, and gives each reconnect task a typedTaskExitreason instead of returning().Changes
crates/nexum-runtime/src/runtime/task.rsis new and holds the task seam: the object-safeTaskExecutortrait (fn spawn(&self, TaskFuture) -> TaskHandle), theTokioExecutordefault, a typedTaskHandle(abort plus join to aTaskExit), theTaskExitexit-reason enum with its first variantReceiverGone, andTaskSet, which owns the handles and aborts-then-drains onshutdown()the same wayJoinSet::shutdowndid.crates/nexum-runtime/src/runtime/mod.rsregisters the newtaskmodule.crates/nexum-runtime/src/runtime/event_loop.rschangesopen_block_streamsandopen_chain_log_streamsto takeexecutor: &dyn TaskExecutorandtasks: &mut TaskSetinstead of&mut JoinSet<()>, and drops their vestigialasyncsince neither body awaits anything.runnow takes aTaskSetrather than aJoinSet<()>. The tworeconnecting_*_taskfunctions returnTaskExit, yieldingTaskExit::ReceiverGoneat the receiver-dropped exit path.crates/nexum-runtime/src/bootstrap.rsbuilds aTokioExecutorand aTaskSetand passes them into the now-synchronous openers.crates/nexum-runtime/src/supervisor/tests.rsupdates its one call site fromtokio::task::JoinSet::new()toTaskSet::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 innexum-runtime/src/bootstrap.rs, the generic launch path;nexum-cli/src/launch.rsis only the composition root that builds backends and delegates tobootstrap::run. The executor is threaded throughbootstrap.rs, where theJoinSetwas actually created, rather than forcing a CLI-side change that would not match the tree.Test plan
cargo fmt --all --checkcargo clippy --workspace --all-targets --all-features -- -D warningscargo test --workspace --all-features --no-fail-fast(all module wasms built in-tree first so the e2e suite exercises them rather than skipping)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_setsupervisor::tests::run_does_not_bail_when_both_stream_kinds_are_emptye2e_supervisor_boots_example_module,e2e_block_subscription_dispatched,restart_flaky_module_recovers_after_backoffNotes for reviewer
The trait is object-safe (
&dyn TaskExecutor, boxedTaskFuture) 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.TaskExitis a single-variant enum today because the reconnect loop is infinite andReceiverGoneis 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 makesTaskHandleandTaskSet::shutdowntyped instead of()-valued.TaskSet::shutdownpreserves 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 ofrunwere already callingtasks.shutdown().awaitand are unchanged in behaviour.Dropping
asyncfrom 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,
bootstrapstill hard-picksTokioExecutor; threading the executor choice from the CLI launcher (the epic's separate imperative launcher) would let embedders supply a non-tokio executor without touchingbootstrap. Second,TaskSet::shutdowncurrently discards each handle'sTaskExit; 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.