zenkey_fleet/bus/
teardown.rs

1//! The teardown shape the crate shares: **drain everything, then report**.
2//!
3//! [`crate::Monitor::shutdown`] states the rule — "every watch is drained even
4//! if one fails to undeclare … and the failures are reported together" —
5//! because a set half torn down is worse than one torn down noisily. The other
6//! teardowns wrote the loop themselves and bailed on the first failure with a
7//! `?`, leaving everything after it declared (#327, #346).
8//!
9//! This is that loop, once, for anything with an `undeclare`.
10
11use std::future::Future;
12
13use crate::{Error, Result};
14
15/// Undeclare every item, then report what failed — all of it, in one error
16/// naming each item by the label it was drained under.
17///
18/// The label is what a reader needs to find the thing that would not go away:
19/// a key, a selector, a querier's role. Nothing is skipped for an earlier
20/// failure.
21pub(crate) async fn drain_undeclare<T, F, Fut>(items: Vec<(String, T)>, undeclare: F) -> Result<()>
22where
23    F: Fn(T) -> Fut,
24    Fut: Future<Output = Result<()>>,
25{
26    let mut failed = Vec::new();
27    for (label, item) in items {
28        if let Err(e) = undeclare(item).await {
29            // The whole chain: `Display` alone names the operation, and the
30            // per-handle reason is what a teardown report is for (#348).
31            failed.push(format!("{label}: {}", crate::one_line(&e)));
32        }
33    }
34    if failed.is_empty() {
35        Ok(())
36    } else {
37        Err(Error::bus(
38            "undeclare",
39            failed.join("; "),
40            "one or more handles refused",
41        ))
42    }
43}
44
45/// How long a declaration may take before this crate calls it a failure.
46///
47/// A `declare_subscriber`/`declare_queryable`/`declare_token` that never
48/// returns hangs the tool with nothing to report, which is exactly the shape
49/// #341 fixed for `zenoh::open` — and [`OPEN_TIMEOUT`](crate::OPEN_TIMEOUT) is
50/// the precedent this follows. Shorter than the open, because a declaration
51/// happens against a session that is already up (#346).
52pub const DECLARE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
53
54/// Await one declaration under [`DECLARE_TIMEOUT`], naming what stalled.
55///
56/// A stall is an [`Error::Bus`] like any other declare failure: the caller's
57/// input was fine, the fabric did not complete the operation. RFC 13 §3 — a
58/// tool that cannot obtain an observation says so rather than waiting forever
59/// in silence.
60pub(crate) async fn declared<T, E>(
61    op: &'static str,
62    target: impl std::fmt::Display,
63    // `IntoFuture`, not `Future`: zenoh's declaration builders are builders
64    // until awaited, which is what lets a caller pass one straight in.
65    builder: impl std::future::IntoFuture<Output = std::result::Result<T, E>>,
66) -> Result<T>
67where
68    E: std::fmt::Display,
69{
70    match tokio::time::timeout(DECLARE_TIMEOUT, builder.into_future()).await {
71        Ok(Ok(v)) => Ok(v),
72        Ok(Err(e)) => Err(Error::bus(op, target.to_string(), e.to_string())),
73        Err(_) => Err(Error::bus(
74            op,
75            target.to_string(),
76            format!("did not complete within {DECLARE_TIMEOUT:?}"),
77        )),
78    }
79}
80
81#[cfg(test)]
82mod tests {
83    use super::*;
84    use std::sync::atomic::{AtomicUsize, Ordering};
85
86    /// The property the `?`-in-a-loop shape did not have: a failure in the
87    /// middle stops nothing, and the report names every one of them.
88    #[tokio::test]
89    async fn a_failure_stops_nothing_and_every_failure_is_reported() {
90        let visited = AtomicUsize::new(0);
91        let err = drain_undeclare(
92            vec![
93                ("first".to_string(), Ok(())),
94                ("second".to_string(), Err("busy")),
95                ("third".to_string(), Ok(())),
96                ("fourth".to_string(), Err("gone")),
97            ],
98            |outcome: std::result::Result<(), &str>| {
99                visited.fetch_add(1, Ordering::Relaxed);
100                async move { outcome.map_err(|e| Error::bus("undeclare", "handle", e)) }
101            },
102        )
103        .await
104        .expect_err("two of the four would not undeclare")
105        .to_string();
106
107        assert_eq!(visited.load(Ordering::Relaxed), 4, "every item was drained");
108        // Each failure is named *and* carries its reason. The exact join is
109        // not the property; both halves being present is (#348 moved the
110        // reason from `Display` into `source`, so this reads the chain).
111        for (label, reason) in [("second", "busy"), ("fourth", "gone")] {
112            assert!(err.contains(label), "{label} missing from: {err}");
113            assert!(err.contains(reason), "{reason} missing from: {err}");
114        }
115    }
116
117    /// A clean teardown says nothing.
118    #[tokio::test]
119    async fn everything_undeclared_is_silent() {
120        let all_fine = drain_undeclare(vec![("only".to_string(), ())], |()| async { Ok(()) }).await;
121        assert!(all_fine.is_ok());
122    }
123}
zenkey_fleet/bus/teardown.rs

zenkey_fleet/bus/
teardown.rs
