From 8fac137f4535b06bac36730b6b7828730c8e7aef Mon Sep 17 00:00:00 2001 From: Mohamed Koubaa <11414628+koubaa@users.noreply.github.com> Date: Fri, 31 Jul 2026 16:42:48 -0500 Subject: [PATCH 1/5] Don't warn on Alive::drop if aborted --- rs/moq-net/src/model/group.rs | 103 +++++++++++++++++++++++++++++++-- rs/moq-net/src/model/track.rs | 106 ++++++++++++++++++++++++++++++++-- 2 files changed, 198 insertions(+), 11 deletions(-) diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 440b830c7f..71387c2534 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -241,19 +241,32 @@ struct Alive { impl Drop for Alive { fn drop(&mut self) { + // `Producer::abort()` records `abort` then closes the channel. `write()` / + // `modify()` then fail, so the cleanliness check must use `read()` or a + // deliberate abort looks like an unfinished drop once the producer is gone. + { + let state = self.state.read(); + if state.fin.is_some() || state.abort.is_some() { + return; + } + } // See track::Alive: the last producer dropping without a clean finish releases // the cached frames so a stale consumer can't pin their buffers forever. A // finished group keeps its cache so consumers can drain. - if let Ok(mut state) = modify(&self.state) - && state.fin.is_none() - { - // Dropped without finish() or abort(), so consumers will see - // Error::Dropped mid-group. Deliberate ends go through finish()/abort(). + if let Ok(mut state) = modify(&self.state) { + if state.fin.is_some() || state.abort.is_some() { + return; + } tracing::warn!( sequence = self.info.sequence, "group::Producer dropped without finish() or abort()" ); state.release(); + } else { + tracing::warn!( + sequence = self.info.sequence, + "group::Producer dropped without finish() or abort()" + ); } } } @@ -802,6 +815,55 @@ mod test { use super::*; use bytes::Bytes; use futures::FutureExt; + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; + use tracing::field::{Field, Visit}; + use tracing::span::{Attributes, Id, Record}; + use tracing::{Event, Level, Metadata, Subscriber}; + + /// Count `group::Producer` unfinished-drop WARN events while running `f`. + /// Uses only the existing `tracing` dependency (no tracing-subscriber). + fn count_drop_warnings(f: impl FnOnce()) -> usize { + struct Count(std::sync::Arc); + struct Msg(bool); + impl Visit for Msg { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + let s = format!("{value:?}"); + if s.contains("group::Producer dropped without finish") { + self.0 = true; + } + } + } + fn record_str(&mut self, field: &Field, value: &str) { + if field.name() == "message" && value.contains("group::Producer dropped without finish") { + self.0 = true; + } + } + } + impl Subscriber for Count { + fn enabled(&self, metadata: &Metadata<'_>) -> bool { + *metadata.level() <= Level::WARN + } + fn new_span(&self, _span: &Attributes<'_>) -> Id { + Id::from_u64(1) + } + fn record(&self, _span: &Id, _values: &Record<'_>) {} + fn record_follows_from(&self, _span: &Id, _follows: &Id) {} + fn event(&self, event: &Event<'_>) { + let mut msg = Msg(false); + event.record(&mut msg); + if msg.0 { + self.0.fetch_add(1, AtomicOrdering::SeqCst); + } + } + fn enter(&self, _span: &Id) {} + fn exit(&self, _span: &Id) {} + } + + let hits = std::sync::Arc::new(AtomicUsize::new(0)); + tracing::subscriber::with_default(Count(hits.clone()), f); + hits.load(AtomicOrdering::SeqCst) + } #[test] fn basic_frame_reading() { @@ -958,6 +1020,37 @@ mod test { assert!(matches!(result, Err(crate::Error::Dropped))); } + #[test] + fn drop_after_abort_does_not_warn() { + let warns = count_drop_warnings(|| { + let producer = Info { sequence: 0 }.produce(); + let keep = producer.clone(); + let mut writer = producer.clone(); + writer + .write_frame(Timestamp::ZERO, Bytes::from_static(b"data")) + .unwrap(); + let _consumer = producer.consume(); + writer.abort(crate::Error::Cancel).unwrap(); + drop(keep); + }); + assert_eq!(warns, 0, "abort-then-drop must not emit unfinished-producer WARN"); + } + + #[test] + fn drop_unfinished_warns() { + let warns = count_drop_warnings(|| { + let producer = Info { sequence: 0 }.produce(); + let mut writer = producer.clone(); + writer + .write_frame(Timestamp::ZERO, Bytes::from_static(b"data")) + .unwrap(); + let _consumer = producer.consume(); + drop(writer); + drop(producer); + }); + assert!(warns >= 1, "unfinished drop must emit unfinished-producer WARN"); + } + #[test] fn drop_finished_keeps_cached_frames() { let mut producer = Info { sequence: 0 }.produce(); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 48ffa8fce8..445f596a59 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1459,22 +1459,36 @@ impl Drop for Alive { if !self.published.load(Ordering::Relaxed) { return; } + // `Producer::abort()` records `abort` then closes the channel. `write()` then + // fails, so the cleanliness check must use `read()` or a deliberate abort is + // indistinguishable from an unfinished drop once the producer is gone. + { + let state = self.state.read(); + if state.final_sequence.is_some() || state.abort.is_some() { + return; + } + } // The last producer going away without finishing is an abrupt teardown: // release the cached groups so a stale consumer can't pin them (and their // frame buffers) forever, the same as an explicit abort. A cleanly // finished track keeps its cache so consumers can still drain it. - if let Ok(mut state) = self.state.write() - && state.final_sequence.is_none() - { - // Dropped without finish() or abort(), so consumers will see - // Error::Dropped instead of a clean end. Deliberate ends go through - // finish()/abort(). + if let Ok(mut state) = self.state.write() { + if state.final_sequence.is_some() || state.abort.is_some() { + return; + } tracing::warn!( track = %self.name, "track::Producer dropped without finish() or abort()" ); state.clear_cache(); state.datagrams.clear(); + } else { + // Closed without finish/abort — still an abrupt teardown (e.g. cancelled + // before abort ran). After a real abort(), the read() check above returned. + tracing::warn!( + track = %self.name, + "track::Producer dropped without finish() or abort()" + ); } } } @@ -2739,6 +2753,56 @@ mod test { .expect("track was closed") } + /// Count `track::Producer` unfinished-drop WARN events while running `f`. + /// Uses only the existing `tracing` dependency (no tracing-subscriber). + fn count_drop_warnings(f: impl FnOnce()) -> usize { + use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; + use tracing::field::{Field, Visit}; + use tracing::span::{Attributes, Id, Record}; + use tracing::{Event, Level, Metadata, Subscriber}; + + struct Count(std::sync::Arc); + struct Msg(bool); + impl Visit for Msg { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + let s = format!("{value:?}"); + if s.contains("track::Producer dropped without finish") { + self.0 = true; + } + } + } + fn record_str(&mut self, field: &Field, value: &str) { + if field.name() == "message" && value.contains("track::Producer dropped without finish") { + self.0 = true; + } + } + } + impl Subscriber for Count { + fn enabled(&self, metadata: &Metadata<'_>) -> bool { + *metadata.level() <= Level::WARN + } + fn new_span(&self, _span: &Attributes<'_>) -> Id { + Id::from_u64(1) + } + fn record(&self, _span: &Id, _values: &Record<'_>) {} + fn record_follows_from(&self, _span: &Id, _follows: &Id) {} + fn event(&self, event: &Event<'_>) { + let mut msg = Msg(false); + event.record(&mut msg); + if msg.0 { + self.0.fetch_add(1, AtomicOrdering::SeqCst); + } + } + fn enter(&self, _span: &Id) {} + fn exit(&self, _span: &Id) {} + } + + let hits = std::sync::Arc::new(AtomicUsize::new(0)); + tracing::subscriber::with_default(Count(hits.clone()), f); + hits.load(AtomicOrdering::SeqCst) + } + #[tokio::test] async fn append_datagram_shares_group_sequence() { let mut producer = track_producer("test", None); @@ -3399,6 +3463,36 @@ mod test { assert!(matches!(result, Err(Error::Dropped))); } + #[tokio::test] + async fn drop_after_abort_does_not_warn() { + // abort() closes the channel after recording `abort`. Drop must treat that as + // clean via read(); without the abort check this emits a false WARN. + let warns = count_drop_warnings(|| { + let producer = track_producer("test", None); + let keep = producer.clone(); + let mut writer = producer.clone(); + let mut group = writer.append_group().unwrap(); + group.finish().unwrap(); + let _consumer = producer.subscribe(None); + writer.abort(Error::Cancel).unwrap(); + drop(keep); + }); + assert_eq!(warns, 0, "abort-then-drop must not emit unfinished-producer WARN"); + } + + #[tokio::test] + async fn drop_unfinished_warns() { + let warns = count_drop_warnings(|| { + let producer = track_producer("test", None); + let mut writer = producer.clone(); + writer.append_group().unwrap(); + let _consumer = producer.subscribe(None); + drop(writer); + drop(producer); + }); + assert!(warns >= 1, "unfinished drop must emit unfinished-producer WARN"); + } + #[tokio::test] async fn drop_finished_keeps_cached_groups() { let mut producer = track_producer("test", None); From be00b09a3cb119f59bbebe8cb4d7e82671c9dd5b Mon Sep 17 00:00:00 2001 From: Mohamed Koubaa <11414628+koubaa@users.noreply.github.com> Date: Sun, 2 Aug 2026 18:49:43 -0500 Subject: [PATCH 2/5] one mutex --- rs/moq-net/src/model/group.rs | 21 ++++++++++----------- rs/moq-net/src/model/track.rs | 24 +++++++++++------------- 2 files changed, 21 insertions(+), 24 deletions(-) diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 71387c2534..9b6fe2b655 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -241,19 +241,13 @@ struct Alive { impl Drop for Alive { fn drop(&mut self) { - // `Producer::abort()` records `abort` then closes the channel. `write()` / - // `modify()` then fail, so the cleanliness check must use `read()` or a - // deliberate abort looks like an unfinished drop once the producer is gone. - { - let state = self.state.read(); - if state.fin.is_some() || state.abort.is_some() { - return; - } - } // See track::Alive: the last producer dropping without a clean finish releases // the cached frames so a stale consumer can't pin their buffers forever. A // finished group keeps its cache so consumers can drain. - if let Ok(mut state) = modify(&self.state) { + // + // Check Ok and Err: Ok is unreachable after a deliberate close. + match self.state.write() { + Ok(mut state) => { if state.fin.is_some() || state.abort.is_some() { return; } @@ -262,7 +256,11 @@ impl Drop for Alive { "group::Producer dropped without finish() or abort()" ); state.release(); - } else { + } + Err(state) => { + if state.fin.is_some() || state.abort.is_some() { + return; + } tracing::warn!( sequence = self.info.sequence, "group::Producer dropped without finish() or abort()" @@ -270,6 +268,7 @@ impl Drop for Alive { } } } +} impl std::ops::Deref for Producer { type Target = Info; diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 445f596a59..f2cde50c3e 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1459,20 +1459,15 @@ impl Drop for Alive { if !self.published.load(Ordering::Relaxed) { return; } - // `Producer::abort()` records `abort` then closes the channel. `write()` then - // fails, so the cleanliness check must use `read()` or a deliberate abort is - // indistinguishable from an unfinished drop once the producer is gone. - { - let state = self.state.read(); - if state.final_sequence.is_some() || state.abort.is_some() { - return; - } - } // The last producer going away without finishing is an abrupt teardown: // release the cached groups so a stale consumer can't pin them (and their // frame buffers) forever, the same as an explicit abort. A cleanly // finished track keeps its cache so consumers can still drain it. - if let Ok(mut state) = self.state.write() { + // + // `abort()`/`finish()` close the channel, so `write()` returns `Err(Ref)`. + // Check Ok and Err: Ok is unreachable after a deliberate close. + match self.state.write() { + Ok(mut state) => { if state.final_sequence.is_some() || state.abort.is_some() { return; } @@ -1482,9 +1477,11 @@ impl Drop for Alive { ); state.clear_cache(); state.datagrams.clear(); - } else { - // Closed without finish/abort — still an abrupt teardown (e.g. cancelled - // before abort ran). After a real abort(), the read() check above returned. + } + Err(state) => { + if state.final_sequence.is_some() || state.abort.is_some() { + return; + } tracing::warn!( track = %self.name, "track::Producer dropped without finish() or abort()" @@ -1492,6 +1489,7 @@ impl Drop for Alive { } } } +} /// Aggregate every live subscriber's preferences into the most demanding request. /// From 88ae2949ca0e7d06f3d7b936c777b743bf1183a3 Mon Sep 17 00:00:00 2001 From: Mohamed Koubaa <11414628+koubaa@users.noreply.github.com> Date: Sun, 2 Aug 2026 18:49:54 -0500 Subject: [PATCH 3/5] dedup test helper --- rs/moq-net/src/model/group.rs | 80 ++++++--------------------- rs/moq-net/src/model/mod.rs | 3 + rs/moq-net/src/model/test_tracing.rs | 80 +++++++++++++++++++++++++++ rs/moq-net/src/model/track.rs | 83 ++++++---------------------- 4 files changed, 116 insertions(+), 130 deletions(-) create mode 100644 rs/moq-net/src/model/test_tracing.rs diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 9b6fe2b655..1c1fafb976 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -248,27 +248,27 @@ impl Drop for Alive { // Check Ok and Err: Ok is unreachable after a deliberate close. match self.state.write() { Ok(mut state) => { - if state.fin.is_some() || state.abort.is_some() { - return; - } - tracing::warn!( - sequence = self.info.sequence, - "group::Producer dropped without finish() or abort()" - ); - state.release(); + if state.fin.is_some() || state.abort.is_some() { + return; + } + tracing::warn!( + sequence = self.info.sequence, + "group::Producer dropped without finish() or abort()" + ); + state.release(); } Err(state) => { if state.fin.is_some() || state.abort.is_some() { return; } - tracing::warn!( - sequence = self.info.sequence, - "group::Producer dropped without finish() or abort()" - ); + tracing::warn!( + sequence = self.info.sequence, + "group::Producer dropped without finish() or abort()" + ); + } } } } -} impl std::ops::Deref for Producer { type Target = Info; @@ -813,56 +813,8 @@ impl Fetch { mod test { use super::*; use bytes::Bytes; + use crate::model::test_tracing::count_drop_warnings; use futures::FutureExt; - use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; - use tracing::field::{Field, Visit}; - use tracing::span::{Attributes, Id, Record}; - use tracing::{Event, Level, Metadata, Subscriber}; - - /// Count `group::Producer` unfinished-drop WARN events while running `f`. - /// Uses only the existing `tracing` dependency (no tracing-subscriber). - fn count_drop_warnings(f: impl FnOnce()) -> usize { - struct Count(std::sync::Arc); - struct Msg(bool); - impl Visit for Msg { - fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { - if field.name() == "message" { - let s = format!("{value:?}"); - if s.contains("group::Producer dropped without finish") { - self.0 = true; - } - } - } - fn record_str(&mut self, field: &Field, value: &str) { - if field.name() == "message" && value.contains("group::Producer dropped without finish") { - self.0 = true; - } - } - } - impl Subscriber for Count { - fn enabled(&self, metadata: &Metadata<'_>) -> bool { - *metadata.level() <= Level::WARN - } - fn new_span(&self, _span: &Attributes<'_>) -> Id { - Id::from_u64(1) - } - fn record(&self, _span: &Id, _values: &Record<'_>) {} - fn record_follows_from(&self, _span: &Id, _follows: &Id) {} - fn event(&self, event: &Event<'_>) { - let mut msg = Msg(false); - event.record(&mut msg); - if msg.0 { - self.0.fetch_add(1, AtomicOrdering::SeqCst); - } - } - fn enter(&self, _span: &Id) {} - fn exit(&self, _span: &Id) {} - } - - let hits = std::sync::Arc::new(AtomicUsize::new(0)); - tracing::subscriber::with_default(Count(hits.clone()), f); - hits.load(AtomicOrdering::SeqCst) - } #[test] fn basic_frame_reading() { @@ -1021,7 +973,7 @@ mod test { #[test] fn drop_after_abort_does_not_warn() { - let warns = count_drop_warnings(|| { + let warns = count_drop_warnings("group::Producer dropped without finish", || { let producer = Info { sequence: 0 }.produce(); let keep = producer.clone(); let mut writer = producer.clone(); @@ -1037,7 +989,7 @@ mod test { #[test] fn drop_unfinished_warns() { - let warns = count_drop_warnings(|| { + let warns = count_drop_warnings("group::Producer dropped without finish", || { let producer = Info { sequence: 0 }.produce(); let mut writer = producer.clone(); writer diff --git a/rs/moq-net/src/model/mod.rs b/rs/moq-net/src/model/mod.rs index 66a8131cb2..318a3bc7c7 100644 --- a/rs/moq-net/src/model/mod.rs +++ b/rs/moq-net/src/model/mod.rs @@ -19,6 +19,9 @@ mod subscription; mod time; mod weak_cache; +#[cfg(test)] +pub(crate) mod test_tracing; + pub(crate) use requests::Requests; pub(crate) use weak_cache::{WeakCache, WeakEntry}; diff --git a/rs/moq-net/src/model/test_tracing.rs b/rs/moq-net/src/model/test_tracing.rs new file mode 100644 index 0000000000..0aaf18c438 --- /dev/null +++ b/rs/moq-net/src/model/test_tracing.rs @@ -0,0 +1,80 @@ +//! Tracing helpers shared by model tests. + +use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; +use std::sync::Arc; + +use tracing::field::{Field, Visit}; +use tracing::span::{Attributes, Id, Record}; +use tracing::{Event, Level, Metadata, Subscriber}; + +/// Count WARN events whose `message` field contains `expected_message` while running `f`. +/// +/// Uses only the existing `tracing` dependency (no tracing-subscriber). +pub(crate) fn count_drop_warnings(expected_message: &str, f: impl FnOnce()) -> usize { + struct Count { + hits: Arc, + expected: String, + } + + struct Msg<'a> { + expected: &'a str, + matched: bool, + } + + impl Visit for Msg<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + let s = format!("{value:?}"); + if s.contains(self.expected) { + self.matched = true; + } + } + } + + fn record_str(&mut self, field: &Field, value: &str) { + if field.name() == "message" && value.contains(self.expected) { + self.matched = true; + } + } + } + + impl Subscriber for Count { + fn enabled(&self, metadata: &Metadata<'_>) -> bool { + *metadata.level() <= Level::WARN + } + + fn new_span(&self, _span: &Attributes<'_>) -> Id { + Id::from_u64(1) + } + + fn record(&self, _span: &Id, _values: &Record<'_>) {} + + fn record_follows_from(&self, _span: &Id, _follows: &Id) {} + + fn event(&self, event: &Event<'_>) { + let mut msg = Msg { + expected: &self.expected, + matched: false, + }; + event.record(&mut msg); + if msg.matched { + self.hits.fetch_add(1, AtomicOrdering::SeqCst); + } + } + + fn enter(&self, _span: &Id) {} + + fn exit(&self, _span: &Id) {} + } + + let hits = Arc::new(AtomicUsize::new(0)); + let expected = expected_message.to_owned(); + tracing::subscriber::with_default( + Count { + hits: hits.clone(), + expected, + }, + f, + ); + hits.load(AtomicOrdering::SeqCst) +} diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index f2cde50c3e..c0af8bfc2e 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1468,28 +1468,28 @@ impl Drop for Alive { // Check Ok and Err: Ok is unreachable after a deliberate close. match self.state.write() { Ok(mut state) => { - if state.final_sequence.is_some() || state.abort.is_some() { - return; - } - tracing::warn!( - track = %self.name, - "track::Producer dropped without finish() or abort()" - ); - state.clear_cache(); - state.datagrams.clear(); + if state.final_sequence.is_some() || state.abort.is_some() { + return; + } + tracing::warn!( + track = %self.name, + "track::Producer dropped without finish() or abort()" + ); + state.clear_cache(); + state.datagrams.clear(); } Err(state) => { if state.final_sequence.is_some() || state.abort.is_some() { return; } - tracing::warn!( - track = %self.name, - "track::Producer dropped without finish() or abort()" - ); + tracing::warn!( + track = %self.name, + "track::Producer dropped without finish() or abort()" + ); + } } } } -} /// Aggregate every live subscriber's preferences into the most demanding request. /// @@ -2720,6 +2720,7 @@ impl Subscriber { #[cfg(test)] mod test { use super::*; + use crate::model::test_tracing::count_drop_warnings; /// Mint a track for tests with a default parent broadcast, since tracks are /// normally born from a [`broadcast::Producer`]. @@ -2751,56 +2752,6 @@ mod test { .expect("track was closed") } - /// Count `track::Producer` unfinished-drop WARN events while running `f`. - /// Uses only the existing `tracing` dependency (no tracing-subscriber). - fn count_drop_warnings(f: impl FnOnce()) -> usize { - use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; - use tracing::field::{Field, Visit}; - use tracing::span::{Attributes, Id, Record}; - use tracing::{Event, Level, Metadata, Subscriber}; - - struct Count(std::sync::Arc); - struct Msg(bool); - impl Visit for Msg { - fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { - if field.name() == "message" { - let s = format!("{value:?}"); - if s.contains("track::Producer dropped without finish") { - self.0 = true; - } - } - } - fn record_str(&mut self, field: &Field, value: &str) { - if field.name() == "message" && value.contains("track::Producer dropped without finish") { - self.0 = true; - } - } - } - impl Subscriber for Count { - fn enabled(&self, metadata: &Metadata<'_>) -> bool { - *metadata.level() <= Level::WARN - } - fn new_span(&self, _span: &Attributes<'_>) -> Id { - Id::from_u64(1) - } - fn record(&self, _span: &Id, _values: &Record<'_>) {} - fn record_follows_from(&self, _span: &Id, _follows: &Id) {} - fn event(&self, event: &Event<'_>) { - let mut msg = Msg(false); - event.record(&mut msg); - if msg.0 { - self.0.fetch_add(1, AtomicOrdering::SeqCst); - } - } - fn enter(&self, _span: &Id) {} - fn exit(&self, _span: &Id) {} - } - - let hits = std::sync::Arc::new(AtomicUsize::new(0)); - tracing::subscriber::with_default(Count(hits.clone()), f); - hits.load(AtomicOrdering::SeqCst) - } - #[tokio::test] async fn append_datagram_shares_group_sequence() { let mut producer = track_producer("test", None); @@ -3465,7 +3416,7 @@ mod test { async fn drop_after_abort_does_not_warn() { // abort() closes the channel after recording `abort`. Drop must treat that as // clean via read(); without the abort check this emits a false WARN. - let warns = count_drop_warnings(|| { + let warns = count_drop_warnings("track::Producer dropped without finish", || { let producer = track_producer("test", None); let keep = producer.clone(); let mut writer = producer.clone(); @@ -3480,7 +3431,7 @@ mod test { #[tokio::test] async fn drop_unfinished_warns() { - let warns = count_drop_warnings(|| { + let warns = count_drop_warnings("track::Producer dropped without finish", || { let producer = track_producer("test", None); let mut writer = producer.clone(); writer.append_group().unwrap(); From b7275ee62a9b150937cfd825ab1638d089ba6766 Mon Sep 17 00:00:00 2001 From: Mohamed Koubaa <11414628+koubaa@users.noreply.github.com> Date: Sun, 2 Aug 2026 21:16:44 -0500 Subject: [PATCH 4/5] Fix rustfmt and count only WARN events in test helper. CI failed on import ordering in group.rs and test_tracing.rs. Also match Level::WARN exactly so ERROR events are not counted. Co-authored-by: Cursor --- rs/moq-net/src/model/group.rs | 2 +- rs/moq-net/src/model/test_tracing.rs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 1c1fafb976..0ded68325b 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -812,8 +812,8 @@ impl Fetch { #[cfg(test)] mod test { use super::*; - use bytes::Bytes; use crate::model::test_tracing::count_drop_warnings; + use bytes::Bytes; use futures::FutureExt; #[test] diff --git a/rs/moq-net/src/model/test_tracing.rs b/rs/moq-net/src/model/test_tracing.rs index 0aaf18c438..63caa9d852 100644 --- a/rs/moq-net/src/model/test_tracing.rs +++ b/rs/moq-net/src/model/test_tracing.rs @@ -1,7 +1,7 @@ //! Tracing helpers shared by model tests. -use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; use tracing::field::{Field, Visit}; use tracing::span::{Attributes, Id, Record}; @@ -40,7 +40,7 @@ pub(crate) fn count_drop_warnings(expected_message: &str, f: impl FnOnce()) -> u impl Subscriber for Count { fn enabled(&self, metadata: &Metadata<'_>) -> bool { - *metadata.level() <= Level::WARN + *metadata.level() == Level::WARN } fn new_span(&self, _span: &Attributes<'_>) -> Id { From 9ec14cf3a9fb02f26bb2ca144d34ae288fe22913 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 3 Aug 2026 10:49:24 -0700 Subject: [PATCH 5/5] fix(moq-net): address final review feedback Co-Authored-By: Codex --- rs/CLAUDE.md | 2 +- rs/justfile | 5 ++--- rs/moq-net/src/model/track.rs | 8 ++++---- 3 files changed, 7 insertions(+), 8 deletions(-) diff --git a/rs/CLAUDE.md b/rs/CLAUDE.md index d1f1bdc393..86880d1b11 100644 --- a/rs/CLAUDE.md +++ b/rs/CLAUDE.md @@ -180,7 +180,7 @@ Then `Config::load()?` (initializes tracing), build clients/servers via `.init() Run the matching recipe by hand when you touch this code, and if you can't (no such host), say plainly in the PR that it's uncompiled rather than implying CI covered it. -- **`just rs loom` is a manual gate. Run it by hand whenever you touch kio's refcount/waiter plumbing (`lock.rs`, `producer.rs`, `consumer.rs`, `weak.rs`, `waiter.rs`) or moq-net's model layer (`model/`), and mention the result in the PR.** Nothing else will run it: `--cfg loom` swaps kio's Mutex/atomics for loom's instrumented ones, which rebuilds the whole dependency tree and can't share artifacts with a normal `cargo test`, so it's deliberately outside `check`/`ci`. Budget about a minute of model checking on top of that build. The search is exhaustive on purpose, so don't reach for `preemption_bound` to speed it up; the recipe already buys the speed back with `--release`, which matters here because a model check reruns the body once per interleaving. +- **`just rs loom` model-checks concurrent handoffs in kio and moq-net.** It stays outside `check`/`ci`: `--cfg loom` swaps kio's Mutex/atomics for loom's instrumented ones, which rebuilds the whole dependency tree and can't share artifacts with a normal `cargo test`. Use it when developing or diagnosing concurrent handoffs. Budget about a minute of model checking on top of that build. The search is exhaustive on purpose, so don't reach for `preemption_bound` to speed it up; the recipe already buys the speed back with `--release`, which matters here because a model check reruns the body once per interleaving. Loom permutes every thread interleaving instead of hoping a stress loop hits the bad one. It caught a `ProducerWeak::produce` race that had been live for months, on iteration 4. Reading the results: diff --git a/rs/justfile b/rs/justfile index 05fe6d3b32..cd09f9508b 100644 --- a/rs/justfile +++ b/rs/justfile @@ -127,9 +127,8 @@ doctest *args: # Permutation-test the concurrent handoffs with loom. # # `--cfg loom` swaps kio's Mutex/atomics for loom's instrumented ones, so it -# rebuilds the world and can't share artifacts with a normal `cargo test`. That -# is why it's a manual gate rather than part of `check`/`ci`; see the Testing -# section of rs/CLAUDE.md for when to run it. +# rebuilds the world and can't share artifacts with a normal `cargo test`, so +# it stays separate from `check`/`ci`. # # `--release` because a model check runs the body once per interleaving, so the # optimizer pays for itself many times over: 411s -> 51s across the two suites. diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index c0af8bfc2e..5749821762 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -1464,8 +1464,8 @@ impl Drop for Alive { // frame buffers) forever, the same as an explicit abort. A cleanly // finished track keeps its cache so consumers can still drain it. // - // `abort()`/`finish()` close the channel, so `write()` returns `Err(Ref)`. - // Check Ok and Err: Ok is unreachable after a deliberate close. + // `abort()` closes the channel, so `write()` returns `Err(Ref)`. `finish()` + // leaves it open with `final_sequence` set, so inspect both outcomes. match self.state.write() { Ok(mut state) => { if state.final_sequence.is_some() || state.abort.is_some() { @@ -3414,8 +3414,8 @@ mod test { #[tokio::test] async fn drop_after_abort_does_not_warn() { - // abort() closes the channel after recording `abort`. Drop must treat that as - // clean via read(); without the abort check this emits a false WARN. + // abort() closes the channel after recording `abort`. Drop must treat the + // read-only guard returned by write() as clean or it emits a false WARN. let warns = count_drop_warnings("track::Producer dropped without finish", || { let producer = track_producer("test", None); let keep = producer.clone();