Repository navigation
feat: Turmoil deterministic simulation testing — BatchAccumulator extraction + pipeline sim tests - #980
Merged
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds a deterministic batch accumulator and integrates it into the pipeline run loop, refactors pipeline buffering/flush/shutdown to use the accumulator, adds a Possibly related PRs
Caution Pre-merge checks failedPlease resolve all errors before merging. Addressing warnings is optional.
❌ Failed checks (2 errors, 2 warnings)
✅ Passed checks (3 passed)
Comment |
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
ApprovabilityVerdict: Needs human review This PR introduces a new simulation testing framework (turmoil) with significant new test infrastructure and feature-gated production code changes. Multiple unresolved review comments exist, including an identified bug where You can customize Macroscope's approvability policy. Learn more. |
There was a problem hiding this comment.
Pull request overview
Adds Turmoil-based deterministic simulation tests for the async pipeline (feature-gated), and extracts the pipeline’s batch accumulation logic into a standalone state machine to make behavior easier to test and reason about.
Changes:
- Extracted batch buffering/checkpoint aggregation into
BatchAccumulatorand refactoredPipeline::run_asyncto use it. - Added
turmoilfeature support in the pipeline (e.g.,scan_maybe_blocking, async input polling loop, andPipeline::for_simulationconstructor). - Added Turmoil simulation integration tests plus a mock
InputSourcefor deterministic input.
Reviewed changes
Copilot reviewed 8 out of 9 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| crates/logfwd/src/batch_accumulator.rs | New accumulator state machine + unit tests. |
| crates/logfwd/src/pipeline.rs | Refactor to accumulator; add Turmoil feature shims and async input loop. |
| crates/logfwd/src/lib.rs | Exposes new module from the crate root. |
| crates/logfwd/tests/turmoil_sim/main.rs | Feature-gated Turmoil integration test harness. |
| crates/logfwd/tests/turmoil_sim/channel_input.rs | Mock InputSource feeding preloaded chunks for simulation. |
| crates/logfwd/tests/turmoil_sim/pipeline_sim.rs | Adds 3 Turmoil simulation tests for pipeline behavior. |
| crates/logfwd/Cargo.toml | Introduces turmoil feature + dependency/dev-dependency. |
| Cargo.toml | Updates workspace unexpected_cfgs check-cfg to include turmoil feature. |
| Cargo.lock | Locks new Turmoil-related transitive dependencies. |
Comment on lines
+1120
to
+1124
| buffered_since = Some(Instant::now()); | ||
| } | ||
| } | ||
|
|
||
| let timeout_elapsed = buffered_since.is_some_and(|t| t.elapsed() >= batch_timeout); |
There was a problem hiding this comment.
async_input_poll_loop measures buffered_since using std::time::Instant, but under Turmoil the simulated time only advances for tokio::time APIs. As a result, t.elapsed() >= batch_timeout may never become true (or may depend on wall-clock), so timeout-based sends become non-deterministic and can be skipped entirely. Track buffered_since using tokio::time::Instant (or compare against tokio::time::Instant::now() / tokio::time::timeout) so the timeout logic follows simulated time.
Suggested change
| buffered_since = Some(Instant::now()); | |
| } | |
| } | |
| let timeout_elapsed = buffered_since.is_some_and(|t| t.elapsed() >= batch_timeout); | |
| buffered_since = Some(tokio::time::Instant::now()); | |
| } | |
| } | |
| let timeout_elapsed = buffered_since | |
| .is_some_and(|t| tokio::time::Instant::now().duration_since(t) >= batch_timeout); |
| @@ -1,2 +1,3 @@ | |||
| pub mod batch_accumulator; | |||
There was a problem hiding this comment.
batch_accumulator is newly exported as a public module, which expands the crate’s public API surface. If this state machine is only an internal implementation detail for Pipeline, consider making it private (mod batch_accumulator;) and scoping the types to pub(crate) to avoid committing to a semver-stable API unintentionally.
Suggested change
| pub mod batch_accumulator; | |
| mod batch_accumulator; |
Comment on lines
+142
to
+146
| let count = delivered.load(Ordering::Relaxed); | ||
| assert_eq!( | ||
| count, 5, | ||
| "expected timeout flush to deliver all 5 rows, got {count}" | ||
| ); |
There was a problem hiding this comment.
This test is named batch_flushes_on_timeout, but it only asserts the final delivered row count after the pipeline is cancelled. As written, it can still pass even if no timeout-based flush happens (e.g., if all data is only sent/drained on shutdown). Consider asserting that a timeout flush occurred (e.g., via a metric like flush_by_timeout, or by cancelling only after observing delivery before shutdown) so the test actually covers the timeout path.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@crates/logfwd/Cargo.toml`:
- Line 37: Move the turmoil dependency into the workspace-level dependencies and
consume it via workspace = true: remove the local inline declaration turmoil = {
version = "0.7", optional = true } from the crate's Cargo.toml and instead add
turmoil = { version = "0.7", optional = true } under [workspace.dependencies],
then update the crate's dependency entry to use turmoil = { workspace = true,
optional = true } (or turmoil = { workspace = true } where appropriate) so the
crate references the workspace dependency rather than defining its own copy.
In `@crates/logfwd/src/batch_accumulator.rs`:
- Line 27: Add missing rustdoc comments for the public API: document the public
struct BatchAccumulator and its public methods new, is_empty, and len with ///
comments that follow the project's CONFIG_REFERENCE.md style; for
BatchAccumulator describe its purpose and usage, for new describe how an
accumulator is constructed and any parameters/ownership behavior, and for
is_empty and len document their semantics and return values; ensure the wording
and tags (e.g., Returns, Examples) match other public docs in the repo.
- Around line 27-105: BatchAccumulator is being used as a state machine but
lacks either a typestate implementation or a justification; either implement
typestate or add a doc comment explaining why typestate is impractical for this
case. To fix: convert BatchAccumulator into an explicit typestate by splitting
states (e.g., Unqueued/Queued) as distinct types and move methods that are only
valid in certain states onto those types (for example, new() returns Unqueued,
ingest(...) transitions Unqueued->Queued and returns Queued which implements
check_timeout(), take_flush(), drain(), etc.), or alternatively add a clear
module-level doc comment above the BatchAccumulator definition explaining why
typestate cannot be used here (address concurrency, API ergonomics, or
performance tradeoffs) and reference the key methods ingest, check_timeout,
drain, take_flush, is_empty and len to justify method placements.
In `@crates/logfwd/src/lib.rs`:
- Line 1: Add a /// rustdoc comment for the newly exported public module
batch_accumulator in lib.rs: describe the module purpose, public API surface and
any important usage notes per CONFIG_REFERENCE.md conventions (brief summary,
example usage if applicable, and links to related modules), ensuring the doc
comment is placed immediately above `pub mod batch_accumulator;` and follows the
repository's documentation style guidelines.
In `@crates/logfwd/src/pipeline.rs`:
- Line 460: The use of expect("default SQL") on SqlTransform::new inside
for_simulation is a vague panic; either propagate the error instead of panicking
by changing for_simulation to return Result and replacing .expect(...) with the
? operator (or map_err to add context) when calling SqlTransform::new, or if
for_simulation must remain infallible, replace the vague message with a concrete
invariant message that includes context and the error (e.g.
"SqlTransform::new(\"SELECT * FROM logs\") failed in for_simulation: {error}")
so failures surface useful details.
- Line 1086: Change the buffered_since variable in async_input_poll_loop from
std::time::Instant to tokio::time::Instant (i.e., let mut buffered_since:
Option<tokio::time::Instant> = None) and update all places that set or compare
it to use tokio::time::Instant::now() (or import tokio::time::Instant) so that
timeouts advance under turmoil's simulated time; ensure any Duration comparisons
remain valid (no other logic changes required).
In `@crates/logfwd/tests/turmoil_sim/channel_input.rs`:
- Line 19: The public constructor ChannelInput::new lacks a required doc
comment; add a /// doc comment above the pub fn new(name: &str, source_id:
SourceId, data: Vec<Vec<u8>>) -> Self that documents the function purpose, each
parameter (name, source_id, data) and the returned ChannelInput according to the
project's CONFIG_REFERENCE.md style (brief description, param tags or sentences,
and return description) so the public API is properly documented.
In `@crates/logfwd/tests/turmoil_sim/pipeline_sim.rs`:
- Around line 107-147: The test currently verifies delivery via an explicit
shutdown drain (CancellationToken.cancel), not that a batch was flushed due to
the batch timeout; change the control flow so the pipeline is stopped by timing
out the run rather than by calling shutdown.cancel. Specifically, in
batch_flushes_on_timeout replace the tokio::spawn that cancels sd with running
pipeline.run_async(&shutdown) under tokio::time::timeout (e.g. 200ms) so the run
is aborted by a timeout (letting Pipeline::set_batch_timeout(50ms) trigger a
flush) and do not call sd.cancel(); keep references to
Pipeline::set_batch_timeout, pipeline.run_async, and the delivered Atomic
counter to assert 5 rows.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro
Run ID: d35bcf7a-de1e-40ba-8f5c-14ded86aaa1f
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (8)
Cargo.tomlcrates/logfwd/Cargo.tomlcrates/logfwd/src/batch_accumulator.rscrates/logfwd/src/lib.rscrates/logfwd/src/pipeline.rscrates/logfwd/tests/turmoil_sim/channel_input.rscrates/logfwd/tests/turmoil_sim/main.rscrates/logfwd/tests/turmoil_sim/pipeline_sim.rs
| tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "time", "sync"] } | ||
| tokio-util = { version = "0.7" } | ||
| logfwd-io = { version = "0.1.0", path = "../logfwd-io" } | ||
| turmoil = { version = "0.7", optional = true } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify where `turmoil` is declared today.
rg -n '^\[workspace.dependencies\]|^\s*turmoil\s*=' Cargo.toml crates/logfwd/Cargo.tomlRepository: strawgate/memagent
Length of output: 273
Move turmoil to workspace dependencies.
turmoil on lines 37 and 51 must be declared in [workspace.dependencies] and consumed via workspace = true, per the workspace dependency policy.
Fix
# Cargo.toml (workspace root)
[workspace.dependencies]
+turmoil = "0.7"
# crates/logfwd/Cargo.toml
-turmoil = { version = "0.7", optional = true }
+turmoil = { workspace = true, optional = true }
[dev-dependencies]
-turmoil = "0.7"
+turmoil = { workspace = true }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/Cargo.toml` at line 37, Move the turmoil dependency into the
workspace-level dependencies and consume it via workspace = true: remove the
local inline declaration turmoil = { version = "0.7", optional = true } from the
crate's Cargo.toml and instead add turmoil = { version = "0.7", optional = true
} under [workspace.dependencies], then update the crate's dependency entry to
use turmoil = { workspace = true, optional = true } (or turmoil = { workspace =
true } where appropriate) so the crate references the workspace dependency
rather than defining its own copy.
| Continue, | ||
| } | ||
|
|
||
| pub struct BatchAccumulator { |
There was a problem hiding this comment.
Public API docs are incomplete on the new accumulator type.
BatchAccumulator, new, is_empty, and len need /// rustdoc comments.
As per coding guidelines: "All new public items must have /// doc comments matching CONFIG_REFERENCE.md where applicable".
Also applies to: 35-35, 86-86, 90-90
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/src/batch_accumulator.rs` at line 27, Add missing rustdoc
comments for the public API: document the public struct BatchAccumulator and its
public methods new, is_empty, and len with /// comments that follow the
project's CONFIG_REFERENCE.md style; for BatchAccumulator describe its purpose
and usage, for new describe how an accumulator is constructed and any
parameters/ownership behavior, and for is_empty and len document their semantics
and return values; ensure the wording and tags (e.g., Returns, Examples) match
other public docs in the repo.
Comment on lines
+27
to
+105
| pub struct BatchAccumulator { | ||
| buf: BytesMut, | ||
| checkpoints: HashMap<SourceId, ByteOffset>, | ||
| queued_at: Option<Instant>, | ||
| batch_target_bytes: usize, | ||
| } | ||
|
|
||
| impl BatchAccumulator { | ||
| pub fn new(batch_target_bytes: usize) -> Self { | ||
| Self { | ||
| buf: BytesMut::with_capacity(batch_target_bytes), | ||
| checkpoints: HashMap::new(), | ||
| queued_at: None, | ||
| batch_target_bytes, | ||
| } | ||
| } | ||
|
|
||
| /// Ingest data from an input. Returns Flush if buffer reached target size. | ||
| pub fn ingest( | ||
| &mut self, | ||
| bytes: Bytes, | ||
| input_checkpoints: Vec<(SourceId, ByteOffset)>, | ||
| queued_at: Instant, | ||
| ) -> AccumulatorAction { | ||
| if self.queued_at.is_none() { | ||
| self.queued_at = Some(queued_at); | ||
| } | ||
| self.buf.extend_from_slice(&bytes); | ||
| for (sid, offset) in input_checkpoints { | ||
| self.checkpoints.insert(sid, offset); | ||
| } | ||
| if self.buf.len() >= self.batch_target_bytes { | ||
| self.take_flush("size") | ||
| } else { | ||
| AccumulatorAction::Continue | ||
| } | ||
| } | ||
|
|
||
| /// Check if a timeout-based flush should happen. | ||
| /// Returns Flush if there is any buffered data. | ||
| pub fn check_timeout(&mut self) -> AccumulatorAction { | ||
| if self.buf.is_empty() { | ||
| AccumulatorAction::Continue | ||
| } else { | ||
| self.take_flush("timeout") | ||
| } | ||
| } | ||
|
|
||
| /// Force-drain remaining data (shutdown path). | ||
| pub fn drain(&mut self) -> Option<(Bytes, HashMap<SourceId, ByteOffset>, Option<Instant>)> { | ||
| if self.buf.is_empty() { | ||
| return None; | ||
| } | ||
| let data = self.buf.split().freeze(); | ||
| let checkpoints = std::mem::take(&mut self.checkpoints); | ||
| let queued_at = self.queued_at.take(); | ||
| Some((data, checkpoints, queued_at)) | ||
| } | ||
|
|
||
| pub fn is_empty(&self) -> bool { | ||
| self.buf.is_empty() | ||
| } | ||
|
|
||
| pub fn len(&self) -> usize { | ||
| self.buf.len() | ||
| } | ||
|
|
||
| fn take_flush(&mut self, reason: &'static str) -> AccumulatorAction { | ||
| let data = self.buf.split().freeze(); | ||
| let checkpoints = std::mem::take(&mut self.checkpoints); | ||
| let queued_at = self.queued_at.take(); | ||
| AccumulatorAction::Flush { | ||
| data, | ||
| checkpoints, | ||
| queued_at, | ||
| reason, | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
State-machine implementation needs typestate (or explicit justification for not using it).
BatchAccumulator is introduced as a new state machine but does not follow typestate and does not document why typestate is impractical here.
As per coding guidelines: "New state machines must use the typestate pattern except when the PR justifies why typestate is impractical for that specific case".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/src/batch_accumulator.rs` around lines 27 - 105,
BatchAccumulator is being used as a state machine but lacks either a typestate
implementation or a justification; either implement typestate or add a doc
comment explaining why typestate is impractical for this case. To fix: convert
BatchAccumulator into an explicit typestate by splitting states (e.g.,
Unqueued/Queued) as distinct types and move methods that are only valid in
certain states onto those types (for example, new() returns Unqueued,
ingest(...) transitions Unqueued->Queued and returns Queued which implements
check_timeout(), take_flush(), drain(), etc.), or alternatively add a clear
module-level doc comment above the BatchAccumulator definition explaining why
typestate cannot be used here (address concurrency, API ergonomics, or
performance tradeoffs) and reference the key methods ingest, check_timeout,
drain, take_flush, is_empty and len to justify method placements.
| @@ -1,2 +1,3 @@ | |||
| pub mod batch_accumulator; | |||
There was a problem hiding this comment.
Add rustdoc for the new public module.
Line 1 exports a new public module without a /// doc comment.
Proposed fix
+/// Batch accumulation state machine used by pipeline batching and flush decisions.
pub mod batch_accumulator;As per coding guidelines: "All new public items must have /// doc comments matching CONFIG_REFERENCE.md where applicable".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/src/lib.rs` at line 1, Add a /// rustdoc comment for the newly
exported public module batch_accumulator in lib.rs: describe the module purpose,
public API surface and any important usage notes per CONFIG_REFERENCE.md
conventions (brief summary, example usage if applicable, and links to related
modules), ensuring the doc comment is placed immediately above `pub mod
batch_accumulator;` and follows the repository's documentation style guidelines.
|
|
||
| let scan_config = ScanConfig::default(); | ||
| let scanner = Scanner::new(scan_config); | ||
| let transform = SqlTransform::new("SELECT * FROM logs").expect("default SQL"); |
There was a problem hiding this comment.
Avoid vague expect in for_simulation.
Line 460 uses expect("default SQL"), which is a vague panic message. Use a concrete invariant message.
Proposed fix
- let transform = SqlTransform::new("SELECT * FROM logs").expect("default SQL");
+ let transform = SqlTransform::new("SELECT * FROM logs")
+ .expect("invariant: hardcoded identity SQL must always parse");As per coding guidelines: "No .unwrap()/.expect("vague") in production paths — use ? or a real invariant explanation".
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| let transform = SqlTransform::new("SELECT * FROM logs").expect("default SQL"); | |
| let transform = SqlTransform::new("SELECT * FROM logs") | |
| .expect("invariant: hardcoded identity SQL must always parse"); |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/src/pipeline.rs` at line 460, The use of expect("default SQL")
on SqlTransform::new inside for_simulation is a vague panic; either propagate
the error instead of panicking by changing for_simulation to return Result and
replacing .expect(...) with the ? operator (or map_err to add context) when
calling SqlTransform::new, or if for_simulation must remain infallible, replace
the vague message with a concrete invariant message that includes context and
the error (e.g. "SqlTransform::new(\"SELECT * FROM logs\") failed in
for_simulation: {error}") so failures surface useful details.
| } | ||
|
|
||
| impl ChannelInputSource { | ||
| pub fn new(name: &str, source_id: SourceId, data: Vec<Vec<u8>>) -> Self { |
There was a problem hiding this comment.
Document the public constructor.
Line 19 introduces a new public function without a /// doc comment.
As per coding guidelines: "All new public items must have /// doc comments matching CONFIG_REFERENCE.md where applicable".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/tests/turmoil_sim/channel_input.rs` at line 19, The public
constructor ChannelInput::new lacks a required doc comment; add a /// doc
comment above the pub fn new(name: &str, source_id: SourceId, data:
Vec<Vec<u8>>) -> Self that documents the function purpose, each parameter (name,
source_id, data) and the returned ChannelInput according to the project's
CONFIG_REFERENCE.md style (brief description, param tags or sentences, and
return description) so the public API is properly documented.
Comment on lines
+107
to
+147
| #[test] | ||
| fn batch_flushes_on_timeout() { | ||
| let mut sim = turmoil::Builder::new() | ||
| .simulation_duration(Duration::from_secs(30)) | ||
| .tick_duration(Duration::from_millis(1)) | ||
| .build(); | ||
|
|
||
| let delivered = Arc::new(AtomicU64::new(0)); | ||
| let delivered_clone = delivered.clone(); | ||
|
|
||
| sim.client("pipeline", async move { | ||
| // 5 small lines -- well below batch_target_bytes | ||
| let lines = generate_json_lines(5); | ||
| let input = ChannelInputSource::new("test", SourceId(1), lines); | ||
| let sink = CountingSink::new(delivered_clone); | ||
|
|
||
| let mut pipeline = Pipeline::for_simulation("sim", Box::new(sink)); | ||
| pipeline.set_batch_timeout(Duration::from_millis(50)); | ||
| let pipeline = pipeline.with_input("test", Box::new(input)); | ||
| let mut pipeline = pipeline; | ||
|
|
||
| let shutdown = CancellationToken::new(); | ||
| let sd = shutdown.clone(); | ||
|
|
||
| tokio::spawn(async move { | ||
| tokio::time::sleep(Duration::from_secs(2)).await; | ||
| sd.cancel(); | ||
| }); | ||
|
|
||
| pipeline.run_async(&shutdown).await.unwrap(); | ||
| Ok(()) | ||
| }); | ||
|
|
||
| sim.run().unwrap(); | ||
|
|
||
| let count = delivered.load(Ordering::Relaxed); | ||
| assert_eq!( | ||
| count, 5, | ||
| "expected timeout flush to deliver all 5 rows, got {count}" | ||
| ); | ||
| } |
There was a problem hiding this comment.
batch_flushes_on_timeout does not verify timeout flushing.
This test currently proves delivery, not timeout-triggered flush. A shutdown drain can satisfy the same assertion.
Proposed fix
sim.client("pipeline", async move {
@@
- let mut pipeline = Pipeline::for_simulation("sim", Box::new(sink));
+ let mut pipeline = Pipeline::for_simulation("sim", Box::new(sink));
+ let metrics = Arc::clone(pipeline.metrics());
@@
pipeline.run_async(&shutdown).await.unwrap();
+ assert!(
+ metrics.flush_by_timeout.load(Ordering::Relaxed) > 0,
+ "expected at least one timeout-based flush"
+ );
Ok(())
});🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/tests/turmoil_sim/pipeline_sim.rs` around lines 107 - 147, The
test currently verifies delivery via an explicit shutdown drain
(CancellationToken.cancel), not that a batch was flushed due to the batch
timeout; change the control flow so the pipeline is stopped by timing out the
run rather than by calling shutdown.cancel. Specifically, in
batch_flushes_on_timeout replace the tokio::spawn that cancels sd with running
pipeline.run_async(&shutdown) under tokio::time::timeout (e.g. 200ms) so the run
is aborted by a timeout (letting Pipeline::set_batch_timeout(50ms) trigger a
flush) and do not call sd.cancel(); keep references to
Pipeline::set_batch_timeout, pipeline.run_async, and the delivered Atomic
counter to assert 5 rows.
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@crates/logfwd/src/pipeline.rs`:
- Line 479: The field last_checkpoint_flush currently uses std::time::Instant
which makes checkpoint throttling use real wall-clock time; change its type to
tokio::time::Instant and initialize it with tokio::time::Instant::now() so
comparisons in the checkpoint throttling logic (the code that checks elapsed
time in the throttling branch and the use alongside buffered_since in
async_input_poll_loop) use the same simulated/timebase; also update the
import/uses in this module (and any function signatures or comparisons) to refer
to tokio::time::Instant to ensure type consistency.
- Around line 483-484: Remove the duplicated doc comment line for the pipeline
runner: delete one of the identical "/// Run the pipeline until `shutdown` is
cancelled. Blocks the calling thread." lines (the triple-slash doc comment
before the pipeline run function) so the function's docblock contains a single
copy; locate the duplicate comment immediately above the pipeline run function
in pipeline.rs and keep only one of the two identical /// lines.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro
Run ID: 730819da-60db-46de-ae0f-18c3f882d307
📒 Files selected for processing (2)
crates/logfwd/src/pipeline.rsdev-docs/references/turmoil-simulation.md
Comment on lines
483
to
484
| /// Run the pipeline until `shutdown` is cancelled. Blocks the calling thread. | ||
| /// Run the pipeline until `shutdown` is cancelled. Blocks the calling thread. |
There was a problem hiding this comment.
Duplicate doc comment line.
/// Run the pipeline until `shutdown` is cancelled. Blocks the calling thread.
- /// Run the pipeline until `shutdown` is cancelled. Blocks the calling thread.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| /// Run the pipeline until `shutdown` is cancelled. Blocks the calling thread. | |
| /// Run the pipeline until `shutdown` is cancelled. Blocks the calling thread. | |
| /// Run the pipeline until `shutdown` is cancelled. Blocks the calling thread. |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/src/pipeline.rs` around lines 483 - 484, Remove the duplicated
doc comment line for the pipeline runner: delete one of the identical "/// Run
the pipeline until `shutdown` is cancelled. Blocks the calling thread." lines
(the triple-slash doc comment before the pipeline run function) so the
function's docblock contains a single copy; locate the duplicate comment
immediately above the pipeline run function in pipeline.rs and keep only one of
the two identical /// lines.
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@crates/logfwd/tests/turmoil_sim/crash_sim.rs`:
- Around line 83-89: Update the test in crash_sim.rs to assert per-source
durable checkpoints rather than only the aggregate: after the simulation
completes, query the pipeline's checkpoint store (e.g.,
checkpoint_store.get_checkpoint(source_id) or equivalent API used elsewhere in
this test) for Source 1 and Source 2 and assert they equal 10 and 5
respectively; also add the same per-source assertions around the out-of-order
ack scenario (the block referenced near lines 129-135) to confirm ack reordering
doesn't corrupt each source's checkpoint state. Ensure you use the existing
checkpoint store accessor (checkpoint_store, get_checkpoint, or the test's
existing checkpoint inspection helper) so the assertions integrate with the
current test harness.
- Around line 24-31: The test in crash_sim.rs currently simulates a crash on
checkpoint flush but never asserts the durable checkpoint state remained
unchanged; update the test (the crash simulation block in the crash_sim test) to
record the durable checkpoint marker before triggering the flush error (e.g.,
the persisted checkpoint id/position you use in the test harness), trigger the
simulated flush crash, then after recovery/restart assert that the durable
checkpoint marker is equal to the pre-crash value (or that the persisted
checkpoint did not advance) and that the pipeline re-reads the same rows
(verifying at-least-once); use the existing helpers around the flush/crash
simulation (e.g., simulate_flush_crash, checkpoint_flush,
durable_checkpoint_state or checkpoint_id variables in this test file) to locate
where to add these assertions.
In `@crates/logfwd/tests/turmoil_sim/network_sim.rs`:
- Around line 24-31: The test that simulates a partition-recovery should assert
full recovery instead of just "count > 0": modify the assertions in this test
(the #[test] block and the later checks around the commented lines) to verify
that all expected items/batches sent after recovery were delivered (e.g.,
assert_eq!(delivered_count, total_sent) or assert_eq!(delivered_batches,
expected_batches)) and also assert the expected number of dropped batches during
the partition (e.g., dropped_batch metric == 3) so the test fails on incomplete
recovery or unexpected drops.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro
Run ID: e2aa2779-8645-494f-bbb8-6d7380666a39
📒 Files selected for processing (5)
crates/logfwd/tests/turmoil_sim/crash_sim.rscrates/logfwd/tests/turmoil_sim/crashable_checkpoint.rscrates/logfwd/tests/turmoil_sim/failure_sinks.rscrates/logfwd/tests/turmoil_sim/main.rscrates/logfwd/tests/turmoil_sim/network_sim.rs
Comment on lines
+24
to
+31
| /// Test: arm a crash on the checkpoint flush. The pipeline should handle | ||
| /// the flush error gracefully (log warning, continue running). After the | ||
| /// crash, the durable checkpoint state should NOT have advanced — meaning | ||
| /// a restart would re-read the data (at-least-once, not at-most-once). | ||
| /// | ||
| /// This test verifies the core crash-recovery guarantee: data is never | ||
| /// silently lost due to a crash between output delivery and checkpoint | ||
| /// persistence. |
There was a problem hiding this comment.
Test does not verify the durable-checkpoint guarantee it claims.
This test only proves some rows were delivered. It does not assert that durable checkpoint state stayed unchanged after the simulated flush crash, so the core guarantee in the test name/doc is currently unverified.
Also applies to: 75-81
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd/tests/turmoil_sim/crash_sim.rs` around lines 24 - 31, The test
in crash_sim.rs currently simulates a crash on checkpoint flush but never
asserts the durable checkpoint state remained unchanged; update the test (the
crash simulation block in the crash_sim test) to record the durable checkpoint
marker before triggering the flush error (e.g., the persisted checkpoint
id/position you use in the test harness), trigger the simulated flush crash,
then after recovery/restart assert that the durable checkpoint marker is equal
to the pre-crash value (or that the persisted checkpoint did not advance) and
that the pipeline re-reads the same rows (verifying at-least-once); use the
existing helpers around the flush/crash simulation (e.g., simulate_flush_crash,
checkpoint_flush, durable_checkpoint_state or checkpoint_id variables in this
test file) to locate where to add these assertions.
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/logfwd/src/pipeline.rs (1)
576-607:⚠️ Potential issue | 🟠 MajorEnsure TLA+ specs are updated to reflect drain/checkpoint sequencing changes.
The code reorders shutdown (drain channel before joining threads, lines 619–643), delays checkpoint persistence to Stopped/ForceStop state (lines 676–717), and changes accumulator flushing semantics (lines 576–607). These directly affect
DrainCompleteness,CheckpointOrderingInvariant, andCommittedMonotonicproperties. Per the learnings, PRs touching drain, checkpoint ordering, or batch sequencing must updatetla/PipelineMachine.tlaand include fresh TLC verification output before merge.Also applies to: 619-717, 927-968
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/logfwd/src/pipeline.rs` around lines 576 - 607, This change alters batch flushing and shutdown/checkpoint sequencing—update the TLA+ specification PipelineMachine.tla to reflect the new semantics (BatchAccumulator.ingest/check_timeout behavior, flush_batch_from invocation with "size"/"timeout", and the reordered shutdown sequence that drains the channel before joining workers and delays checkpoint persistence until Stopped/ForceStop), then re-run TLC and include the verification output proving the core properties still hold: DrainCompleteness, CheckpointOrderingInvariant, and CommittedMonotonic; ensure the spec models the shutdown cancel path (shutdown.cancelled), channel message handling (ChannelMsg::Data), and checkpoint persistence transitions so the TLC results correspond to the modified logic.
♻️ Duplicate comments (1)
crates/logfwd/src/pipeline.rs (1)
119-122:⚠️ Potential issue | 🟠 Major
set_checkpoint_flush_interval()still runs on wall-clock time underturmoil.The new test hook only changes the duration;
ack_all_tickets()still gates flushes withstd::time::Instant::elapsed(). Under Turmoil, simulated time will not advance that clock, so intermediate checkpoint flushes remain non-deterministic and these sims effectively only exercise the final shutdown flush. Please switch this path totokio::time::Instantor narrow the API/docs to real-time builds only.Also applies to: 482-483, 495-499, 959-962
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@crates/logfwd/src/pipeline.rs` around lines 119 - 122, The ack_all_tickets() path still uses std::time::Instant (last_checkpoint_flush/elasped) while set_checkpoint_flush_interval() only adjusts a Duration, causing non-determinism under Turmoil; change the code to use tokio::time::Instant and tokio::time::Instant::now() for last_checkpoint_flush and all elapsed/advance checks so simulated time advances correctly, and ensure checkpoint_flush_interval remains a tokio::time::Duration-compatible value (or document/guard the API to be real-time-only if you prefer not to switch); update any related uses (e.g., where last_checkpoint_flush is initialized/compared in ack_all_tickets and any other locations flagged) to the tokio types.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@crates/logfwd/tests/turmoil_sim/observable_checkpoint.rs`:
- Around line 27-34: The ObservableCheckpointStore is consumed by
with_checkpoint_store(Box::new(store)) so callers lose access to
durable_offset(), flush_count(), assert_monotonic(), and update_history after
sim.run(); change the store to hold its mutable state inside a shareable probe
(e.g., struct ObservableCheckpointProbe(Arc<Mutex<Inner>>>) where Inner contains
durable: BTreeMap<u64,SourceCheckpoint>, pending, flush_count, update_history,
and crash_armed) and make ObservableCheckpointStore hold a cloneable Arc to that
probe; construct the probe before calling with_checkpoint_store and pass a clone
into the store so tests can keep the probe handle alive and inspect
durable_offset(), flush_count(), and call assert_monotonic() against the probe
after sim.run(), and ensure with_checkpoint_store receives the store that clones
the Arc rather than the sole ownership of the inner state.
---
Outside diff comments:
In `@crates/logfwd/src/pipeline.rs`:
- Around line 576-607: This change alters batch flushing and shutdown/checkpoint
sequencing—update the TLA+ specification PipelineMachine.tla to reflect the new
semantics (BatchAccumulator.ingest/check_timeout behavior, flush_batch_from
invocation with "size"/"timeout", and the reordered shutdown sequence that
drains the channel before joining workers and delays checkpoint persistence
until Stopped/ForceStop), then re-run TLC and include the verification output
proving the core properties still hold: DrainCompleteness,
CheckpointOrderingInvariant, and CommittedMonotonic; ensure the spec models the
shutdown cancel path (shutdown.cancelled), channel message handling
(ChannelMsg::Data), and checkpoint persistence transitions so the TLC results
correspond to the modified logic.
---
Duplicate comments:
In `@crates/logfwd/src/pipeline.rs`:
- Around line 119-122: The ack_all_tickets() path still uses std::time::Instant
(last_checkpoint_flush/elasped) while set_checkpoint_flush_interval() only
adjusts a Duration, causing non-determinism under Turmoil; change the code to
use tokio::time::Instant and tokio::time::Instant::now() for
last_checkpoint_flush and all elapsed/advance checks so simulated time advances
correctly, and ensure checkpoint_flush_interval remains a
tokio::time::Duration-compatible value (or document/guard the API to be
real-time-only if you prefer not to switch); update any related uses (e.g.,
where last_checkpoint_flush is initialized/compared in ack_all_tickets and any
other locations flagged) to the tokio types.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro
Run ID: 32018235-b61c-4cc7-9251-87f390607bad
📒 Files selected for processing (6)
crates/logfwd/src/pipeline.rscrates/logfwd/tests/turmoil_sim/crash_sim.rscrates/logfwd/tests/turmoil_sim/instrumented_sink.rscrates/logfwd/tests/turmoil_sim/main.rscrates/logfwd/tests/turmoil_sim/network_sim.rscrates/logfwd/tests/turmoil_sim/observable_checkpoint.rs
This was referenced Apr 5, 2026
Extract BatchAccumulator state machine from pipeline.rs to enable deterministic testing. Add turmoil feature flag that replaces OS threads with async tasks and block_in_place with direct calls, enabling single-threaded simulation. Includes Pipeline::for_simulation constructor and three passing simulation tests (happy path, graceful shutdown drain, timeout flush). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…n sim Add dev-docs/references/turmoil-simulation.md covering full API, pitfalls, best practices, and logfwd-specific patterns. Based on research across Turmoil docs, s2n-quic, RisingWave, WarpStream, Polar Signals, and ecosystem blog posts. Fix: async_input_poll_loop used std::time::Instant for buffered_since timing. Under Turmoil, std::time::Instant returns real wall-clock time, not simulated time — so batch_timeout checks were non-deterministic. Switched to tokio::time::Instant which uses Turmoil's simulated clock. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
5 new Turmoil simulation tests exercising failure paths: Network simulation: - pipeline_survives_network_partition: IO errors (connection refused) for first 3 batches, then recovery — pipeline doesn't panic/hang - retry_on_503_then_succeeds: RetryAfter for first 2 calls, then success — worker pool retries and delivers all 10 rows - high_latency_does_not_stall_batching: 500ms output latency — pipeline still accumulates and delivers all 30 rows Crash consistency: - crash_before_checkpoint_does_not_advance_durable_state: armed crash on checkpoint flush — pipeline handles error gracefully - multi_source_checkpoint_independence: two sources with different row counts — checkpoints advance independently New test infrastructure: - CrashableCheckpointStore: durable/pending split with arm_crash() - PartitionThenRecoverSink: IO errors then recovery - RetryThenSucceedSink: RetryAfter then success - HighLatencySink: configurable async delay per send All 8 simulation tests pass in 0.11s (deterministic simulated time). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…iant-based tests Replace weak mock sinks (PartitionThenRecoverSink, RetryThenSucceedSink, HighLatencySink) with InstrumentedSink that follows programmable FailureAction scripts and records all calls via shared atomic counters. Replace CrashableCheckpointStore with ObservableCheckpointStore that records full update history for post-test invariant verification. Add Pipeline accessors: in_flight_count() for test assertions, configurable checkpoint_flush_interval (was hardcoded 5s) for faster test feedback. Rewrite failure/crash tests with strong invariant assertions: - ordered_ack_under_out_of_order_delivery: verifies checkpoint monotonicity when first batch is delayed - retry_exhaustion_drops_batch_and_advances: verifies pipeline does not hang when all attempts fail (MAX_RETRIES=3) - shutdown_drain_with_inflight_work: verifies drain completes with slow sink - crash_checkpoint_consistency: verifies at-least-once guarantee under flush crash - multi_source_independent_checkpoint: verifies per-source checkpoint independence Keeps 3 happy-path tests unchanged. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
… flush bug Rewrote simulation tests to actually verify invariants instead of just counting rows: **New test infrastructure:** - ObservableCheckpointStore with Arc-shared state — durable offsets, update history, crash/flush counts inspectable from OUTSIDE the Pipeline closure via CheckpointHandle - InstrumentedSink with FailureScript — programmable per-call behavior (Succeed, Delay, IoError, RetryAfter, Reject) - Deleted old failure_sinks.rs and crashable_checkpoint.rs **Invariant assertions now verified:** - Checkpoint monotonicity (assert_monotonic): offsets never decrease - Checkpoint upper bound (assert_checkpoint_le): durable ≤ delivered - Crash recovery: crash_count > 0 AND flush_count > 0 after recovery - Per-source independence: both sources have updates + durable offsets - Retry exhaustion: ≥4 calls (1 + MAX_RETRIES=3), 0 rows delivered **Bug found: checkpoint flush throttle broken under simulated time** The 5-second checkpoint flush throttle in ack_all_tickets() uses std::time::Instant::elapsed(), which returns ~0 under Turmoil's simulated time (real wall-clock is near-zero). This means checkpoints were NEVER flushed during simulation — the first crash test exposed this because flush_count remained 0 even after the crash was armed and should have recovered. Fix: under turmoil feature, always flush when checkpoints advance (the throttle exists to amortize real fsync cost, not sim cost). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…i-worker test Two real fixes and one new test: 1. Switch last_checkpoint_flush from std::time::Instant to tokio::time::Instant. std::time returns wall-clock time which is ~0 under simulated time, so the checkpoint flush throttle never fired under Turmoil (or tokio::time::pause()). This is the proper fix — removes the cfg hack. 2. Add Pipeline::for_simulation_with_factory() for multi-worker testing. The existing for_simulation() uses OnceAsyncFactory (1 worker), making concurrent ack testing impossible. 3. Add multi_worker_out_of_order_ack_checkpoint_ordering test. 2 workers: worker 1 slow (3s delay), worker 2 fast. Verifies the PipelineMachine ordered-ack invariant with ACTUAL concurrency — the single-worker test could never exercise this path. 4. Fix crash_checkpoint_consistency test to document a real design finding: when the final shutdown flush crashes, ALL checkpoint progress is lost. At-least-once delivery is maintained (data re-read on restart) but the re-read window is the entire run. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Add TurmoilTcpSink and TcpServer for turmoil::net-based network testing: - TurmoilTcpSink: Sink impl that sends NDJSON over turmoil::net::TcpStream with lazy connect and reconnect-on-error - TcpServer: controlled TCP listener that counts received lines/connections - Test A (real_tcp_delivery_through_turmoil_net): end-to-end proof that turmoil::net works for pipeline delivery across simulated hosts - Test B (tcp_partition_causes_retry_and_recovery): partition/repair exercises real worker pool retry logic under network failure - Test C (tcp_server_crash_triggers_reconnect): crash/bounce validates connection recovery after server restart Also tightens existing tests: - Delete ordered_ack_under_out_of_order_delivery (superseded by multi_worker_out_of_order_ack_checkpoint_ordering) - crash_checkpoint_consistency: assert count == 20 (was count > 0) - shutdown_drain_with_inflight_work: assert delivered > 0 - batch_flushes_on_timeout: add mid-test assertion before shutdown Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…duction bugs Each test targets a specific bug hypothesis from code audit: 1. Ack starvation under sustained load (select! fairness) - PASSED 2. Worker panic does not block drain (ticket leak) - PASSED 3. Rejected batch checkpoint behavior - CONFIRMED: checkpoint advances past undelivered rejected data, permanently skipping it on restart 4. Drain path ack processing (checkpoint burst) - PASSED 5. RetryAfter respects server backoff timing - PASSED Also adds FailureAction::Panic variant to InstrumentedSink and set_batch_target_bytes() setter to Pipeline for test configurability. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
… assertions Based on final audit: Deleted: - happy_path_all_data_delivered: redundant with graceful_shutdown (same invariant, weaker — timeout flush catches everything before shutdown) - drain_path_ack_processing: redundant with ack_starvation (both verify intermediate checkpoint advancement, starvation test is stronger) Strengthened assertions: - shutdown_drain_with_inflight_work: assert_eq(10) not assert > 0 - tcp_partition: assert received >= 5 not > 0 - tcp_server_crash: document retry exhaustion during crash window as a real finding — transient downtime > retry window causes batch loss - rejected_batch: assert durable > 0 when delivered == 0, documenting the confirmed checkpoint-advances-past-reject bug (#1001) 14 tests remain, each covering a unique invariant. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…ease and fail_rate tests Replace all inline turmoil::Builder::new() calls with shared helpers: - build_sim(duration, tick) for tests without sim.host() - sim_builder() for TCP tests needing host registration before build Fix sim_builder() to compile (turmoil Builder methods return &mut Self, not Self). Add turmoil_seed() helper for tests that need the seed value. Add two new network simulation tests: - tcp_hold_release_burst_delivery: exercises turmoil hold/release for delayed (not dropped) delivery, validating burst recovery - tcp_intermittent_failures_with_fail_rate: exercises turmoil fail_rate for random TCP connection breakage with pipeline retry recovery Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
strawgate
force-pushed
the
feat/turmoil-simulation
branch
from
April 5, 2026 02:16
c33a605 to
627170f
Compare
This was referenced Apr 11, 2026
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
Adds deterministic simulation testing using Turmoil for the async pipeline. This replaces flaky timing-dependent tests with seeded, reproducible simulations where time advances instantly.
BatchAccumulator extraction (roadmap 7d):
batch_accumulator.rs— pure state machine for batch accumulation withingest(),check_timeout(),drain()pipeline.rsto delegate toBatchAccumulatorinstead of inline stateTurmoil integration:
turmoilfeature flag — opt-in, no impact on default buildsscan_maybe_blocking()— calls scanner directly under turmoil (avoidsblock_in_placepanic on single-thread runtime)async_input_poll_loop()— async replacement for OS-thread input polling under turmoil (usestokio::time::sleepfor deterministic time)Pipeline::for_simulation()— minimal constructor bypassing config/filesystem/OTel3 simulation tests:
happy_path_all_data_delivered— 50 JSON lines flow through scan → transform → outputgraceful_shutdown_drains_buffered_data— cancel token triggers drain, all 10 rows deliveredbatch_flushes_on_timeout— 5 small lines flush via timeout, not size thresholdTest plan
cargo test -p logfwd— 60 existing tests pass (no turmoil feature)cargo test -p logfwd --features turmoil— 81 tests pass (existing + 3 sim)cargo clippy -p logfwd -- -D warnings— cleancargo clippy -p logfwd --features turmoil -- -D warnings— cleanCloses #423
🤖 Generated with Claude Code
Note
Add Turmoil deterministic simulation tests and extract
BatchAccumulatorfrom pipelineBatchAccumulatorstate machine that handles size-based, timeout-based, and drain flushes, replacing ad-hoc buffer management inpipeline.rs.turmoiloptional feature flag and a suite of deterministic simulation tests covering: graceful shutdown drain, batch timeout flushing, checkpoint consistency under crashes, multi-source independent checkpoints, out-of-order ack ordering, network partitions, server crash/reconnect, and retry exhaustion.ChannelInputSource(mock input),InstrumentedSink/InstrumentedSinkFactory(scriptable sink outcomes),ObservableCheckpointStore(checkpoint assertions with crash injection), and aTurmoilTcpSinkfor end-to-end TCP tests.turmoilfeature,Pipelinegainsfor_simulation/for_simulation_with_factoryconstructors and an async input poll loop (async_input_poll_loop) that respects simulated time instead of blocking the runtime.block_in_placeunderturmoilto avoid panics on single-threaded runtimes; checkpoint flush throttling switches from wall-clock totokio::time::Instantwith a configurable interval.Macroscope summarized 627170f.