Change broadcast replacement strategy to queue backups instead of reannouncing (#1319)

Co-authored-by: Claude <noreply@anthropic.com>

Author: kixelated

rs/moq-lite/src/model/origin.rs

@@ -1,5 +1,5 @@
 use std::{
-	collections::HashMap,
+	collections::{HashMap, VecDeque},
 	sync::atomic::{AtomicU64, Ordering},
 };
 use tokio::sync::mpsc;
@@ -19,11 +19,11 @@ impl ConsumerId {
 	}
 }
 
-// If there are multiple broadcasts with the same path, we use the most recent one but keep the others around.
+// If there are multiple broadcasts with the same path, we keep the oldest active and queue the others.
 struct OriginBroadcast {
 	path: PathOwned,
 	active: BroadcastConsumer,
-	backup: Vec<BroadcastConsumer>,
+	backup: VecDeque<BroadcastConsumer>,
 }
 
 #[derive(Clone)]
@@ -145,17 +145,15 @@ impl OriginNode {
 			self.entry(dir).lock().publish(&full, broadcast, &relative);
 		} else if let Some(existing) = &mut self.broadcast {
 			// This node is a leaf with an existing broadcast.
-			let old = existing.active.clone();
-			existing.active = broadcast.clone();
-			existing.backup.push(old);
-
-			self.notify.lock().reannounce(full, broadcast);
+			// Keep the older broadcast active; queue the new one as a backup.
+			// This avoids reannouncing and potentially disrupting subscribers.
+			existing.backup.push_back(broadcast.clone());
 		} else {
 			// This node is a leaf with no existing broadcast.
 			self.broadcast = Some(OriginBroadcast {
 				path: full.to_owned(),
 				active: broadcast.clone(),
-				backup: Vec::new(),
+				backup: VecDeque::new(),
 			});
 			self.notify.lock().announce(full, broadcast);
 		}
@@ -227,9 +225,9 @@ impl OriginNode {
 			// Okay so it must be the active broadcast or else we fucked up.
 			assert!(entry.active.is_clone(&broadcast));
 
-			// If there's a backup broadcast, then announce it.
-			if let Some(active) = entry.backup.pop() {
-				entry.active = active;
+			// If there's a backup broadcast, promote the oldest one.
+			if let Some(next) = entry.backup.pop_front() {
+				entry.active = next;
 				self.notify.lock().reannounce(full, &entry.active);
 			} else {
 				// No more backups, so remove the entry.
@@ -367,9 +365,10 @@ impl OriginProducer {
 	/// Publish a broadcast, announcing it to all consumers.
 	///
 	/// The broadcast will be unannounced when it is closed.
-	/// If there is already a broadcast with the same path, then it will be replaced and reannounced.
-	/// If the old broadcast is closed before the new one, then nothing will happen.
-	/// If the new broadcast is closed before the old one, then the old broadcast will be reannounced.
+	/// If there is already a broadcast with the same path, then the older broadcast remains active
+	/// and the new one is queued as a backup (no reannounce is triggered).
+	/// When the active broadcast closes, the oldest queued backup is promoted and reannounced.
+	/// A queued backup that closes before it is promoted is silently dropped with no announcement.
 	///
 	/// Returns false if the broadcast is not allowed to be published.
 	pub fn publish_broadcast(&self, path: impl AsPath, broadcast: BroadcastConsumer) -> bool {
@@ -624,6 +623,8 @@ mod tests {
 
 	#[tokio::test]
 	async fn test_announce() {
+		tokio::time::pause();
+
 		let origin = Origin::produce();
 		let broadcast1 = Broadcast::produce();
 		let broadcast2 = Broadcast::produce();
@@ -688,6 +689,8 @@ mod tests {
 
 	#[tokio::test]
 	async fn test_duplicate() {
+		tokio::time::pause();
+
 		let origin = Origin::produce();
 
 		let broadcast1 = Broadcast::produce();
@@ -705,13 +708,11 @@ mod tests {
 		origin.publish_broadcast("test", consumer3.clone());
 		assert!(consumer.consume_broadcast("test").is_some());
 
+		// Only the oldest broadcast is announced; later publishes go to the backup queue.
 		consumer.assert_next("test", &consumer1);
-		consumer.assert_next_none("test");
-		consumer.assert_next("test", &consumer2);
-		consumer.assert_next_none("test");
-		consumer.assert_next("test", &consumer3);
+		consumer.assert_next_wait();
 
-		// Drop the backup, nothing should change.
+		// Drop a backup, nothing should change.
 		drop(broadcast2);
 
 		// Wait for the async task to run.
@@ -720,18 +721,18 @@ mod tests {
 		assert!(consumer.consume_broadcast("test").is_some());
 		consumer.assert_next_wait();
 
-		// Drop the active, we should reannounce.
-		drop(broadcast3);
+		// Drop the active, we should reannounce with the oldest remaining backup.
+		drop(broadcast1);
 
 		// Wait for the async task to run.
 		tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
 
 		assert!(consumer.consume_broadcast("test").is_some());
 		consumer.assert_next_none("test");
-		consumer.assert_next("test", &consumer1);
+		consumer.assert_next("test", &consumer3);
 
 		// Drop the final broadcast, we should unannounce.
-		drop(broadcast1);
+		drop(broadcast3);
 
 		// Wait for the async task to run.
 		tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
@@ -741,8 +742,55 @@ mod tests {
 		consumer.assert_next_wait();
 	}
 
+	#[tokio::test]
+	async fn test_duplicate_fifo_order() {
+		tokio::time::pause();
+
+		let origin = Origin::produce();
+
+		let broadcast1 = Broadcast::produce();
+		let broadcast2 = Broadcast::produce();
+		let broadcast3 = Broadcast::produce();
+
+		let consumer1 = broadcast1.consume();
+		let consumer2 = broadcast2.consume();
+		let consumer3 = broadcast3.consume();
+
+		let mut consumer = origin.consume();
+
+		origin.publish_broadcast("test", consumer1.clone());
+		origin.publish_broadcast("test", consumer2.clone());
+		origin.publish_broadcast("test", consumer3.clone());
+
+		// The oldest broadcast is active; the rest are queued in publish order.
+		consumer.assert_next("test", &consumer1);
+		consumer.assert_next_wait();
+
+		// Drop the active; the next-oldest (not the newest) should be promoted.
+		drop(broadcast1);
+		tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
+		consumer.assert_next_none("test");
+		consumer.assert_next("test", &consumer2);
+		consumer.assert_next_wait();
+
+		// Drop the now-active; the remaining backup is promoted.
+		drop(broadcast2);
+		tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
+		consumer.assert_next_none("test");
+		consumer.assert_next("test", &consumer3);
+		consumer.assert_next_wait();
+
+		// Drop the last broadcast; the entry is fully unannounced.
+		drop(broadcast3);
+		tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
+		consumer.assert_next_none("test");
+		consumer.assert_next_wait();
+	}
+
 	#[tokio::test]
 	async fn test_duplicate_reverse() {
+		tokio::time::pause();
+
 		let origin = Origin::produce();
 		let broadcast1 = Broadcast::produce();
 		let broadcast2 = Broadcast::produce();
@@ -767,6 +815,8 @@ mod tests {
 
 	#[tokio::test]
 	async fn test_double_publish() {
+		tokio::time::pause();
+
 		let origin = Origin::produce();
 		let broadcast = Broadcast::produce();
 
@@ -1322,6 +1372,8 @@ mod tests {
 	// Verify unannounce also doesn't panic with trailing slash
 	#[tokio::test]
 	async fn test_with_root_trailing_slash_unannounce() {
+		tokio::time::pause();
+
 		let origin = Origin::produce();
 
 		let prefix = "some_prefix/".to_string();