proxima_primitives/pipe/
interval_pipe.rs

1//! Periodic producer [`SourcePipe`](crate::pipe::source::SourcePipe) — wraps an
2//! interval driver and dispatches a per-tick [`Request`] through an inner
3//! [`Handler`](crate::pipe::handler::Handler).
4//!
5//! `IntervalPipe` is the [`SourcePipe`](crate::pipe::source::SourcePipe) that
6//! converts "fire every N" into the source-shape
7//! [`ProducerLifecycle`](crate::pipe::ProducerLifecycle) drives via
8//! `spawn_from_source`. Each tick calls the configured request factory, then
9//! dispatches the resulting `Request` through the inner handler.
10//!
11//! `IntervalPipe` is a source, not a handler: mounting one on a listener is a
12//! compile error (its `SendPipe` shape is `Signal -> ()`, not
13//! `Request<Bytes> -> Response<Bytes>`) — strictly better than the old
14//! runtime 405 a caller only discovered by dispatching a request into it.
15//!
16//! # Composed primitives
17//!
18//! - [`proxima_core::time::interval`] — the tick source, runtime-agnostic.
19//!   Tests observe real short-period ticks event-driven, bounded by
20//!   [`proxima_core::time::timeout`].
21//! - [`proxima_core::signal::Signal`] — the cooperative cancellation the
22//!   lifecycle passes as `SendPipe::call`'s input; the loop returns once it
23//!   observes the signal fire.
24//!
25//! # Why a wrapper exists vs. composing the primitives directly
26//!
27//! Every caller could spawn a `tokio::time::interval` future themselves
28//! and dispatch through their inner handler. The wrapper centralises:
29//! - cancellation-aware bias (cancel wins over a tick if both are ready,
30//!   so shutdown doesn't fire one last spurious request)
31//! - first-tick suppression so the first dispatch happens one period
32//!   after startup, not immediately at construction
33//! - the request-factory boundary that keeps `IntervalPipe` content-agnostic
34//!   (`carry: Some(Carry::new(MyEvent::heartbeat()))`,
35//!   `body: Bytes::from_static(b"ping")`, or a `carry` payload — decide what
36//!   the tick payload is)
37
38use alloc::boxed::Box;
39use alloc::format;
40use alloc::string::String;
41use alloc::sync::Arc;
42use core::time::Duration;
43
44use bytes::Bytes;
45use futures::FutureExt;
46use proxima_core::factory::Named;
47use proxima_core::signal::Signal;
48
49use crate::pipe::SendPipe;
50use crate::pipe::handler::PipeHandle;
51use crate::pipe::header_list::HeaderList;
52use crate::pipe::request::{Request, RequestContext};
53use crate::pipe::source::{SourceFactory, SourceHandle, into_source_handle};
54use proxima_core::ProximaError;
55
56/// Type-erased request-factory: called per tick to produce the [`Request`]
57/// dispatched into the inner Pipe.
58///
59/// `Send + Sync + 'static` because the factory is moved into the spawned
60/// background task and may be called from any worker thread.
61pub type RequestFactory = Arc<dyn Fn() -> Request<Bytes> + Send + Sync + 'static>;
62
63/// Default method-byte for an interval tick: `b"TICK"`. Configurable via
64/// [`IntervalPipeBuilder::method`].
65pub const DEFAULT_TICK_METHOD: &[u8] = b"TICK";
66
67/// Default path for an interval tick: `"/tick"`. Configurable via
68/// [`IntervalPipeBuilder::path`].
69pub const DEFAULT_TICK_PATH: &[u8] = b"/tick";
70
71/// Periodic producer Pipe. Fires every `period`, dispatches a Request
72/// built by `request_factory` to `inner`.
73///
74/// The pipe is content-agnostic: the request factory decides whether
75/// the body is empty `Bytes`, `Bytes::from_static(b"...")`,
76/// a `Carry` typed payload, or stream — anything the spine can hold. The
77/// method-byte and path are configurable per the discriminant
78/// convention; default to `TICK` / `/tick` so a bare `IntervalPipe`
79/// works as a heartbeat source without any extra config.
80pub struct IntervalPipe {
81    period: Duration,
82    inner: PipeHandle,
83    request_factory: RequestFactory,
84    name: String,
85}
86
87impl IntervalPipe {
88    /// Construct directly. Most callers should use [`Self::builder`].
89    #[must_use]
90    pub fn new(
91        period: Duration,
92        inner: PipeHandle,
93        request_factory: RequestFactory,
94        name: impl Into<String>,
95    ) -> Self {
96        Self {
97            period,
98            inner,
99            request_factory,
100            name: name.into(),
101        }
102    }
103
104    /// Fluent builder entry point (workspace principle 4).
105    #[must_use]
106    pub fn builder() -> IntervalPipeBuilder {
107        IntervalPipeBuilder::default()
108    }
109
110    /// The source's label, set via [`IntervalPipeBuilder::name`] (default
111    /// `"interval"`). Used as the `ProducerLifecycle::spawn_from_source`
112    /// task name by convention; carries no other runtime behaviour.
113    #[must_use]
114    pub fn name(&self) -> &str {
115        &self.name
116    }
117
118    /// Build the default `RequestFactory` — emits a [`Request`] with
119    /// method [`DEFAULT_TICK_METHOD`], path [`DEFAULT_TICK_PATH`], no
120    /// headers, and an empty `Bytes` body. Useful when the inner Pipe only
121    /// cares about the timing edge, not the payload.
122    #[must_use]
123    pub fn empty_request_factory() -> RequestFactory {
124        Arc::new(|| Request {
125            method: crate::pipe::method::Method::from_bytes(DEFAULT_TICK_METHOD),
126            path: Bytes::from_static(DEFAULT_TICK_PATH),
127            query: HeaderList::new(),
128            metadata: HeaderList::new(),
129            payload: Bytes::new(),
130            stream: None,
131            context: RequestContext::default(),
132        })
133    }
134}
135
136impl SendPipe for IntervalPipe {
137    type In = Signal;
138    type Out = ();
139    type Err = ProximaError;
140
141    fn call(
142        &self,
143        cancel: Signal,
144    ) -> impl core::future::Future<Output = Result<(), ProximaError>> + Send {
145        let inner = self.inner.clone();
146        let period = self.period;
147        let factory = self.request_factory.clone();
148        async move {
149            run_interval_loop(period, inner, factory, cancel).await;
150            Ok(())
151        }
152    }
153}
154
155/// Config-driven [`SourceFactory`] for [`IntervalPipe`]. `spec` carries the
156/// tick period in milliseconds; the inner handler and name are supplied by
157/// the caller building the factory instance (mirrors the request-factory
158/// boundary: the factory decides shape, the handler decides content).
159pub struct IntervalSourceFactory {
160    name: String,
161    inner: PipeHandle,
162}
163
164impl IntervalSourceFactory {
165    /// Register a thin factory that builds an [`IntervalPipe`] under `name`,
166    /// dispatching every tick to `inner`.
167    #[must_use]
168    pub fn new(name: impl Into<String>, inner: PipeHandle) -> Self {
169        Self {
170            name: name.into(),
171            inner,
172        }
173    }
174}
175
176impl Named for IntervalSourceFactory {
177    fn name(&self) -> &str {
178        &self.name
179    }
180}
181
182impl SourceFactory for IntervalSourceFactory {
183    fn build(&self, spec: &serde_json::Value) -> Result<SourceHandle, ProximaError> {
184        let period_ms = spec
185            .get("period_ms")
186            .and_then(serde_json::Value::as_u64)
187            .ok_or_else(|| ProximaError::Config("interval source requires period_ms".into()))?;
188        let pipe = IntervalPipe::builder()
189            .period(Duration::from_millis(period_ms))
190            .inner(self.inner.clone())
191            .name(self.name.clone())
192            .build()
193            .map_err(|err| ProximaError::Config(format!("interval source: {err}")))?;
194        Ok(into_source_handle(pipe))
195    }
196}
197
198async fn run_interval_loop(
199    period: Duration,
200    inner: PipeHandle,
201    request_factory: RequestFactory,
202    cancel: Signal,
203) {
204    let mut interval = proxima_core::time::interval(period);
205    interval.set_missed_tick_behavior(proxima_core::time::MissedTickBehavior::Skip);
206    // First tick fires immediately; skip it so the first dispatch is one
207    // period after startup. Matches the principle-of-least-surprise for
208    // heartbeats: "every 5s starting in 5s" rather than "at t=0 and every
209    // 5s thereafter".
210    interval.tick().await;
211    loop {
212        // futures::select_biased! per workspace tokio-elimination
213        // discipline. biased: cancel wins over tick if both are ready,
214        // so shutdown doesn't fire one last spurious dispatch.
215        let cancel_fut = Box::pin(cancel.fired()).fuse();
216        let tick_fut = Box::pin(interval.tick()).fuse();
217        futures::pin_mut!(cancel_fut);
218        futures::pin_mut!(tick_fut);
219        let should_fire = futures::select_biased! {
220            _ = cancel_fut => {
221                tracing::debug!(target = "interval_pipe", "interval cancelled");
222                return;
223            }
224            _ = tick_fut => true,
225        };
226        if should_fire {
227            let request = request_factory();
228            if let Err(err) = inner.call(request).await {
229                tracing::error!(?err, target = "interval_pipe", "inner call failed");
230            }
231        }
232    }
233}
234
235/// Builder for [`IntervalPipe`] (workspace principle 4).
236#[derive(Default)]
237pub struct IntervalPipeBuilder {
238    period: Option<Duration>,
239    inner: Option<PipeHandle>,
240    request_factory: Option<RequestFactory>,
241    method: Option<Bytes>,
242    path: Option<Bytes>,
243    name: Option<String>,
244}
245
246impl IntervalPipeBuilder {
247    /// Set the tick period.
248    #[must_use]
249    pub fn period(mut self, period: Duration) -> Self {
250        self.period = Some(period);
251        self
252    }
253
254    /// Set the inner Pipe that receives the dispatched [`Request`].
255    #[must_use]
256    pub fn inner(mut self, inner: PipeHandle) -> Self {
257        self.inner = Some(inner);
258        self
259    }
260
261    /// Set a fully custom request factory. Each tick calls this closure
262    /// to construct the Request. Overrides any [`method`]/[`path`]
263    /// settings (those only affect the default factory).
264    ///
265    /// [`method`]: Self::method
266    /// [`path`]: Self::path
267    #[must_use]
268    pub fn request_factory<F>(mut self, factory: F) -> Self
269    where
270        F: Fn() -> Request<Bytes> + Send + Sync + 'static,
271    {
272        self.request_factory = Some(Arc::new(factory));
273        self
274    }
275
276    /// Override the method-byte used by the default factory. Ignored if
277    /// a custom [`request_factory`] is set.
278    ///
279    /// [`request_factory`]: Self::request_factory
280    #[must_use]
281    pub fn method(mut self, method: impl Into<Bytes>) -> Self {
282        self.method = Some(method.into());
283        self
284    }
285
286    /// Override the path used by the default factory. Ignored if a
287    /// custom [`request_factory`] is set.
288    ///
289    /// [`request_factory`]: Self::request_factory
290    #[must_use]
291    pub fn path(mut self, path: impl Into<Bytes>) -> Self {
292        self.path = Some(path.into());
293        self
294    }
295
296    /// Set the source's label (default `"interval"`). Used as the
297    /// `spawn_from_source` task name and in tracing.
298    #[must_use]
299    pub fn name(mut self, name: impl Into<String>) -> Self {
300        self.name = Some(name.into());
301        self
302    }
303
304    /// Build the immutable [`IntervalPipe`].
305    ///
306    /// # Errors
307    ///
308    /// Returns `Err` if `period` or `inner` is unset.
309    pub fn build(self) -> Result<IntervalPipe, IntervalBuildError> {
310        let period = self.period.ok_or(IntervalBuildError::MissingPeriod)?;
311        let inner = self.inner.ok_or(IntervalBuildError::MissingInner)?;
312        let factory = match self.request_factory {
313            Some(factory) => factory,
314            None => {
315                let method = crate::pipe::method::Method::from_wire(
316                    self.method
317                        .unwrap_or_else(|| Bytes::from_static(DEFAULT_TICK_METHOD)),
318                );
319                let path = self
320                    .path
321                    .unwrap_or_else(|| Bytes::from_static(DEFAULT_TICK_PATH));
322                Arc::new(move || Request {
323                    method: method.clone(),
324                    path: path.clone(),
325                    query: HeaderList::new(),
326                    metadata: HeaderList::new(),
327                    payload: Bytes::new(),
328                    stream: None,
329                    context: RequestContext::default(),
330                })
331            }
332        };
333        let name = self.name.unwrap_or_else(|| "interval".into());
334        Ok(IntervalPipe::new(period, inner, factory, name))
335    }
336}
337
338/// Errors from [`IntervalPipeBuilder::build`].
339#[derive(Debug, thiserror::Error)]
340#[non_exhaustive]
341pub enum IntervalBuildError {
342    /// `.period(...)` was not called.
343    #[error("IntervalPipe builder missing period")]
344    MissingPeriod,
345    /// `.inner(...)` was not called.
346    #[error("IntervalPipe builder missing inner")]
347    MissingInner,
348}
349
350// `#[proxima::test]` pulls in the `proxima` dev-dependency, which the
351// loom build keeps out of the graph (see
352// `[target.'cfg(not(loom))'.dev-dependencies]` in Cargo.toml); these
353// tests are unrelated to the Notify/watch loom protocol.
354#[cfg(all(test, not(loom)))]
355#[allow(clippy::unwrap_used, clippy::expect_used)]
356mod tests {
357    use alloc::sync::Arc;
358    use std::sync::Mutex;
359
360    use crate::pipe::SendPipe;
361    use crate::pipe::handler::PipeHandle;
362    use crate::pipe::request::Response;
363
364    use super::*;
365
366    struct CountingPipe {
367        count: Mutex<usize>,
368        last_method: Mutex<crate::pipe::method::Method>,
369    }
370
371    impl CountingPipe {
372        fn new() -> Arc<Self> {
373            Arc::new(Self {
374                count: Mutex::new(0),
375                last_method: Mutex::new(crate::pipe::method::Method::default()),
376            })
377        }
378
379        fn count(&self) -> usize {
380            *self.count.lock().unwrap()
381        }
382
383        fn last_method(&self) -> crate::pipe::method::Method {
384            self.last_method.lock().unwrap().clone()
385        }
386    }
387
388    impl SendPipe for CountingPipe {
389        type In = Request<Bytes>;
390        type Out = Response<Bytes>;
391        type Err = ProximaError;
392
393        fn call(
394            &self,
395            request: Request<Bytes>,
396        ) -> impl core::future::Future<Output = Result<Response<Bytes>, ProximaError>> + Send
397        {
398            let method = request.method.clone();
399            *self.count.lock().unwrap() += 1;
400            *self.last_method.lock().unwrap() = method;
401            async { Ok(Response::ok("")) }
402        }
403    }
404
405    fn into_handle(pipe: Arc<CountingPipe>) -> PipeHandle {
406        pipe
407    }
408
409    #[proxima::test]
410    async fn builder_returns_error_on_missing_period() {
411        let inner = into_handle(CountingPipe::new());
412        let result = IntervalPipe::builder().inner(inner).build();
413        assert!(matches!(result, Err(IntervalBuildError::MissingPeriod)));
414    }
415
416    #[proxima::test]
417    async fn builder_returns_error_on_missing_inner() {
418        let result = IntervalPipe::builder()
419            .period(Duration::from_secs(1))
420            .build();
421        assert!(matches!(result, Err(IntervalBuildError::MissingInner)));
422    }
423
424    /// Drive `IntervalPipe` as a `SourcePipe` inline until `capture` has seen
425    /// `ticks` dispatches, then fire the cancel signal so the loop returns.
426    /// The loop future never resolves on its own before that, so the watcher
427    /// arm always wins; the outer timeout bounds a broken loop.
428    async fn drive_until_ticks(pipe: &IntervalPipe, capture: &Arc<CountingPipe>, ticks: usize) {
429        let cancel = Signal::new();
430        let loop_future = SendPipe::call(pipe, cancel.clone()).fuse();
431        let capture_for_watch = capture.clone();
432        let watcher = async move {
433            while capture_for_watch.count() < ticks {
434                proxima_core::time::sleep(Duration::from_millis(1)).await;
435            }
436        }
437        .fuse();
438        futures::pin_mut!(loop_future, watcher);
439        let raced = async {
440            futures::select_biased! {
441                () = watcher => (),
442                _ = loop_future => panic!("interval loop exited without being fired"),
443            }
444        };
445        proxima_core::time::timeout(Duration::from_secs(10), raced)
446            .await
447            .expect("interval_pipe should reach the tick count before the timeout");
448        cancel.fire();
449    }
450
451    #[proxima::test]
452    async fn interval_pipe_fires_three_ticks_with_default_method() {
453        let capture = CountingPipe::new();
454        let inner = into_handle(capture.clone());
455        let pipe = IntervalPipe::builder()
456            .period(Duration::from_millis(5))
457            .inner(inner)
458            .name("heartbeat")
459            .build()
460            .expect("builder");
461        drive_until_ticks(&pipe, &capture, 3).await;
462        assert!(capture.count() >= 3);
463        assert_eq!(
464            capture.last_method(),
465            crate::pipe::method::Method::from_bytes(DEFAULT_TICK_METHOD)
466        );
467    }
468
469    #[proxima::test]
470    async fn interval_pipe_respects_custom_request_factory() {
471        let capture = CountingPipe::new();
472        let inner = into_handle(capture.clone());
473        let pipe = IntervalPipe::builder()
474            .period(Duration::from_millis(5))
475            .inner(inner)
476            .request_factory(|| Request {
477                method: crate::pipe::method::Method::from_bytes(b"CUSTOM"),
478                path: Bytes::from_static(b"/custom"),
479                query: HeaderList::new(),
480                metadata: HeaderList::new(),
481                payload: Bytes::new(),
482                stream: None,
483                context: RequestContext::default(),
484            })
485            .build()
486            .expect("builder");
487        drive_until_ticks(&pipe, &capture, 1).await;
488        assert!(capture.count() >= 1);
489        assert_eq!(
490            capture.last_method(),
491            crate::pipe::method::Method::from_bytes(b"CUSTOM")
492        );
493    }
494
495    #[proxima::test]
496    async fn interval_source_factory_builds_from_period_ms_spec() {
497        let capture = CountingPipe::new();
498        let factory = IntervalSourceFactory::new("heartbeat", into_handle(capture.clone()));
499        let source = factory
500            .build(&serde_json::json!({ "period_ms": 5 }))
501            .expect("factory build");
502
503        let cancel = Signal::new();
504        let call_future = SendPipe::call(&source, cancel.clone()).fuse();
505        let capture_for_watch = capture.clone();
506        let watcher = async move {
507            while capture_for_watch.count() < 1 {
508                proxima_core::time::sleep(Duration::from_millis(1)).await;
509            }
510        }
511        .fuse();
512        futures::pin_mut!(call_future, watcher);
513        futures::select_biased! {
514            () = watcher => (),
515            _ = call_future => panic!("interval source exited without ticking"),
516        }
517        cancel.fire();
518    }
519
520    #[test]
521    fn interval_source_factory_rejects_a_spec_missing_period_ms() {
522        let factory = IntervalSourceFactory::new("heartbeat", into_handle(CountingPipe::new()));
523        let outcome = factory.build(&serde_json::json!({}));
524        assert!(matches!(outcome, Err(ProximaError::Config(_))));
525    }
526}
proxima_primitives/pipe/interval_pipe.rs

proxima_primitives/pipe/
interval_pipe.rs
