Skip to content

EMRY-010: Processor trait + pipeline thread - #55

Merged
vxfemboy merged 3 commits into
mainfrom
emry-010-pipeline
Jun 25, 2026
Merged

EMRY-010: Processor trait + pipeline thread#55
vxfemboy merged 3 commits into
mainfrom
emry-010-pipeline

Conversation

@vxfemboy

Copy link
Copy Markdown
Member

Closes EMRY-010 (#8). Branched off main. First M1 (engine) issue.

What

  • processor.rs: Processor trait (on_event(&mut self, &Event) -> Vec<DerivedMetric>) + DerivedMetric.
  • pipeline.rs: Pipeline spawns one thread that drains the ring, forwards each raw event to the EventBus, runs processors, and emits derived metrics on a bounded channel.

Shutdown contract

  • RunFinished → graceful, ordered: all prior events drained before the thread exits.
  • stop()/Drop → prompt best-effort (flag checked each iteration), so Drop is bounded-time even under a flooding producer.

Verification

  • ./scripts/pre-commit-rust.sh green: clippy -D warnings, 5 engine tests, pipeline.rs 98% lines, total 95.35%
  • Tests: no-op integration (drain + bus forwarding + RunFinished), derived emission, full-channel drop accounting, prompt stop, Drop-without-join

Review-driven changes

  1. Derived channel unbounded → bounded + drop-count (matches ring/bus philosophy; reviewer conf 95).
  2. Drop could block forever under a flooding producer → prompt stop semantics; bounded-time Drop.
  3. stop flag Release/Acquire (control signal, distinct from the ring's standalone Relaxed counter).

vxfemboy added 3 commits June 24, 2026 21:52
emry-engine pipeline: a single thread drains the event ring, forwards each
raw Event to the EventBus for observers, runs every Processor, and emits
DerivedMetrics on a channel. Graceful shutdown on RunFinished or stop()/Drop;
exits only when the ring is empty so pending events are always drained.
Integration tests: no-op processor, derived-metric emission, stop-drain, Drop.
…Release/Acquire

- Derived-metric channel is now bounded (DEFAULT_DERIVED_CAPACITY=16_384) with
  try_send + derived_dropped counter, matching the ring/bus drop policy.
- stop()/Drop is prompt best-effort (stop flag checked at loop top), so Drop is
  bounded-time even under a flooding producer; RunFinished remains the graceful
  ordered-drain path. Documented the two contracts.
- stop flag uses Release/Acquire (it gates control flow, unlike the ring's
  standalone Relaxed counter).
- Add spawn_with_capacity; deterministic full-channel drop test.
CI's pre-commit hook regenerated Cargo.lock and flagged the uncommitted edge.
@vxfemboy
vxfemboy merged commit da5a92a into main Jun 25, 2026
3 checks passed
@vxfemboy
vxfemboy deleted the emry-010-pipeline branch July 2, 2026 23:34
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