Ref/major refactor - #268
Merged
Merged
Conversation
… types
types/execution/{ids,bindings,flow_ir,signature}.rs and
runtime/execution/{registry,store}.rs defined a typed FlowId/NodeId/
HandlerRegistry/ExecutionStore cluster referenced nowhere outside itself;
the live engine uses raw i64 node ids and ValueStore instead. Same for
types/errors/error.rs, an unused Error enum duplicating RuntimeError, and
the dead ExecutionId(String) newtype shadowing the actually-used
ExecutionId = Uuid alias.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
Both packages declared name = "manual" / "tests" instead of matching their directory name, inconsistent with taurus/taurus-core/taurus-provider. Updates the CI workflow's `cargo run --package tests` reference to match. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
engine.rs exposed 16 near-duplicate execute_flow*/execute_graph* wrapper methods (sync/async x with/without explicit execution id x with/without per-node report) funneling into one private implementation. Only 4 were ever called anywhere in the workspace (production code, CLI tools, or the engine's own tests); the other 12 had zero callers. Removes the 12 dead wrappers and has the two kept graph-level methods call the private core directly instead of routing through an extra indirection layer. No call site outside this file needed to change. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
Case/Cases/Input/RemoteFixture fixture-loading types and the normalize_value/normalize_node_execution_result/null_value proto-value normalization helpers were copy-pasted near-verbatim across taurus, taurus-manual, and taurus-tests. Extracts both into new taurus-core modules (fixtures.rs, normalize.rs) and rewires all three consumers to use them. Also drops CaseResult/Testable from taurus-manual, which were defined but never invoked there (manual never ran cases the way taurus-tests does). Adds serde as a taurus-core dependency for the fixture Deserialize impls; taurus-core already carries serde_json (used in http.rs) so this doesn't introduce a new dependency category. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
NATSRespondEmitter owns a background publish task and a shutdown() method that drains queued lifecycle events before exiting, but the service's wait_for_shutdown only ever .abort()ed the worker task tree (which owns the emitter), so in-flight/queued emitter publishes could be silently dropped on SIGTERM/SIGINT. Replaces the abort with a cooperative tokio::sync::Notify signal: the worker's select! loop now also races the shutdown notification, exits its subscription loop on signal, and calls runtime_emitter.shutdown().await before returning. wait_for_shutdown notifies and then awaits the worker task's own graceful exit instead of cutting it off. Verified end-to-end against a local NATS server: SIGTERM -> graceful worker exit -> emitter drain -> shutdown complete, in that order. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
worker.rs hardcoded with_trace=true on every processed NATS message, so production unconditionally paid for Trace V2 collection: ValueStore:: trace_snapshot deep-clones every entry in latest_results (plus builds a JSON-preview string for each) on every node enter/exit, and tracer:: StoreDiff::from rebuilds two full HashMaps from those snapshots just to diff them per frame exit -- O(n^2) in executed nodes. engine.rs then renders and blocking-println!s the whole trace to stdout after every execution. Nothing in worker.rs consumes the returned trace_run; this was wired for taurus-manual's debug tooling, not the service. Benchmarked (crates/taurus-core/benches/engine_execution.rs) on a 200-node linear chain: 11.5ms with_trace=true vs 168us with_trace=false (~69x). Flamegraph confirms StoreDiff::from and Vec/NodeExecutionResult cloning as the dominant frames before this change. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
Adds //! module docs (matching aquila's style: what a module does, why,
and intra-doc links to related modules) to the files that had none:
taurus's main.rs, client/mod.rs, client/runtime_status.rs, config/mod.rs,
telemetry/{mod,metrics}.rs, and all of taurus-provider (lib.rs, both
provider submodules, and their NATS implementations). Also adds a struct
doc on NATSRespondEmitter explaining why publishing happens on a
background task instead of inline in emit().
No logic changes.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
…orage Three independent clone reductions in the argument/result hot path: - ValueStore::get now takes &ReferenceValue instead of by value, removing a clone on every reference-argument resolution (the common case for node parameters). - node_execution_results() takes the result history via mem::take instead of cloning it; callers use it once at report time right before the store is discarded. - latest_results/result_history now store Arc<NodeExecutionResult> instead of owning two independent copies of every result; insert_node_ result's double-clone becomes one deep clone + one refcount bump. node_execution_results() clears latest_results first so each entry's Arc can be try_unwrap'd for free, falling back to a clone only if some other owner still exists. Benchmarked: ~6-13% additional improvement on top of the trace-disable fix across chain_add/array_map at various sizes (see benches/ engine_execution.rs). All 117 tests + 13-fixture integration suite still pass; behavior unchanged. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
crates/taurus-core/benches/engine_execution.rs (Criterion, cargo bench -p taurus-core): chain_add (linear N-node chains, isolates per-node result-storage/reference-resolution cost), array_map (per-iteration callback cost inside a single node), value_store_get_by_node_id (micro), and compile_vs_encode (a gate-check: measures whether a correctness-safe compiled-flow cache key would even be cheap -- it isn't, see commit message discussion; kept so that conclusion doesn't need re-deriving). crates/taurus-core/examples/profile_workload.rs: the same chain_add workload outside Criterion's own measurement overhead, for flamegraph/ sampling profiling. Adds a `profiling` build profile (release + debug symbols) instead of enabling debug info on [profile.release] itself, so ordinary release builds stay free of the extra symbol/binary-size cost; profiling work uses `cargo build --profile profiling`. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
push_runtime_trace_label always built its String argument (format!, and for array.rs's per-iteration label, a recursive preview_value call) before the method could check whether tracing was even active -- the label is popped and handed to trace_link_child, which already no-ops without a tracer, so with_trace=false paid the formatting cost for nothing. Most visible in map/filter/for_each/sort callbacks, which call this once per iteration. ValueStore now carries a trace_enabled flag (set from with_trace at construction) and push_runtime_trace_label takes a closure, only evaluating and storing it when tracing is on. The two production ValueStore::new call sites in engine.rs collapse into one now that trace_enabled is a constructor argument rather than inferred from a separate branch. Benchmarked array_map/trace_off/1000: 787us -> 753us (~4.5%); the per-item values here are plain integers, so this understates the win for flows whose per-iteration values are larger structs/lists, since preview_value's cost is recursive in the value's shape. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
worker.rs's loop awaited process_execution_message() inline before pulling the next NATS message, so despite running on tokio's multi-thread runtime, exactly one flow ever executed at a time -- concurrent load just queued up and was handled strictly sequentially, regardless of available CPU cores. The loop now only decodes and dispatches: each message gets its own spawned task, bounded by a semaphore (Config::max_concurrent_executions, env MAX_CONCURRENT_EXECUTIONS, default = 4x available CPU parallelism). Once a message is dequeued from NATS we always run it to completion (waiting for a permit if needed, not dropping it), since core NATS has no redelivery for an already-claimed message. Made the per-flow shared resources cheap to hand to spawned tasks: - NATSRemoteRuntime and TaurusRuntimeExecutionService now derive Clone (both wrap Arc-backed/multiplexed connections -- Client and tonic's Channel-based client -- so this is the intended sharing pattern, not a new cost). - ExecutionEngine and NATSRespondEmitter are now held via Arc so spawned tasks can share them; shutdown still drains the emitter exactly once, after every in-flight execution (JoinSet-tracked) has finished and dropped its Arc clone. Verified against a real NATS server + taurus process: 8 flows each calling an HTTP action against a server with a hardcoded 1s delay completed within a 40ms window of each other, instead of ~8s apart sequentially. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
http::request::send does a fully synchronous, blocking ureq call. Now that flows execute concurrently, a blocking HTTP call ties up a shared Tokio worker thread for its full duration -- with N flows in flight and only `available_parallelism()` worker threads, any blocking call beyond that count has to wait for a thread to free up before it can even start, independent of the semaphore admission from the previous commit. Rather than making HandlerFn/ThunkRunner async (which would cascade into every handler that takes a callback -- map, filter, for_each, if/if_else, sort comparators -- for a narrower payoff), taurus stays synchronous at the handler-signature level and instead moves the blocking call off the worker thread via tokio::task::block_in_place, which works regardless of call depth (including HTTP calls made from inside a map/filter/for_each callback). To keep taurus-core's zero-Tokio-dependency invariant intact, the block_in_place wrapper lives in the taurus binary crate (crates/taurus/ src/handler_overrides.rs), not taurus-core. This needed one small new capability in taurus-core: ExecutionEngine::with_overrides, which lets a caller replace specific handler ids after the default registry is built. taurus-core's http::request::send handler is now `pub` so the override can call the real implementation instead of duplicating it; taurus-tests and taurus-manual --offline keep using it directly and are unaffected (block_in_place would panic outside a Tokio runtime, which is exactly where those two never run it). Verified against a real NATS server + taurus process with 16 concurrent flows each calling the http action against a server with a hardcoded 1s delay (8 CPU cores available): - before: two waves of 8 completions, ~1s apart (workers exhausted at 8) - after: all 16 complete within a 44ms window Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
NATSRespondEmitter used mpsc::unbounded_channel for outbound lifecycle events. With serial flow execution this was never a real risk (emit rate was capped by execution rate), but now that flows run concurrently (previous commits), a burst of concurrently-executing flows can call emit() faster than the single publish worker drains the queue -- an unbounded channel would grow without limit if NATS publishing ever lags. Switches to a bounded mpsc::channel (capacity: Config:: emitter_channel_capacity, default 8x max_concurrent_executions) and emit() to try_send instead of send, dropping the event on a Full channel rather than blocking the caller or growing unbounded. These are best-effort lifecycle notifications, not the authoritative execution result (that goes back via gRPC separately in dynamic mode), so dropping under genuine backpressure is an acceptable degradation -- logged at warn (distinct from the existing debug-level "worker unavailable" log for the channel-closed case, so the two are distinguishable in practice). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
block_in_place (previous commit) offloads blocking handlers like the http action onto Tokio's blocking-thread pool, which defaults to a soft cap of 512 threads -- reasonable, but a guess, not a number derived from real workload. Tokio only accepts max_blocking_threads at runtime construction time, so it can't live in Config (which is read from inside an already-running runtime); main() now builds the runtime manually via tokio::runtime::Builder instead of #[tokio::main], reading MAX_BLOCKING_THREADS (default 512, matching Tokio's own default) before constructing it. .env loading also moves from app::run() into main(), since it needs to happen before this env var is read. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
GitLab Pipeline ActionGeneral informationLink to pipeline: https://gitlab.com/code0-tech/development/taurus/-/pipelines/2723990231 Status: Passed Job summariesdocs:previewDocumentation preview available at https://code0-tech.gitlab.io/-/development/telescopium/-/jobs/15659264522/artifacts/out/index.html |
Contributor
There was a problem hiding this comment.
🟡 Not ready to approve
There is at least one likely compile-breaking change (ValueStore::get matching via match *target) plus a few operational/test gaps that should be addressed before approval.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.
Pull request overview
This PR is a broad Taurus runtime refactor focused on improving operational behavior under concurrency (bounded in-flight flow execution and graceful shutdown), consolidating shared helper code into taurus-core, and expanding module-level documentation across the workspace.
Changes:
- Add bounded worker concurrency + cooperative shutdown draining (including emitter queue drain) in
taurus’s NATS execution loop. - Introduce bounded, non-blocking NATS emitter publishing via a background worker and configurable channel capacity.
- Move duplicated JSON fixture loading and proto “null normalization” logic into
taurus-corefor reuse bytaurus-testsandtaurus-manual(and rename those crates accordingly).
File summaries
| File | Description |
|---|---|
| crates/taurus/src/telemetry/mod.rs | Adds module docs clarifying telemetry re-exports and Taurus-specific metrics. |
| crates/taurus/src/telemetry/metrics.rs | Adds docs describing metric initialization and access pattern. |
| crates/taurus/src/main.rs | Replaces #[tokio::main] with explicit Tokio runtime construction and env preload. |
| crates/taurus/src/handler_overrides.rs | Adds Tokio-host wrapper for blocking HTTP handler via block_in_place. |
| crates/taurus/src/config/mod.rs | Adds concurrency and emitter capacity configuration with documented defaults. |
| crates/taurus/src/client/runtime_status.rs | Adds module docs for runtime status/heartbeat client behavior. |
| crates/taurus/src/client/runtime_execution.rs | Centralizes normalization helpers in taurus-core and documents client cloning intent. |
| crates/taurus/src/client/mod.rs | Adds module docs for Aquila-facing clients and dynamic/static behavior. |
| crates/taurus/src/app/worker.rs | Implements bounded concurrency, in-flight draining, and cooperative shutdown signaling. |
| crates/taurus/src/app/mod.rs | Switches to cooperative shutdown (Notify) and wires new engine/emitter configuration. |
| crates/taurus-tests/src/main.rs | Switches to shared fixture helpers from taurus-core and updates docs. |
| crates/taurus-tests/Cargo.toml | Renames crate to taurus-tests and removes now-unneeded serde dep. |
| crates/taurus-provider/src/providers/remote/nats_remote_runtime.rs | Adds docs and derives Clone for per-execution cloning. |
| crates/taurus-provider/src/providers/remote/mod.rs | Adds module docs. |
| crates/taurus-provider/src/providers/mod.rs | Adds crate/module docs and reorders provider modules. |
| crates/taurus-provider/src/providers/emitter/nats_emitter.rs | Converts emitter to bounded channel + background publish worker + shutdown(). |
| crates/taurus-provider/src/providers/emitter/mod.rs | Adds module docs. |
| crates/taurus-provider/src/lib.rs | Adds crate-level docs explaining transport-agnostic core vs NATS provider. |
| crates/taurus-manual/src/main.rs | Switches to shared fixtures + normalization; updates emitter construction. |
| crates/taurus-manual/Cargo.toml | Renames crate to taurus-manual and removes now-unneeded serde dep. |
| crates/taurus-core/src/types/mod.rs | Removes types::execution module from the public types surface. |
| crates/taurus-core/src/types/execution/signature.rs | Removes legacy execution signature types. |
| crates/taurus-core/src/types/execution/mod.rs | Removes legacy execution model module. |
| crates/taurus-core/src/types/execution/ids.rs | Removes legacy execution identifier newtypes. |
| crates/taurus-core/src/types/execution/flow_ir.rs | Removes legacy IR flow graph types. |
| crates/taurus-core/src/types/execution/bindings.rs | Removes legacy argument binding/expression types. |
| crates/taurus-core/src/types/errors/mod.rs | Removes legacy application-level error module export. |
| crates/taurus-core/src/types/errors/error.rs | Removes legacy application-level error type and its tests. |
| crates/taurus-core/src/runtime/functions/mod.rs | Exposes http module publicly for override wrappers. |
| crates/taurus-core/src/runtime/functions/http.rs | Exports send_request and documents Tokio-host override strategy. |
| crates/taurus-core/src/runtime/functions/control.rs | Avoids eager trace-label formatting by deferring via closure. |
| crates/taurus-core/src/runtime/functions/array.rs | Avoids eager trace-label formatting in hot loops via closure. |
| crates/taurus-core/src/runtime/execution/value_store.rs | Introduces Arc-backed result storage and trace gating to reduce cloning/cost. |
| crates/taurus-core/src/runtime/execution/store.rs | Removes legacy execution store type. |
| crates/taurus-core/src/runtime/execution/registry.rs | Removes legacy handler registry metadata types. |
| crates/taurus-core/src/runtime/execution/mod.rs | Removes exports for deleted execution submodules. |
| crates/taurus-core/src/runtime/engine/executor.rs | Updates ValueStore::get call sites to pass references. |
| crates/taurus-core/src/runtime/engine.rs | Adds handler override construction and simplifies/reshapes execution API surface. |
| crates/taurus-core/src/normalize.rs | Adds shared proto value normalization utilities. |
| crates/taurus-core/src/lib.rs | Exposes new shared modules (fixtures, normalize) and expands crate docs. |
| crates/taurus-core/src/fixtures.rs | Adds shared JSON fixture loading/reporting helpers used by tooling binaries. |
| crates/taurus-core/examples/profile_workload.rs | Adds standalone workload driver for sampled profiling runs. |
| crates/taurus-core/Cargo.toml | Adds serde + benchmarking/dev deps and registers benchmark target(s). |
| crates/taurus-core/benches/engine_execution.rs | Adds Criterion benchmarks for engine/value-store hot paths. |
| Cargo.toml | Adds a profiling build profile for release-like builds with debug symbols. |
| Cargo.lock | Updates lockfile for new crates and renamed packages. |
| .github/workflows/build-and-test.yml | Updates CI to run the renamed taurus-tests package. |
Review details
Suppressed comments (2)
crates/taurus/src/app/worker.rs:122
- When draining the JoinSet, join errors (task panics/cancellations) are currently ignored, which can silently drop executions and make failures hard to diagnose. Consider logging
JoinErrorwhile draining so a panicked per-message task is visible in logs/telemetry.
crates/taurus-core/src/runtime/execution/value_store.rs:70 reference.targetis matched viamatch *target, which attempts to move theTargetenum out of a shared reference. Unlesstucana::shared::reference_value::TargetisCopy(unlikely for a prost oneof), this won’t compile. Matching on the reference and cloning only theInputTypepayload avoids the move.
let result = match *target {
tucana::shared::reference_value::Target::FlowInput(_) => self.get_flow_input(),
tucana::shared::reference_value::Target::NodeId(id) => self.get_result(id),
tucana::shared::reference_value::Target::InputType(input_type) => {
self.get_input_type(input_type)
}
- Files reviewed: 46/47 changed files
- Comments generated: 3
- Review effort level: Lite
We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.
Criterion (with rayon, plotters, tinytemplate, etc.) was a dev-dependency of taurus-core, so cargo test -p taurus-core --all-targets -- and CI's blanket cargo test --workspace --all-targets -- compiled it on every routine test run of taurus-core's own unit tests, and executed the benchmark suite as a side effect (Criterion runs a fast single-pass smoke check under `cargo test`, but the with_trace=true scenarios still println! full execution traces, flooding CI logs). Moves benches/engine_execution.rs and examples/profile_workload.rs into a new taurus-bench crate (taurus-core + tucana as regular dependencies, criterion + prost as its own dev-dependencies) and excludes it from CI's workspace test sweep (--exclude taurus-bench), so a routine test run no longer touches the benchmark dependency tree or output at all. `cargo bench -p taurus-bench` and `cargo build --profile profiling -p taurus-bench --example profile_workload` work exactly as before -- only the crate boundary changed, not the benchmark code or the profiling workflow. taurus-bench is a workspace member (shared Cargo.lock, consistent dependency versions with the rest of the repo) but is not part of the default cargo test --workspace sweep now that CI excludes it; running it requires explicitly targeting -p taurus-bench. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
Replaces the ExecutionEngine::with_overrides + taurus-side wrapper function from the previous commit with a single implementation: send_request now checks tokio::runtime::Handle::try_current() itself and runs its blocking call via block_in_place only when an active Tokio multi-thread runtime is found, calling it directly otherwise. taurus-tests and taurus-manual --offline (which run the engine with no runtime at all) are unaffected -- verified by their existing test coverage, which already exercises this handler outside a runtime. This removes the override mechanism entirely: ExecutionEngine:: with_overrides, taurus/src/handler_overrides.rs, and the pub-ification of taurus-core's handler module and http::send_request all go away, since nothing needs to reach into taurus-core's internals from outside anymore. Trade: taurus-core now has a real (non-dev) dependency on tokio, used only for this one runtime-detection check -- not for networking, not for NATS/gRPC, so the crate's "drivable with no transport at all" property is otherwise unchanged. Re-verified against a real NATS server + taurus process: 16 concurrent flows each calling the http action against a server with a hardcoded 1s delay completed within a 67ms window, matching the override-based version's result. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ARriYmjJVUSxoyaMi8vHCK
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.
No description provided.