diff --git a/.claude/commands/new-op.md b/.claude/commands/new-op.md index 0b60bfb79..ff7fa0b2d 100644 --- a/.claude/commands/new-op.md +++ b/.claude/commands/new-op.md @@ -474,6 +474,39 @@ actually reaches for and no more: the same review that added those also had to delete a cumulative `snapshots()` twin, a `with(|stats| ..)` beside `borrow()`, and `merge` at three levels, none of which ever acquired a caller. +## 4c. `#[must_use]` — on by default, off for sinks + +Dropping a combinator's result is **not** a no-op here: `Stream::wire` has +already registered the node with the shared `Builder`, nothing prunes +unreachable nodes, so a discarded stream stays wired and cycles every tick for +the whole run producing a value nobody reads (#830). Transforms and sources +therefore carry, verbatim: + +```rust +#[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] +``` + +It goes on the **hand-written trait declaration**, and there is one trap: + +- **Never inside the `__wf_fluent_*` expansion.** That lands in a trait + `impl`, where `#[must_use]` is inert (rustc resolves a method call to the + *trait's* item) and rustc warns `unused_attributes` — a future hard error — + for good measure. The declaration in `fluent.rs` is the only place it fires, + which is also where the trait's documented public surface already lives. +- **Sinks do not get it.** Leave it off the declaration. `for_each` / + `for_each_mut` / `print` / `logged` / `inspect` / `timed` / `finally` — and + `StreamOps::feedback`, which sends to its sink whether or not the + pass-through is kept — are called as bare statements throughout this tree + (`examples/adapters/fix/main.rs`), and their side effect happens regardless + of the handle. A false positive here teaches users to ignore the warning, so + **when a new op is genuinely ambiguous — the handle is useful *and* the node + earns its keep unread — leave the attribute off.** + +Pin any addition in `tests/trybuild/must_use_combinators.rs`: it `#![deny]`s +`unused_must_use` so a warning becomes compiler output trybuild can compare, +and it exercises the sinks under the same `deny` so a mis-scoped attribute +breaks the fixture. Regenerate with `TRYBUILD=overwrite`. + ## 5. `nitro!` / compiled coverage For any `#[op]` op this is **zero-touch**: the attribute emits the diff --git a/crates/wingfoil-derive/src/lib.rs b/crates/wingfoil-derive/src/lib.rs index b4ea1e4f2..45f6511f0 100644 --- a/crates/wingfoil-derive/src/lib.rs +++ b/crates/wingfoil-derive/src/lib.rs @@ -2558,6 +2558,15 @@ fn subst_ident(ts: TokenStream2, from: &Ident, to: &TokenStream2) -> TokenStream /// signature that drifts from the op's shape is a compile error rather than a /// silent mismatch. /// +/// That split is also why the combinators' `#[must_use]` (#830) lives on those +/// hand-written declarations and **not** in this quote. What this macro emits +/// lands inside a trait `impl`, and `#[must_use]` on a trait-impl method is +/// inert — rustc resolves a method call to the *trait's* item, so the attribute +/// there would never fire, and it warns `unused_attributes` ("cannot be used on +/// trait methods in impl blocks", a future hard error) into the bargain. So a +/// new transform or source gets the attribute written on its declaration in +/// `fluent.rs`; there is nothing this generator can do for it. +/// /// One shape stays hand-written by construction rather than oversight: an op /// whose fluent signature orders its parameters differently from the generated /// `(edges.., init, cfg)` — `delay_with_reset(delay, trigger)` — cannot be diff --git a/crates/wingfoil/src/adapters/statistics.rs b/crates/wingfoil/src/adapters/statistics.rs index 22d452a57..59542c955 100644 --- a/crates/wingfoil/src/adapters/statistics.rs +++ b/crates/wingfoil/src/adapters/statistics.rs @@ -45,99 +45,123 @@ use crate::ops::*; pub trait StatisticsOps { /// Exponentially-weighted moving average with an explicit /// [`EwmaDecay`] policy (per-tick alpha or clock half-life). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn ewma(&self, decay: EwmaDecay) -> Stream; /// EWMA with a fixed smoothing factor `alpha` applied once per tick, /// seeded on the first sample. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn ewma_per_tick(&self, alpha: f64) -> Stream; /// EWMA whose weights decay off engine time: a sample's weight halves /// every `half_life` of elapsed time, independent of tick rate. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn ewma_half_life(&self, half_life: Duration) -> Stream; /// Sum over a sliding window of the last `window` values. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_sum(&self, window: usize) -> Stream; /// Mean over a sliding window of the last `window` values. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_mean(&self, window: usize) -> Stream; /// Minimum over a sliding window of the last `window` values. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_min(&self, window: usize) -> Stream; /// Maximum over a sliding window of the last `window` values. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_max(&self, window: usize) -> Stream; /// Sample variance (ddof = 1) over a sliding window of the last `window` /// values — the legacy statistics adapter's count-weighted convention /// (divisor `n - 1`, `0.0` until two samples are present). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_var(&self, window: usize) -> Stream; /// Sample standard deviation over a sliding window of the last `window` /// values — the square root of [`rolling_var`](Self::rolling_var) under the /// same (ddof = 1) convention. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_std(&self, window: usize) -> Stream; /// Median over a sliding window of the last `window` values (an even window /// averages its two middle values). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_median(&self, window: usize) -> Stream; /// Cumulative sum over every value seen so far (an unbounded / expanding /// window) — a running total, O(1) per tick. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_sum(&self) -> Stream; /// Cumulative arithmetic mean over every value seen so far — O(1) per tick /// via Welford's online moments. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_mean(&self) -> Stream; /// Cumulative minimum over every value seen so far — a running extreme. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_min(&self) -> Stream; /// Cumulative maximum over every value seen so far — a running extreme. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_max(&self) -> Stream; /// Cumulative **sample** variance (ddof = 1) over every value seen so far — /// the legacy statistics adapter's count-weighted convention (divisor /// `n - 1`, `0.0` until two values are present), maintained incrementally /// with Welford's online moments. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_var(&self) -> Stream; /// Cumulative **sample** standard deviation over every value seen so far — /// the square root of [`cumulative_var`](Self::cumulative_var) under the /// same (ddof = 1) convention. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_std(&self) -> Stream; /// Cumulative median over every value seen so far (an even count averages /// its two middle values). Retains all samples, so its memory grows with /// the stream. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_median(&self) -> Stream; /// Sum over a bounded **time** window — the samples seen in the last /// `window` of graph time (an entry exactly `window` old is retained). O(1) /// per tick. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_sum(&self, window: Duration) -> Stream; /// Arithmetic mean over a bounded time window (count-weighted — the ordinary /// mean of the samples in the window). O(1) amortised per tick. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_mean(&self, window: Duration) -> Stream; /// Minimum over a bounded time window, via a monotonic deque — O(1) /// amortised per tick. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_min(&self, window: Duration) -> Stream; /// Maximum over a bounded time window, via a monotonic deque — O(1) /// amortised per tick. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_max(&self, window: Duration) -> Stream; /// **Sample** variance (ddof = 1) over a bounded time window — `0.0` until /// two samples are in the window. O(1) amortised per tick. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_var(&self, window: Duration) -> Stream; /// **Sample** standard deviation over a bounded time window — the square /// root of [`time_windowed_var`](Self::time_windowed_var). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_std(&self, window: Duration) -> Stream; /// Median over a bounded time window (an even count averages its two middle /// values). Recomputed per tick over the retained window. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_median(&self, window: Duration) -> Stream; // ── time-weighted moments (Weighting::Time) ────────────────────────────── @@ -154,38 +178,47 @@ pub trait StatisticsOps { /// Cumulative time-weighted mean over every sample seen so far (an unbounded /// window) — each sample weighted by how long it was in effect. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_mean_time_weighted(&self) -> Stream; /// Cumulative time-weighted **population** variance over every sample seen so /// far — `m2 / w_sum`, `0.0` until weight is present. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_var_time_weighted(&self) -> Stream; /// Cumulative time-weighted standard deviation over every sample seen so far /// — the square root of [`cumulative_var_time_weighted`](Self::cumulative_var_time_weighted). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_std_time_weighted(&self) -> Stream; /// Time-weighted mean over the most recent `window` samples (a count window). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_mean_time_weighted(&self, window: usize) -> Stream; /// Time-weighted **population** variance over the most recent `window` /// samples (a count window). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_var_time_weighted(&self, window: usize) -> Stream; /// Time-weighted standard deviation over the most recent `window` samples (a /// count window) — the square root of /// [`rolling_var_time_weighted`](Self::rolling_var_time_weighted). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_std_time_weighted(&self, window: usize) -> Stream; /// Time-weighted mean over a bounded time window — the samples seen in the /// last `window` of graph time, each weighted by how long it was in effect. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_mean_time_weighted(&self, window: Duration) -> Stream; /// Time-weighted **population** variance over a bounded time window. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_var_time_weighted(&self, window: Duration) -> Stream; /// Time-weighted standard deviation over a bounded time window — the square /// root of /// [`time_windowed_var_time_weighted`](Self::time_windowed_var_time_weighted). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_std_time_weighted(&self, window: Duration) -> Stream; // ── time-weighted median (Weighting::Time) ─────────────────────────────── @@ -201,14 +234,17 @@ pub trait StatisticsOps { /// Cumulative time-weighted median over every sample seen so far (an /// unbounded window) — each sample weighted by how long it was in effect. /// Retains all samples, so its memory grows with the stream. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn cumulative_median_time_weighted(&self) -> Stream; /// Time-weighted median over the most recent `window` samples (a count /// window), each weighted by how long it was in effect. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn rolling_median_time_weighted(&self, window: usize) -> Stream; /// Time-weighted median over a bounded time window — the samples seen in the /// last `window` of graph time, each weighted by how long it was in effect. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn time_windowed_median_time_weighted(&self, window: Duration) -> Stream; } diff --git a/crates/wingfoil/src/fluent.rs b/crates/wingfoil/src/fluent.rs index 529f58bf5..2022622aa 100644 --- a/crates/wingfoil/src/fluent.rs +++ b/crates/wingfoil/src/fluent.rs @@ -20,6 +20,22 @@ //! This layer is *wiring-time only* — it adds nothing to execution (the built //! [`Runner`] is identical). //! +//! # Combinators are `#[must_use]` +//! +//! Dropping a combinator's result is **not** a no-op: [`Stream::wire`] has +//! already registered the node with the shared [`Builder`], nothing prunes +//! unreachable nodes, so a discarded stream stays wired and cycles every tick +//! for the whole run producing a value nobody reads. Every transform and source +//! declaration here therefore carries `#[must_use]`. +//! +//! **Sinks deliberately do not.** `for_each` / `for_each_mut` / `print` / +//! `logged` / `inspect` / `timed` / `finally` — and `feedback`, which closes +//! the loop by sending to its sink whether or not you keep the pass-through — +//! exist for their side effect, which happens regardless of what you do with +//! the handle they hand back, so `stream.logged("in", Level::Info);` as a bare +//! statement is correct code. Warning on it would train people to ignore the +//! warning that matters. +//! //! # Single-threaded //! //! [`GraphBuilder`] and [`Stream`] are **`!Send` and `!Sync`** — they share @@ -252,6 +268,7 @@ impl GraphBuilder { /// on a stream because no input is privileged; contrast /// [`StreamOps::merge_all`](crate::fluent::StreamOps::merge_all), which /// picks one winner and so has a natural receiver. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] pub fn combine(&self, streams: &[Stream]) -> Stream> { let handles: Vec> = streams.iter().map(|s| s.handle()).collect(); let handle = self.with_builder(|b| b.combine(&handles)); @@ -306,6 +323,7 @@ impl GraphBuilder { /// [`replay_lines`](crate::adapters::lines::replay_lines) / /// [`csv_read`](crate::adapters::csv::csv_read) sources; `csv_read` relies /// on the error-then-stop shape to surface a decode failure. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] pub fn replay_results(&self, rows: I) -> Stream> where T: Clone + Default + 'static, @@ -336,15 +354,18 @@ impl GraphBuilder { /// combinator vocabulary is. pub trait SourceOps { /// A source that ticks at a fixed interval. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn ticker(&self, period: Duration) -> Stream<()>; /// A source that ticks once with `value` on the first cycle. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn constant(&self, value: T) -> Stream; /// An external source: values sent through the returned [`ExternalSource`] /// (from any thread or async task) tick the stream. Emits a [`Burst`] of /// every value that arrived since the last cycle — never latest-wins. /// Realtime mode only. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn external(&self) -> (Stream>, ExternalSource); /// A channel source fed by the returned [`ChannelSender`] (moved to @@ -373,6 +394,7 @@ pub trait SourceOps { /// /// Realtime is unaffected — it is waker-driven and honours its bound whether /// or not anything ever arrives. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn channel(&self) -> (Stream>, ChannelSender); /// [`channel`](Self::channel) with an optional transport bound. `None` is the @@ -383,6 +405,7 @@ pub trait SourceOps { /// starts* (e.g. a `replay_results` feed queued at wiring — with no running /// consumer to drain it, a bounded send blocks at wiring); use plain /// [`channel`](Self::channel) there. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn channel_bounded( &self, buffer: Option, @@ -397,6 +420,7 @@ pub trait SourceOps { /// the backpressure. Note the bounds: `T` needs **no `Clone`** — the /// handle is what the graph clones. See the [`pool`](crate::pool) /// module docs for the design and the loan-budget rules. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn pooled_channel( &self, capacity: usize, @@ -407,6 +431,7 @@ pub trait SourceOps { /// interior capacity (e.g. `|| Book::with_depth(256)` backed by /// `Vec::with_capacity`) so even first uses avoid growth reallocation — /// and for payload types with no `Default`. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn pooled_channel_with( &self, capacity: usize, @@ -419,6 +444,7 @@ pub trait SourceOps { /// A busy-poll source: `f` runs once per engine cycle, ticking on `Some`. /// Lossless and ordered — one value per cycle, no coalescing. The graph /// becomes a busy-spin loop: the kernel never parks. Realtime runs only. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn poll(&self, f: F) -> Stream where T: Clone + Default + 'static, @@ -433,6 +459,7 @@ pub trait SourceOps { /// registry lookup, historical-mode rejection) stays pure and unit-testable; /// a `setup` error aborts the run at start with node context. See /// [`Builder::source_at_start`](crate::interp::Builder::source_at_start). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn source_at_start(&self, setup: Setup) -> Stream> where T: Clone + Default + 'static, @@ -441,6 +468,7 @@ pub trait SourceOps { /// Open a feedback edge: a source stream (no upstreams — the graph stays /// acyclic) plus the [`FeedbackSink`] that feeds it. Close the loop with /// [`StreamOps::feedback`]; values arrive on the source one cycle later. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn feedback(&self) -> (Stream, FeedbackSink) where T: Clone + Default + PartialEq + 'static; @@ -448,6 +476,7 @@ pub trait SourceOps { /// A source that never ticks (the legacy `never`). Useful as an inert /// trigger — e.g. a [`delay_with_reset`](StreamOps::delay_with_reset) that /// never resets behaves like a plain [`delay`](StreamOps::delay). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn never(&self) -> Stream<()>; /// Run a producer sub-graph on its own worker thread and surface its output @@ -864,6 +893,7 @@ impl From<&Stream> for Upstream { /// their own traits (e.g. [`StatisticsOps`](crate::adapters::statistics::StatisticsOps)). pub trait StreamOps: Sized { /// Apply a closure to each value. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn map(&self, f: F) -> Stream where B: Clone + Default + 'static, @@ -871,12 +901,14 @@ pub trait StreamOps: Sized { /// Apply a fallible closure to each value; a returned `Err` aborts the /// run with context. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn try_map(&self, f: F) -> Stream where B: Clone + Default + 'static, F: Fn(&T) -> Result + 'static; /// Map and filter in one pass: `f` returns `(value, emit?)`. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn map_filter(&self, f: F) -> Stream where B: Clone + Default + 'static, @@ -887,22 +919,26 @@ pub trait StreamOps: Sized { /// true))` ticks `v`; `Ok((_, false))` means no value this tick and the /// run continues; `Err(e)` aborts the run with `e` as context — `false` /// and `Err` are not interchangeable. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn try_map_filter(&self, f: F) -> Stream where B: Clone + Default + 'static, F: Fn(&T) -> Result<(B, bool)> + 'static; /// Pair each value with the current engine time: `(time, value)`. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn with_time(&self) -> Stream<(NanoTime, T)> where T: Clone + 'static; /// Emit the current engine time whenever this stream ticks. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn ticked_at(&self) -> Stream where T: 'static; /// Emit elapsed engine time (`now - start`) whenever this stream ticks. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn ticked_at_elapsed(&self) -> Stream where T: 'static; @@ -910,6 +946,7 @@ pub trait StreamOps: Sized { /// Running count of ticks: 1, 2, 3, … — regardless of what this stream /// carries. The values are counted, not inspected, so this is the tick /// count of any stream, not just a `Stream<()>` clock. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn count(&self) -> Stream where T: 'static; @@ -917,6 +954,7 @@ pub trait StreamOps: Sized { /// Fold values into an accumulator, emitting it after each fold. The /// closure mutates the accumulator in place; [`scan`](StreamOps::scan) is /// the same op with the closure returning the new accumulator instead. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn fold(&self, init: B, f: F) -> Stream where B: Clone + 'static, @@ -937,6 +975,7 @@ pub trait StreamOps: Sized { /// Prefer `fold` when the accumulator is expensive to rebuild (a `Vec`, a /// map, an order book) and `scan` when it is a small `Copy` value. See /// [`Scan`](crate::ops::Scan) for the full cost note. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn scan(&self, init: B, f: F) -> Stream where B: Clone + 'static, @@ -956,11 +995,13 @@ pub trait StreamOps: Sized { /// [`inspect`](StreamOps::inspect); for a bounded look-back, use /// [`window`](StreamOps::window) or [`buffer`](StreamOps::buffer). See /// [`ops::Accumulate`](crate::ops::Accumulate). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn accumulate(&self) -> Stream> where T: Clone + Default + 'static; /// Combine with another stream; ticks when either input ticks. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn join(&self, other: &Stream, f: F) -> Stream where B: 'static, @@ -970,6 +1011,7 @@ pub trait StreamOps: Sized { /// Combine with another stream read *passively*: this stream triggers the /// combine, `other`'s current value is read but does not trigger — the /// `bimap(Active, Passive)` shape a feedback input takes. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn join_passive(&self, other: &Stream, f: F) -> Stream where B: 'static, @@ -977,6 +1019,7 @@ pub trait StreamOps: Sized { F: Fn(&T, &B) -> C + 'static; /// Combine three streams (all active); ticks when any input ticks. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn join3(&self, b: &Stream, c: &Stream, f: F) -> Stream where B: 'static, @@ -987,6 +1030,7 @@ pub trait StreamOps: Sized { /// Combine with another stream via a *fallible* closure — the `try_` /// counterpart to [`join`](StreamOps::join). Both inputs active; a returned /// `Err` aborts the run with context (the legacy `try_bimap`). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn try_join(&self, other: &Stream, f: F) -> Stream where B: 'static, @@ -996,6 +1040,7 @@ pub trait StreamOps: Sized { /// [`join_passive`](StreamOps::join_passive) with a *fallible* closure: /// this stream triggers the combine, `other` is read passively, and a /// returned `Err` aborts the run with context. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn try_join_passive(&self, other: &Stream, f: F) -> Stream where B: 'static, @@ -1005,6 +1050,7 @@ pub trait StreamOps: Sized { /// Combine three streams (all active) via a *fallible* closure — the /// `try_` counterpart to [`join3`](StreamOps::join3). A returned `Err` /// aborts the run with context (the legacy `try_trimap`). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn try_join3(&self, b: &Stream, c: &Stream, f: F) -> Stream where B: 'static, @@ -1019,6 +1065,7 @@ pub trait StreamOps: Sized { /// Gates on a *stream*. When the test is a pure function of this stream's /// own value, [`filter_value`](StreamOps::filter_value) says it in one /// node and without a second stream. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn filter(&self, condition: &Stream) -> Stream where T: Clone + Default + 'static; @@ -1038,12 +1085,14 @@ pub trait StreamOps: Sized { /// its own schedule. Neither is sugar for the other: a `Stream` /// deliberately carries a value that may be stale relative to this tick (it /// is a latch), where a predicate always sees the value it is gating. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn filter_value(&self, predicate: F) -> Stream where T: Clone + Default + 'static, F: Fn(&T) -> bool + 'static; /// Emit the current value whenever `trigger` ticks (passive read). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn sample(&self, trigger: &Stream<()>) -> Stream where T: Clone + Default + 'static; @@ -1064,6 +1113,7 @@ pub trait StreamOps: Sized { /// nothing may be dropped, use [`join`](StreamOps::join) instead: it ticks /// when either input ticks and its closure is handed *both* values, so a /// tie is something you handle rather than something you lose. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn merge(&self, other: &Stream) -> Stream where T: Clone + Default + 'static; @@ -1079,6 +1129,7 @@ pub trait StreamOps: Sized { /// merge's earliest-wins tie-break is associative), but a chain costs /// `n - 1` extra nodes and `n - 1` extra depth, which measured 1.86x /// legacy on a busy 256-wide fan-in; see [`MergeN`](crate::ops::MergeN). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn merge_all(&self, others: &[&Stream]) -> Stream where T: Clone + Default + 'static; @@ -1087,6 +1138,7 @@ pub trait StreamOps: Sized { /// identity (a pass-through). Bounded repetition sugar for a straight deep /// chain; inside `nitro!` the count must be a literal so the DAG stays /// static. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn map_n(&self, n: usize, f: F) -> Stream where T: Clone + Default + 'static, @@ -1100,34 +1152,40 @@ pub trait StreamOps: Sized { /// The fan-in is one n-ary [`merge_all`](StreamOps::merge_all) node, so a /// 256-way fan costs 256 branch tails plus **one** merge — not the /// 255-node, 255-deep binary chain it used to build. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn fan(&self, n: usize, branch: F) -> Stream where B: Clone + Default + 'static, F: Fn(Stream) -> Stream; /// Pass through the first `limit` values, then stay quiet. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn limit(&self, limit: usize) -> Stream where T: Clone + Default + 'static; /// Suppress the first `n` values, then pass every later value through. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn skip(&self, n: usize) -> Stream where T: Clone + Default + 'static; /// Rate-limit: emit at most once per `interval`. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn throttle(&self, interval: Duration) -> Stream where T: Clone + Default + 'static; /// Buffer values and flush them as a `Vec` on each `interval` boundary /// (and once more on the last cycle). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn window(&self, interval: Duration) -> Stream> where T: Clone + Default + 'static; /// Buffer values and flush them as a `Vec` once `capacity` accumulate /// (and once more on the last cycle). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn buffer(&self, capacity: usize) -> Stream> where T: Clone + Default + 'static; @@ -1148,6 +1206,7 @@ pub trait StreamOps: Sized { T: Clone + Default + Debug + 'static; /// Suppress consecutive duplicate values (emit on change only). + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn distinct(&self) -> Stream where T: Clone + Default + PartialEq + 'static; @@ -1156,24 +1215,28 @@ pub trait StreamOps: Sized { /// small: `is_small(current, last_emitted)` returning `true` drops the /// tick. The first value always ticks; the reference is the last value /// emitted, not the last seen, so a slow drift still eventually ticks. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn drop_small_change(&self, is_small: F) -> Stream where T: Clone + Default + 'static, F: Fn(&T, &T) -> bool + 'static; /// Emit the successive difference `value - previous`; quiet on the first. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn difference(&self) -> Stream where T: Clone + Default + Sub + 'static; /// Emit pairs of successive values `(previous, current)`. /// Quiet on the first value. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn pairwise(&self) -> Stream<(T, T)> where T: Clone + 'static, (T, T): Default + 'static; /// Negate each value (`!value`) — sugar over `map`. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn not(&self) -> Stream where T: Clone + Default + Not + 'static; @@ -1193,6 +1256,7 @@ pub trait StreamOps: Sized { T: Clone + Default + 'static; /// Re-emit each value `delay` later. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn delay(&self, delay: Duration) -> Stream where T: Clone + Default + PartialEq + 'static; @@ -1201,6 +1265,7 @@ pub trait StreamOps: Sized { /// `delay_with_reset`): when `trigger` ticks, the output snaps to the /// current value and any pending (delayed) values are dropped. `trigger` /// is read for its tick only, so its value type is irrelevant. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn delay_with_reset(&self, delay: Duration, trigger: &Stream) -> Stream where T: Clone + Default + PartialEq + 'static, @@ -1221,6 +1286,7 @@ pub trait StreamOps: Sized { /// IntoIterator`), so only the emitted item is cloned — /// [`Burst`](crate::Burst) and `Vec` both satisfy that through their /// slice iterators. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] fn collapse(&self) -> Stream where T: 'static, @@ -1544,6 +1610,7 @@ impl Stream> { /// for the same reasons: it grows for the whole run. To emit burst values /// as they arrive, `collapse()` into a streaming edge /// (`print` / `logged` / `for_each`) instead. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] pub fn collapse_accumulate(&self) -> Stream> { self.fold(Vec::new(), |acc, burst: &Burst| { acc.extend(burst.iter().cloned()) @@ -1559,6 +1626,7 @@ where /// Decompose a stream of pairs into its two component streams (the legacy /// `split`) — sugar over two [`map`](StreamOps::map)s. Both branches tick /// whenever the source does. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] pub fn split(&self) -> (Stream, Stream) { (self.map(|t| t.0.clone()), self.map(|t| t.1.clone())) } @@ -1569,6 +1637,7 @@ impl Stream> { /// (the legacy `filter_none`) — sugar over /// [`map_filter`](StreamOps::map_filter). A node that has nothing to say /// this cycle emits `None` and the downstream simply does not tick. + #[must_use = "a dropped stream stays wired and cycles every tick, producing an unread value"] pub fn filter_none(&self) -> Stream { self.map_filter(|opt: &Option| match opt.clone() { Some(v) => (v, true), diff --git a/crates/wingfoil/tests/trybuild.rs b/crates/wingfoil/tests/trybuild.rs index fbd03fee8..81bf3dd6f 100644 --- a/crates/wingfoil/tests/trybuild.rs +++ b/crates/wingfoil/tests/trybuild.rs @@ -3,6 +3,12 @@ //! the macro is meant to reject. Run with `TRYBUILD=overwrite` to regenerate //! the expected `.stderr` after an intentional message change. //! +//! `must_use_combinators.rs` is here for a related reason rather than a macro +//! one: `#[must_use]` produces a *warning*, and a warning is invisible to an +//! ordinary test. Under this harness the fixture can `#![deny]` it, so the +//! attribute firing — and, just as important, the sinks it must *not* fire on — +//! is pinned as compiler output like any other diagnostic here. +//! //! `tests/trybuild/pass/` is the other direction, and it is here rather than //! in an ordinary integration test for a reason trybuild is uniquely good for: //! it compiles each file as **its own crate, in its own throwaway Cargo diff --git a/crates/wingfoil/tests/trybuild/must_use_combinators.rs b/crates/wingfoil/tests/trybuild/must_use_combinators.rs new file mode 100644 index 000000000..8fabf1f90 --- /dev/null +++ b/crates/wingfoil/tests/trybuild/must_use_combinators.rs @@ -0,0 +1,64 @@ +// `#[must_use]` on the transform/source combinators (#830) has to actually fire +// at the *call site*, and that is not a given: the attribute is inert when it +// sits on a trait-impl method (rustc resolves a method call to the trait's +// item), so an annotation that drifted onto the `__wf_fluent_*` expansions +// instead of the hand-written `StreamOps` / `SourceOps` declarations would +// silently stop warning. `deny` turns the lint into the compile error trybuild +// can pin. +// +// The other half of the pin is the sinks: `for_each` / `for_each_mut` / `print` +// / `logged` / `inspect` / `timed` / `finally` / `feedback` are called as bare +// statements all over this tree (`examples/adapters/fix/main.rs`), so they must +// **not** warn. They are exercised below under the same `deny`; if one of them +// ever grows the attribute, this file stops compiling and the expected stderr +// no longer matches. +#![deny(unused_must_use)] + +use std::time::Duration; + +use wingfoil::log::Level; +use wingfoil::prelude::*; + +fn transforms(g: &GraphBuilder) { + let count = g.ticker(Duration::from_nanos(10)).count(); + + // Each of these wires a node that will cycle every tick for the whole run + // and produce a value nobody reads. + count.map(|i: &u64| i * 2); + count.filter_value(|i: &u64| *i > 2); + count.accumulate(); + count.with_time(); + count.fold(0u64, |acc: &mut u64, v: &u64| *acc += v); + count.delay(Duration::from_nanos(10)); + count.pairwise(); + count.try_map_filter(|i: &u64| Ok((i * 2, *i > 2))); +} + +fn sources(g: &GraphBuilder) { + // A source is the same bug with nothing upstream of it. + g.ticker(Duration::from_nanos(10)); + g.constant(1u64); + g.never(); + g.channel::(); +} + +// Sinks: no warning, so `deny(unused_must_use)` leaves them alone. +fn sinks(g: &GraphBuilder) { + let count = g.ticker(Duration::from_nanos(10)).count(); + count.print(); + count.logged("count", Level::Info); + count.inspect(|_: &u64| {}); + count.timed(); + count.for_each(|_: &u64| Ok(())); + count.for_each_mut(Vec::::new(), |w: &mut Vec, v: &u64| { + w.push(*v); + Ok(()) + }); + count.finally(|_: &u64| Ok(())); + + let (loop_in, sink) = g.feedback::(); + let _ = loop_in; + count.feedback(&sink); +} + +fn main() {} diff --git a/crates/wingfoil/tests/trybuild/must_use_combinators.stderr b/crates/wingfoil/tests/trybuild/must_use_combinators.stderr new file mode 100644 index 000000000..510b307f1 --- /dev/null +++ b/crates/wingfoil/tests/trybuild/must_use_combinators.stderr @@ -0,0 +1,148 @@ +error: unused return value of `wingfoil::fluent::StreamOps::map` that must be used + --> tests/trybuild/must_use_combinators.rs:27:5 + | +27 | count.map(|i: &u64| i * 2); + | ^^^^^^^^^^^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +note: the lint level is defined here + --> tests/trybuild/must_use_combinators.rs:15:9 + | +15 | #![deny(unused_must_use)] + | ^^^^^^^^^^^^^^^ +help: use `let _ = ...` to ignore the resulting value + | +27 | let _ = count.map(|i: &u64| i * 2); + | +++++++ + +error: unused return value of `filter_value` that must be used + --> tests/trybuild/must_use_combinators.rs:28:5 + | +28 | count.filter_value(|i: &u64| *i > 2); + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +28 | let _ = count.filter_value(|i: &u64| *i > 2); + | +++++++ + +error: unused return value of `accumulate` that must be used + --> tests/trybuild/must_use_combinators.rs:29:5 + | +29 | count.accumulate(); + | ^^^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +29 | let _ = count.accumulate(); + | +++++++ + +error: unused return value of `with_time` that must be used + --> tests/trybuild/must_use_combinators.rs:30:5 + | +30 | count.with_time(); + | ^^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +30 | let _ = count.with_time(); + | +++++++ + +error: unused return value of `wingfoil::fluent::StreamOps::fold` that must be used + --> tests/trybuild/must_use_combinators.rs:31:5 + | +31 | count.fold(0u64, |acc: &mut u64, v: &u64| *acc += v); + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +31 | let _ = count.fold(0u64, |acc: &mut u64, v: &u64| *acc += v); + | +++++++ + +error: unused return value of `delay` that must be used + --> tests/trybuild/must_use_combinators.rs:32:5 + | +32 | count.delay(Duration::from_nanos(10)); + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +32 | let _ = count.delay(Duration::from_nanos(10)); + | +++++++ + +error: unused return value of `pairwise` that must be used + --> tests/trybuild/must_use_combinators.rs:33:5 + | +33 | count.pairwise(); + | ^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +33 | let _ = count.pairwise(); + | +++++++ + +error: unused return value of `try_map_filter` that must be used + --> tests/trybuild/must_use_combinators.rs:34:5 + | +34 | count.try_map_filter(|i: &u64| Ok((i * 2, *i > 2))); + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +34 | let _ = count.try_map_filter(|i: &u64| Ok((i * 2, *i > 2))); + | +++++++ + +error: unused return value of `ticker` that must be used + --> tests/trybuild/must_use_combinators.rs:39:5 + | +39 | g.ticker(Duration::from_nanos(10)); + | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +39 | let _ = g.ticker(Duration::from_nanos(10)); + | +++++++ + +error: unused return value of `constant` that must be used + --> tests/trybuild/must_use_combinators.rs:40:5 + | +40 | g.constant(1u64); + | ^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +40 | let _ = g.constant(1u64); + | +++++++ + +error: unused return value of `never` that must be used + --> tests/trybuild/must_use_combinators.rs:41:5 + | +41 | g.never(); + | ^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +41 | let _ = g.never(); + | +++++++ + +error: unused return value of `wingfoil::fluent::SourceOps::channel` that must be used + --> tests/trybuild/must_use_combinators.rs:42:5 + | +42 | g.channel::(); + | ^^^^^^^^^^^^^^^^^^ + | + = note: a dropped stream stays wired and cycles every tick, producing an unread value +help: use `let _ = ...` to ignore the resulting value + | +42 | let _ = g.channel::(); + | +++++++ diff --git a/docs/adding-an-op.md b/docs/adding-an-op.md index 6288ca37a..09c134fde 100644 --- a/docs/adding-an-op.md +++ b/docs/adding-an-op.md @@ -129,6 +129,15 @@ Where to touch when adding an op — **the compiled path is zero-touch**: | Source (`In = ()`), lifecycle hooks, tick-flag edges | same — `ops.rs` (`impl` + attr) + fluent method | nothing | | Passive edges (`passive = [..]`) / seeded accumulators (`init_arg`) | same — `ops.rs` (`impl` + attr, with the flag) + fluent method | nothing — attribute flags on `#[op]` | | Interpreted signature ≠ the op's shape (`with_time`) | `no_builder` + a hand-written `Builder` method + fluent method | nothing | +| Side-effect sink (`print`, `for_each`) | same — and leave the combinators' `#[must_use]` off the fluent declaration | nothing | + +Two things ride on the fluent *declaration* being hand-written, and +`#[must_use]` is now one of them: a discarded combinator result is not a no-op +(the node stays wired and cycles every tick), so every transform and source +declaration carries `#[must_use = "…"]`. It has to sit on the declaration — +inside the `__wf_fluent_*` expansion it would land in a trait `impl`, where the +attribute is inert. Sinks are the exception and carry nothing. See `/new-op` +step 4c. Constraint #1 still holds (a proc macro sees tokens, not types), but it is routed around rather than paid per-op. Delay's engine-level special cases became