Don't warn on Alive::drop if aborted

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:
 

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.

rs/moq-net/src/model/group.rs

@@ -244,16 +244,28 @@ impl Drop for Alive {
 		// 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().
-			tracing::warn!(
-				sequence = self.info.sequence,
-				"group::Producer dropped without finish() or abort()"
-			);
-			state.release();
+		//
+		// 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();
+			}
+			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()"
+				);
+			}
 		}
 	}
 }
@@ -800,6 +812,7 @@ impl Fetch {
 #[cfg(test)]
 mod test {
 	use super::*;
+	use crate::model::test_tracing::count_drop_warnings;
 	use bytes::Bytes;
 	use futures::FutureExt;
 
@@ -958,6 +971,37 @@ mod test {
 		assert!(matches!(result, Err(crate::Error::Dropped)));
 	}
 
+	#[test]
+	fn drop_after_abort_does_not_warn() {
+		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();
+			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("group::Producer dropped without finish", || {
+			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();

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};
 

rs/moq-net/src/model/test_tracing.rs

@@ -0,0 +1,80 @@
+//! Tracing helpers shared by model tests.
+
+use std::sync::Arc;
+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 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<AtomicUsize>,
+		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)
+}

rs/moq-net/src/model/track.rs

@@ -1463,18 +1463,30 @@ impl Drop for Alive {
 		// 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().
-			tracing::warn!(
-				track = %self.name,
-				"track::Producer dropped without finish() or abort()"
-			);
-			state.clear_cache();
-			state.datagrams.clear();
+		//
+		// `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() {
+					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()"
+				);
+			}
 		}
 	}
 }
@@ -2708,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`].
@@ -3399,6 +3412,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 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();
+			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("track::Producer dropped without finish", || {
+			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);