From bcee78edf36c550190c5afd41451440d8046aabc Mon Sep 17 00:00:00 2001 From: Nathan Perry Date: Wed, 2 Sep 2026 17:38:34 -0400 Subject: [PATCH 1/3] ts_future_util: init and write Debounce Signed-off-by: Nathan Perry Change-Id: I5d6fa4c5ab801f0cdae1138f904afcec6a6a6964 --- Cargo.lock | 11 + Cargo.toml | 2 + ts_future_util/Cargo.toml | 34 +++ ts_future_util/README.md | 3 + ts_future_util/src/debounce.rs | 429 +++++++++++++++++++++++++++++++++ ts_future_util/src/lib.rs | 14 ++ 6 files changed, 493 insertions(+) create mode 100644 ts_future_util/Cargo.toml create mode 100644 ts_future_util/README.md create mode 100644 ts_future_util/src/debounce.rs create mode 100644 ts_future_util/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 44e857b5..a142f988 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6075,6 +6075,17 @@ dependencies = [ "ts_keys", ] +[[package]] +name = "ts_future_util" +version = "0.5.0" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "tokio", + "tokio-stream", +] + [[package]] name = "ts_hexdump" version = "0.5.0" diff --git a/Cargo.toml b/Cargo.toml index a4359250..eb9bfb26 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ members = [ "ts_dynbitset", "ts_elixir/native/ts_elixir", "ts_ffi", + "ts_future_util", "ts_hexdump", "ts_http_util", "ts_keys", @@ -123,6 +124,7 @@ ts_dataplane = { path = "ts_dataplane", version = "0.5.0" } ts_derp = { path = "ts_derp", version = "0.5.0" } ts_disco_protocol = { path = "ts_disco_protocol", version = "0.5.0" } ts_dynbitset = { path = "ts_dynbitset", version = "0.5.0" } +ts_future_util = { path = "ts_future_util", version = "0.5.0" } ts_hexdump = { path = "ts_hexdump", version = "0.5.0" } ts_keys = { path = "ts_keys", version = "0.5.0" } ts_kv_store = { path = "ts_kv_store", version = "0.5.0" } diff --git a/ts_future_util/Cargo.toml b/ts_future_util/Cargo.toml new file mode 100644 index 00000000..7effa1b0 --- /dev/null +++ b/ts_future_util/Cargo.toml @@ -0,0 +1,34 @@ +[package] +name = "ts_future_util" +version.workspace = true +description = "tailscale futures utilities" +categories = ["asynchronous"] +keywords = ["tailscale", "futures", "async"] + +repository.workspace = true +edition.workspace = true +license.workspace = true +rust-version.workspace = true + +[dependencies] +futures-core = { version = "0.3", default-features = false } +pin-project-lite = "0.2" + +tokio = { workspace = true, optional = true, features = ["time"] } + +[dev-dependencies] +futures-util = "0.3" +tokio-stream = "0.1" + +tokio.workspace = true + +[features] +default = ["tokio"] + +std = ["alloc", "futures-core/std"] +alloc = ["futures-core/alloc"] + +tokio = ["dep:tokio", "std"] + +[lints] +workspace = true diff --git a/ts_future_util/README.md b/ts_future_util/README.md new file mode 100644 index 00000000..34baca95 --- /dev/null +++ b/ts_future_util/README.md @@ -0,0 +1,3 @@ +# `ts_future_util` + +Various utils for futures. diff --git a/ts_future_util/src/debounce.rs b/ts_future_util/src/debounce.rs new file mode 100644 index 00000000..23f2794e --- /dev/null +++ b/ts_future_util/src/debounce.rs @@ -0,0 +1,429 @@ +//! Modeled after +//! , with the +//! deviation that this yields on both the debounce window leading edge and trailing edge (if there +//! was anything received inside the window). + +use core::{ + pin::Pin, + task::{Context, Poll}, + time::Duration, +}; + +use futures_core::Stream; + +/// [`Debounce`] using [`tokio::time::sleep`] as its timer. +#[cfg(feature = "tokio")] +pub type TokioDebounce = Debounce tokio::time::Sleep, tokio::time::Sleep>; + +pin_project_lite::pin_project! { + /// A wrapper that debounces stream items. + /// + /// The initial element is always yielded immediately. Successive elements within a configurable + /// debounce window are suppressed. If any elements were yielded during the window, the last one + /// is yielded at the end of the window (starting a new window). All previous elements within + /// the window are dropped. + /// + /// # Examples + /// + /// When elements are coming from the inner stream slower than the debounce window, `Debounce` + /// is transparent: + /// + /// ```text + /// input: e1 e2 + /// windows: |<-- w1 -->| |<-- w2 -->| + /// output: e1 e2 + /// ``` + /// + /// When elements come faster than the window, they're suppressed until the window elapses: + /// + /// ```text + /// input: e1 e2 e3 e4 e5 + /// windows: |<-- w1 -->|<-- w2 -->| + /// output: e1 e3 e5 + /// ``` + pub struct Debounce + where + S: Stream, + S: ?Sized, + { + slot: Option, + stream_done: bool, + + #[pin] + timer: Option, + make_timer: F, + window: Duration, + + #[pin] + stream: S, + } +} + +impl Debounce +where + S: Stream, +{ + /// Build a new [`Debounce`] with the given timer creation function and window duration. + pub fn new(stream: S, make_timer: F, window: Duration) -> Self { + Self { + stream, + timer: None, + slot: None, + stream_done: false, + make_timer, + window, + } + } +} + +#[cfg(feature = "tokio")] +impl Debounce tokio::time::Sleep, tokio::time::Sleep> +where + S: Stream, +{ + /// Build a new [`Debounce`] using a tokio timer. + pub fn tokio(stream: S, window: Duration) -> Self { + Debounce::new(stream, tokio::time::sleep, window) + } +} + +impl Stream for Debounce +where + S: Stream, + F: FnMut(Duration) -> Timer, + Timer: Future, +{ + type Item = S::Item; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let mut slf = self.as_mut().project(); + + // Invariant: loop falls through when the stream is polled to exhaustion (once). + loop { + if *slf.stream_done { + // If there's nothing in the slot, there never will be (the stream has stopped). + // Drop the timer if we have one. + if slf.slot.is_none() { + slf.timer.set(None); + } + + // If the stream is done and there's no timer, we're done, stop polling. + if slf.timer.is_none() { + return Poll::Ready(None); + } + } + + if let Some(timer) = slf.timer.as_mut().as_pin_mut() { + match timer.poll(cx) { + Poll::Pending => { + // If the stream is done, the timer drives the future polling exclusively. + // We're just waiting for this debounce window to end in order to yield the + // final item. + if *slf.stream_done { + return Poll::Pending; + } + + // Otherwise, fall through to poll the stream: we shouldn't yield anything + // from it until the timer is ready, but we need it to make progress and + // update the slot if it produces items. + } + Poll::Ready(_) => { + slf.timer.set(None); + + // Timer was the first thing to be polled; there may be items available in + // the stream. Drain everything that's synchronously available before seeing + // if we can yield an item from the slot. + while !*slf.stream_done + && let Poll::Ready(x) = slf.stream.as_mut().poll_next(cx) + { + if let Some(item) = x { + *slf.slot = Some(item); + } else { + *slf.stream_done = true; + } + } + + // If we have an item in the slot, yield it and start a new window timer. + if let Some(item) = slf.slot.take() { + // Don't bother with a new window if we know the stream won't yield + // anything else. + if !*slf.stream_done { + slf.timer.set(Some((slf.make_timer)(*slf.window))); + } + + return Poll::Ready(Some(item)); + } + } + } + } + + // Invariant: getting to this point means that an item is expected from the stream. If + // it's done, there is no more work to do. + if *slf.stream_done { + debug_assert!(slf.timer.is_none()); + return Poll::Ready(None); + } + + // Drain any ready items out of the stream. + while let Some(item) = core::task::ready!(slf.stream.as_mut().poll_next(cx)) { + // If we don't have a timer, we're not in a debounce window right now. + // Start one and immediately yield the item. + if slf.timer.is_none() { + debug_assert!(slf.slot.is_none()); + slf.timer.set(Some((slf.make_timer)(*slf.window))); + + return Poll::Ready(Some(item)); + } + + // We're in a debounce window: just update the value in the slot and try to read + // more out of the stream. + *slf.slot = Some(item); + } + + // Stream has just been polled to exhaustion. Fall through the loop to try to poll the + // timer. + *slf.stream_done = true; + } + } + + fn size_hint(&self) -> (usize, Option) { + S::size_hint(&self.stream) + } +} + +/// Extension trait to add debounce wrapper methods to [`Stream`] implementors. +/// +/// See [`Debounce`] for details on the returned stream wrapper. +pub trait DebounceExt: Stream + Sized { + /// Debounce this stream using a window of duration `window`. Uses `tokio::time::sleep` as the + /// timer future. + /// + /// See [`DebounceExt::debounce_with`] for more info on debouncing. + #[cfg(feature = "tokio")] + fn debounce( + self, + window: Duration, + ) -> Debounce tokio::time::Sleep, tokio::time::Sleep> { + Debounce::tokio(self, window) + } + + /// Debounce this stream with a configurable timer function. + /// + /// This debounce has both leading- and trailing-edge behavior. In the idle state, the first + /// item from the underlying stream is yielded immediately. This puts the stream into a debounce + /// state, where all items from the underlying stream are suppressed. At the end of the debounce + /// window, if any stream items were suppressed, the most recent one is yielded and a new + /// debounce window starts. If not, nothing is yielded and the stream returns to the idle state. + /// + /// This is useful e.g. in cases where the underlying stream is a series of full-state + /// snapshots which may come in large bursts. In these cases, you don't necessarily need to see + /// all the items from the stream, but you will always need to receive the final state in order + /// to have an accurate view of the world in steady-state. The leading-edge behavior improves + /// responsiveness in non-bursty scenarios. + /// + /// # Timer + /// + /// The return value of the timer function is any future: when it completes, the debounce window + /// is considered elapsed. Generally, this will just be a constructor for a timer which elapses + /// after the passed-in [`Duration`]. + fn debounce_with(self, window: Duration, make_timer: F) -> Debounce + where + F: FnMut(Duration) -> Timer, + Timer: Future, + { + Debounce::new(self, make_timer, window) + } +} + +impl DebounceExt for T where T: Stream + Sized {} + +#[cfg(test)] +mod test { + use std::{boxed::Box, sync::Arc}; + + use futures_util::StreamExt; + use tokio::sync::{Mutex, mpsc}; + use tokio_stream::wrappers::UnboundedReceiverStream; + + use super::*; + + type BoxFut = Pin + Send>>; + type BoxDebounce = Debounce BoxFut>, BoxFut>; + + #[track_caller] + fn with_noop_cx(f: impl FnOnce(&mut Context) -> T) -> T { + let mut cx = Context::from_waker(core::task::Waker::noop()); + f(&mut cx) + } + + struct TestStream { + tx: Option>, + timer: mpsc::UnboundedSender<()>, + stream: Pin>>>, + } + + impl TestStream { + fn new() -> Self + where + T: Send + 'static, + { + let (stream_tx, stream_rx) = mpsc::unbounded_channel(); + + // "Timer channel" simulates a timer without requiring an actual timer. Values yielded from + // the channel represent the timer elapsing. + let (timer_tx, timer_rx) = mpsc::unbounded_channel::<()>(); + let timer = Arc::new(Mutex::new(timer_rx)); + + let debounced = UnboundedReceiverStream::new(stream_rx).debounce_with( + Duration::from_secs(1), + Box::new(move |_dur| { + let rx = timer.clone(); + + Box::pin(async move { + let mut rx = rx.lock().await; + rx.recv().await.unwrap(); + }) as BoxFut + }) as Box BoxFut>, + ); + + Self { + tx: Some(stream_tx), + timer: timer_tx, + stream: Box::pin(debounced), + } + } + + /// Assert that the stream is currently pending. + #[track_caller] + fn assert_pending(&mut self) { + with_noop_cx(|cx| { + assert!(self.stream.poll_next_unpin(cx).is_pending()); + }); + } + + /// Assert the value of the next item in the stream, which must be available + /// synchronously (poll-once). + #[track_caller] + fn assert_next(&mut self, value: T) + where + T: PartialEq + core::fmt::Debug, + { + assert_eq!( + with_noop_cx(|cx| self.stream.poll_next_unpin(cx)), + Poll::Ready(Some(value)), + ) + } + + /// Send the value through the sender and assert it comes back through the stream + /// (synchronously/immediately: it must not pend). + #[track_caller] + fn assert_roundtrip(&mut self, value: T) + where + T: PartialEq + Clone + core::fmt::Debug, + { + self.send(value.clone()); + self.assert_next(value); + } + + /// Assert that the stream is done. + #[track_caller] + fn assert_done(&mut self) + where + T: PartialEq + core::fmt::Debug, + { + assert_eq!( + with_noop_cx(|cx| self.stream.poll_next_unpin(cx)), + Poll::Ready(None), + ); + assert!(self.stream.stream_done); + } + + fn send(&self, t: T) { + self.tx.as_ref().unwrap().send(t).unwrap(); + } + + fn release_timer(&self) { + self.timer.send(()).unwrap(); + } + + fn drop_sender(&mut self) { + self.tx.take().unwrap(); + } + } + + /// Send values through a debounced stream slowly enough to never trigger the debounce behavior. + #[test] + fn slow_transparent() { + let mut ts = TestStream::new(); + + for i in 0..30 { + // Nothing available in the stream (stream channel is empty): nothing yielded. + ts.assert_pending(); + ts.assert_roundtrip(i); + ts.release_timer(); + } + + // When the underlying stream closes, the debounced one does as well (if nothing is in the + // slot). + ts.drop_sender(); + ts.assert_done(); + } + + /// Debounce one message. + #[test] + fn single_debounce() { + let mut ts = TestStream::new(); + ts.assert_pending(); + + // Initial message comes back. + ts.assert_roundtrip(1234); + + // Second message is in window: nothing ready. + ts.send(5678); + ts.assert_pending(); + + // Release timer, message comes back + ts.release_timer(); + ts.assert_next(5678); + + // We're in another window: dropping the underlying stream with no items available will + // immediately end the debounced stream. + ts.assert_pending(); + ts.drop_sender(); + ts.assert_done(); + } + + /// Debounce many messages in one window. + #[test] + fn multi_debounce() { + let mut ts = TestStream::new(); + ts.assert_pending(); + + // Initial message comes back. + ts.assert_roundtrip(1234); + + // Second message is in window: nothing ready. + ts.send(5); + ts.send(6); + ts.send(7); + ts.send(8); + ts.assert_pending(); + + // Release timer, only _last_ message comes back + ts.release_timer(); + ts.assert_next(8); + ts.assert_pending(); + + // In another window – send a value but then drop the sender. + ts.send(9); + ts.drop_sender(); + ts.assert_pending(); + + // Even after we drop tx, the slot item still yields. + ts.release_timer(); + ts.assert_next(9); + + // But because we dropped tx, the stream could finish immediately. + ts.assert_done(); + } +} diff --git a/ts_future_util/src/lib.rs b/ts_future_util/src/lib.rs new file mode 100644 index 00000000..65c21020 --- /dev/null +++ b/ts_future_util/src/lib.rs @@ -0,0 +1,14 @@ +#![doc = include_str!("../README.md")] +#![no_std] + +#[cfg(feature = "alloc")] +extern crate alloc; + +#[cfg(any(feature = "std", test))] +extern crate std; + +mod debounce; + +#[cfg(feature = "tokio")] +pub use debounce::TokioDebounce; +pub use debounce::{Debounce, DebounceExt}; From e5da9b7bd55b05bd05775efe842580f032c62ad2 Mon Sep 17 00:00:00 2001 From: Nathan Perry Date: Tue, 15 Sep 2026 02:33:22 -0400 Subject: [PATCH 2/3] ts_future_util/debounce: write debounce_fold Signed-off-by: Nathan Perry Change-Id: I8ea03ed2be24981ce41e6a129f2fa43f6a6a6964 --- ts_future_util/src/debounce.rs | 179 ++++++++++++++++++++++++++++----- 1 file changed, 155 insertions(+), 24 deletions(-) diff --git a/ts_future_util/src/debounce.rs b/ts_future_util/src/debounce.rs index 23f2794e..310c611d 100644 --- a/ts_future_util/src/debounce.rs +++ b/ts_future_util/src/debounce.rs @@ -11,9 +11,19 @@ use core::{ use futures_core::Stream; +/// Signature of the tokio sleep function, aliased for convenience. +#[cfg(feature = "tokio")] +pub type TokioSleepFn = fn(Duration) -> tokio::time::Sleep; + /// [`Debounce`] using [`tokio::time::sleep`] as its timer. #[cfg(feature = "tokio")] -pub type TokioDebounce = Debounce tokio::time::Sleep, tokio::time::Sleep>; +pub type TokioDebounce::Item> = + Debounce::Item>>; + +/// Type of the default fold function, which just passes through stream updates. +/// +/// Convenience alias. +pub type DefaultFold = fn(&mut Option, U); pin_project_lite::pin_project! { /// A wrapper that debounces stream items. @@ -41,12 +51,14 @@ pin_project_lite::pin_project! { /// windows: |<-- w1 -->|<-- w2 -->| /// output: e1 e3 e5 /// ``` - pub struct Debounce - where - S: Stream, - S: ?Sized, - { - slot: Option, + /// + /// # Customizing updates (`debounce_fold`) + /// + /// It's possible to provide a custom function that defines how updates are applied to + /// the slot (pending-value) in order to produce different semantics than the most- + /// recent stream value: see [`DebounceExt::debounce_fold_with`]. + pub struct Debounce { + slot: Option, stream_done: bool, #[pin] @@ -54,12 +66,14 @@ pin_project_lite::pin_project! { make_timer: F, window: Duration, + fold: Fold, + #[pin] stream: S, } } -impl Debounce +impl Debounce> where S: Stream, { @@ -72,12 +86,34 @@ where stream_done: false, make_timer, window, + fold: fold_passthru, + } + } +} + +impl Debounce +where + S: Stream, +{ + /// Update the fold function. + fn with_fold(self, fold: NewFold) -> Debounce + where + NewFold: FnMut(&mut Option, S::Item), + { + Debounce { + fold, + stream: self.stream, + make_timer: self.make_timer, + slot: None, + window: self.window, + timer: self.timer, + stream_done: self.stream_done, } } } #[cfg(feature = "tokio")] -impl Debounce tokio::time::Sleep, tokio::time::Sleep> +impl Debounce> where S: Stream, { @@ -87,13 +123,14 @@ where } } -impl Stream for Debounce +impl Stream for Debounce where S: Stream, F: FnMut(Duration) -> Timer, Timer: Future, + Fold: FnMut(&mut Option, S::Item), { - type Item = S::Item; + type Item = T; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let mut slf = self.as_mut().project(); @@ -137,7 +174,7 @@ where && let Poll::Ready(x) = slf.stream.as_mut().poll_next(cx) { if let Some(item) = x { - *slf.slot = Some(item); + (slf.fold)(slf.slot, item); } else { *slf.stream_done = true; } @@ -157,7 +194,7 @@ where } } - // Invariant: getting to this point means that an item is expected from the stream. If + // Invariant: getting to this point means that we want to poll an item from the stream. If // it's done, there is no more work to do. if *slf.stream_done { debug_assert!(slf.timer.is_none()); @@ -166,18 +203,21 @@ where // Drain any ready items out of the stream. while let Some(item) = core::task::ready!(slf.stream.as_mut().poll_next(cx)) { - // If we don't have a timer, we're not in a debounce window right now. - // Start one and immediately yield the item. - if slf.timer.is_none() { - debug_assert!(slf.slot.is_none()); + // Immediately fold the value into the slot. + (slf.fold)(slf.slot, item); + + // If we don't have a timer, we're not in a debounce window right now. If the fold + // function slotted a value, start a timer and immediately yield the value. + if slf.timer.is_none() + && let Some(item) = slf.slot.take() + { slf.timer.set(Some((slf.make_timer)(*slf.window))); return Poll::Ready(Some(item)); } - // We're in a debounce window: just update the value in the slot and try to read - // more out of the stream. - *slf.slot = Some(item); + // Otherwise, either the fold fn cleared the slot or we're in a debounce window: + // try to read more out of the stream. } // Stream has just been polled to exhaustion. Fall through the loop to try to poll the @@ -187,7 +227,11 @@ where } fn size_hint(&self) -> (usize, Option) { - S::size_hint(&self.stream) + // We may yield fewer items than the underlying stream's lower bound, but we will never + // yield more than its upper bound. + let (_lower, upper) = S::size_hint(&self.stream); + + (0, upper) } } @@ -203,10 +247,29 @@ pub trait DebounceExt: Stream + Sized { fn debounce( self, window: Duration, - ) -> Debounce tokio::time::Sleep, tokio::time::Sleep> { + ) -> Debounce> { Debounce::tokio(self, window) } + /// Debounce this stream using a window of duration `window` and the given `fold` + /// function. The `fold` function is used to specify how the debounced item is updated + /// (see [`DebounceExt::fold_debounce_with`] for more info). + /// + /// Uses `tokio::time::sleep` as the timer future. + /// + /// See [`DebounceExt::debounce_with`] for more info on debouncing generally. + #[cfg(feature = "tokio")] + fn fold_debounce( + self, + window: Duration, + fold_fn: Fold, + ) -> Debounce + where + Fold: FnMut(&mut Option, Self::Item), + { + Debounce::tokio(self, window).with_fold(fold_fn) + } + /// Debounce this stream with a configurable timer function. /// /// This debounce has both leading- and trailing-edge behavior. In the idle state, the first @@ -226,17 +289,84 @@ pub trait DebounceExt: Stream + Sized { /// The return value of the timer function is any future: when it completes, the debounce window /// is considered elapsed. Generally, this will just be a constructor for a timer which elapses /// after the passed-in [`Duration`]. - fn debounce_with(self, window: Duration, make_timer: F) -> Debounce + /// + /// # Customizing updates + /// + /// For customizing how values from the underlying stream are folded into the + /// [`Debounce`]'s pending slot, see [`DebounceExt::fold_debounce_with`]. + fn debounce_with( + self, + window: Duration, + make_timer: F, + ) -> Debounce> where F: FnMut(Duration) -> Timer, Timer: Future, { Debounce::new(self, make_timer, window) } + + /// Debounce this stream with the given timer function and fold function. + /// + /// See [`DebounceExt::debounce_with`] for the essential details of how the debouncing + /// functionality works and how it interacts with the timer function. + /// + /// The fold function describes how the slot value is updated by items yielded from the + /// underlying stream. The slot value is the value that will be yielded whenever the + /// current debounce period is over. If not currently in a debounce period, setting the + /// slot value to `Some` causes that value to yield immediately and a debounce period to + /// start. + /// + /// Standard behavior (plain [`DebounceExt::debounce_with`]) is achieved by setting the + /// slot value to `Some(next_value)`. + /// + /// Setting the slot item to `None` means that when the current debounce period ends, + /// nothing will be yielded (if this is still the slot value). This can be used to + /// implement cancellation semantics. + /// + /// # Examples + /// + /// ```rust + /// # use std::time::Duration; + /// # use ts_future_util::DebounceExt; + /// # use futures_util::StreamExt; + /// + /// # fn my_timer(dur: Duration) -> futures_util::future::Ready<()> { futures_util::future::ready(()) } + /// # let xs = tokio::runtime::Runtime::new().unwrap().block_on(async move { + /// let xs = futures_util::stream::iter([1u8, 2, 3, 4]) + /// .fold_debounce_with(Duration::from_millis(100), my_timer, |slot: &mut Option, x| { + /// *slot.get_or_insert_default() += x; + /// }) + /// .collect::>() + /// .await; + /// # xs + /// # }); + /// + /// // The first value is yielded right away, then the next 3 items are + /// // debounced and summed (2 + 3 + 4 = 9). + /// assert_eq!(xs, [1, 9]); + /// ``` + fn fold_debounce_with( + self, + window: Duration, + make_timer: F, + fold: Fold, + ) -> Debounce + where + F: FnMut(Duration) -> Timer, + Timer: Future, + Fold: FnMut(&mut Option, Self::Item), + { + Debounce::new(self, make_timer, window).with_fold(fold) + } } impl DebounceExt for T where T: Stream + Sized {} +fn fold_passthru(acc: &mut Option, next: T) { + *acc = Some(next); +} + #[cfg(test)] mod test { use std::{boxed::Box, sync::Arc}; @@ -248,7 +378,8 @@ mod test { use super::*; type BoxFut = Pin + Send>>; - type BoxDebounce = Debounce BoxFut>, BoxFut>; + type BoxDebounce::Item> = + Debounce BoxFut>, BoxFut, DefaultFold>; #[track_caller] fn with_noop_cx(f: impl FnOnce(&mut Context) -> T) -> T { From 7ecb90870bfb4a0fe56c0464f5e6fe9bb45fda11 Mon Sep 17 00:00:00 2001 From: Nathan Perry Date: Mon, 14 Sep 2026 20:59:27 -0400 Subject: [PATCH 3/3] workspace: add futures-core, tokio-stream deps Also propagate these to dependent crates as `workspace = true` deps, and fix a few that manually specified a version. Signed-off-by: Nathan Perry Change-Id: I758b99269ffde8de5a7acb211ec6ddd06a6a6964 --- Cargo.toml | 2 ++ ts_future_util/Cargo.toml | 8 ++++---- ts_netmon/Cargo.toml | 4 ++-- ts_runtime/Cargo.toml | 2 +- 4 files changed, 9 insertions(+), 7 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index eb9bfb26..705256ea 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,6 +74,7 @@ crypto_box = "0.9" divan = "0.1" flume = "0.12" futures = { version = "0.3", default-features = false } +futures-core = { version = "0.3", default-features = false } futures-util = { version = "0.3", default-features = false } gethostname = "1.1" hashbrown = "0.17" @@ -101,6 +102,7 @@ smoltcp = { version = "0.13", default-features = false } thiserror = { version = "2", default-features = false } tokio = { version = "1", default-features = false, features = ["rt-multi-thread", "macros"] } tokio-util = { version = "0.7", default-features = false } +tokio-stream = { version = "0.1", default-features = false } tracing = { version = "0.1", default-features = false, features = ["attributes"] } tracing-test = "0.2" url = { version = "2", default-features = false } diff --git a/ts_future_util/Cargo.toml b/ts_future_util/Cargo.toml index 7effa1b0..a43ad9e0 100644 --- a/ts_future_util/Cargo.toml +++ b/ts_future_util/Cargo.toml @@ -11,14 +11,14 @@ license.workspace = true rust-version.workspace = true [dependencies] -futures-core = { version = "0.3", default-features = false } -pin-project-lite = "0.2" +futures-core.workspace = true +pin-project-lite.workspace = true tokio = { workspace = true, optional = true, features = ["time"] } [dev-dependencies] -futures-util = "0.3" -tokio-stream = "0.1" +futures-util.workspace = true +tokio-stream = { workspace = true, features = ["default"] } tokio.workspace = true diff --git a/ts_netmon/Cargo.toml b/ts_netmon/Cargo.toml index 5cf88dcb..fb9890b9 100644 --- a/ts_netmon/Cargo.toml +++ b/ts_netmon/Cargo.toml @@ -15,10 +15,10 @@ cfg-if.workspace = true flume.workspace = true futures-util.workspace = true ipnet.workspace = true -pin-project-lite = "0.2" +pin-project-lite.workspace = true smallvec.workspace = true tokio.workspace = true -tokio-stream = "0.1" +tokio-stream.workspace = true tracing.workspace = true [dev-dependencies] diff --git a/ts_runtime/Cargo.toml b/ts_runtime/Cargo.toml index f25dd75f..6e07660c 100644 --- a/ts_runtime/Cargo.toml +++ b/ts_runtime/Cargo.toml @@ -39,7 +39,7 @@ kameo_actors = "0.6" thiserror.workspace = true tokio.workspace = true tokio-util = { workspace = true, features = ["rt"] } -tokio-stream = { version = "0.1", default-features = false, features = ["time", "sync"] } +tokio-stream = { workspace = true, features = ["time", "sync"] } tracing.workspace = true smallvec.workspace = true smol_str = "0.3"