test: retain Kotlin snapshot writers and require late-publisher refusals

js/net/src/integration.test.ts

@@ -1943,7 +1943,9 @@ test("integration: a blind handle picks up a publisher that arrives late", async
 
 	// Nobody publishes the path yet, so a subscribe is how the caller finds out. That kills the
 	// track, not the handle: a consumed broadcast is scoped to the path, not to one publisher.
-	await expect(blind.track("video").subscribe().ordered().readString()).rejects.toThrow();
+	await expect(blind.track("video").subscribe().ordered().readString()).rejects.toMatchObject({
+		code: StreamCode.NotFound,
+	});
 	expect(watched.active.peek()).toBe(blind);
 
 	// So a subscribe made after the publisher finally shows up still works, on the same handle.
@@ -2006,7 +2008,7 @@ test("integration: ietf blind handle picks up a publisher that arrives late", as
 	if (!blind) throw new Error("expected a blind consumer");
 
 	// Rejects with 404 rather than resetting the whole handle.
-	await expect(blind.track("video").subscribe().ordered().readString()).rejects.toThrow();
+	await expect(blind.track("video").subscribe().ordered().readString()).rejects.toThrow("does_not_exist");
 	expect(watched.active.peek()).toBe(blind);
 
 	const producer = publish(origin, Path.from("later"));

kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt

@@ -509,26 +509,27 @@ class SmokeTest {
         BroadcastProducer().use { broadcast ->
             val config = SnapshotConfig(deltas = true)
             val track = broadcast.publishTrack("status", TrackInfo(timescale = 1_000_000uL))
-            val producer = SnapshotProducer(broadcast, track, config)
-            val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null))
-            for (n in 1..4) producer.update(
-                Status(state = "live$n"),
-                timestamp = n.milliseconds.toTimestamp(),
-            )
+            SnapshotProducer(broadcast, track, config).use { producer ->
+                val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null))
+                for (n in 1..4) producer.update(
+                    Status(state = "live$n"),
+                    timestamp = n.milliseconds.toTimestamp(),
+                )
 
-            // `next` yields every state in order; `latest` skips the buffered third.
-            assertEquals(
-                Timed(Status(state = "live1"), Timestamp(1_000uL, 1_000_000uL)),
-                consumer.next()?.decode<Status>(),
-            )
-            assertEquals(
-                Timed(Status(state = "live2"), Timestamp(2_000uL, 1_000_000uL)),
-                consumer.next()?.decode<Status>(),
-            )
-            assertEquals(
-                Timed(Status(state = "live4"), Timestamp(4_000uL, 1_000_000uL)),
-                consumer.latest()?.decode<Status>(),
-            )
+                // `next` yields every state in order; `latest` skips the buffered third.
+                assertEquals(
+                    Timed(Status(state = "live1"), Timestamp(1_000uL, 1_000_000uL)),
+                    consumer.next()?.decode<Status>(),
+                )
+                assertEquals(
+                    Timed(Status(state = "live2"), Timestamp(2_000uL, 1_000_000uL)),
+                    consumer.next()?.decode<Status>(),
+                )
+                assertEquals(
+                    Timed(Status(state = "live4"), Timestamp(4_000uL, 1_000_000uL)),
+                    consumer.latest()?.decode<Status>(),
+                )
+            }
         }
     }
 
@@ -573,34 +574,37 @@ class SmokeTest {
     fun `json producer demand follows subscribers`() = runTest {
         BroadcastProducer().use { broadcast ->
             val config = SnapshotConfig(deltas = false)
-            val producer =
-                SnapshotProducer(broadcast, broadcast.publishTrack("status", null), config)
-            val demand: TrackDemand = producer.demand()
-            assertEquals("status", demand.name())
-            assertEquals(null, demand.wanted())
-
-            val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null))
-            while (demand.wantedChanged() == null) {}
-            consumer.cancel()
-            while (demand.wantedChanged() != null) {}
+            SnapshotProducer(broadcast, broadcast.publishTrack("status", null), config).use {
+                producer ->
+                val demand: TrackDemand = producer.demand()
+                assertEquals("status", demand.name())
+                assertEquals(null, demand.wanted())
+
+                val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null))
+                while (demand.wantedChanged() == null) {}
+                consumer.cancel()
+                while (demand.wantedChanged() != null) {}
+            }
         }
     }
 
     @Test
     fun `cancelling a flow collection cancels its native handle`() = runTest {
         BroadcastProducer().use { broadcast ->
             val config = SnapshotConfig(deltas = false)
-            val producer =
-                SnapshotProducer(broadcast, broadcast.publishTrack("status", null), config)
-            val demand: TrackDemand = producer.demand()
-            producer.update(Status(state = "live"))
-
-            val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null))
-            while (demand.wantedChanged() == null) {}
-            // `first` cancels the collection, which must release the subscription without an
-            // explicit cancel.
-            assertEquals(Status(state = "live"), consumer.valuesAs<Status>().first().value)
-            while (demand.wantedChanged() != null) {}
+            // A demand watcher does not own the writer; keep it alive through collection.
+            SnapshotProducer(broadcast, broadcast.publishTrack("status", null), config).use {
+                producer ->
+                val demand: TrackDemand = producer.demand()
+                producer.update(Status(state = "live"))
+
+                val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null))
+                while (demand.wantedChanged() == null) {}
+                // `first` cancels the collection, which must release the subscription without an
+                // explicit cancel.
+                assertEquals(Status(state = "live"), consumer.valuesAs<Status>().first().value)
+                while (demand.wantedChanged() != null) {}
+            }
         }
     }
 
@@ -612,12 +616,16 @@ class SmokeTest {
     fun `raw json string passes through unencoded`() = runTest {
         BroadcastProducer().use { broadcast ->
             val config = SnapshotConfig(deltas = false)
-            val producer =
-                SnapshotProducer(broadcast, broadcast.publishTrack("status", null), config)
-            producer.update("""{"state":"raw"}""")
-
-            val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null))
-            assertEquals(Timed(Status(state = "raw"), null), consumer.valuesAs<Status>().first())
+            SnapshotProducer(broadcast, broadcast.publishTrack("status", null), config).use {
+                producer ->
+                producer.update("""{"state":"raw"}""")
+
+                val consumer = SnapshotConsumer(broadcast.consume().subscribeTrack("status", null))
+                assertEquals(
+                    Timed(Status(state = "raw"), null),
+                    consumer.valuesAs<Status>().first(),
+                )
+            }
         }
     }
 

quest/m0/test-flakes-2/README.md

@@ -56,6 +56,14 @@ moq-relay's auth-trust test was never load-bound either: its fixture left each
 lease's background `end` POST unserved, so a reused port let one steal the next
 trial's accept. Fixed by [moq-dev/moq#7550](https://github.com/moq-dev/moq/pull/7550).
 
+The Kotlin flow-cancel test's `unroutable` was not load-bound: its snapshot
+writer became unreachable, so JVM cleanup aborted the track mid-read. The JS
+IETF blind-handle test's one identified cause was the session-closing reset
+fixed by [moq-dev/moq#6988](https://github.com/moq-dev/moq/pull/6988). A single
+event-loop stall at each of its 124 host tasks passes all 124 at 1.5 seconds;
+at 4.5 seconds 3 runs and at 11 seconds 114 runs hit bun's own 5-second test
+timeout. No run fails on a protocol deadline, so no clock change applies. Both handled in [moq-dev/moq#7622](https://github.com/moq-dev/moq/pull/7622).
+
 Public API: none. Wire: none.
 
 ## Required
@@ -65,4 +73,3 @@ Public API: none. Wire: none.
 - [moq-tokio's close-drain test survives load](/quest/m0/test-flakes-2/tokio-close-drain.md) - `quic_client_close_drains_finished_track` stops failing land batches with "Unroutable"
 - [@moq/net's close-drain tests survive load](/quest/m0/test-flakes-2/js-close-drain.md) - the lite close-drain cases pass on every full suite run
 - [@moq/net's boundary-sync test survives load](/quest/m0/test-flakes-2/js-boundary-sync.md) - "a version without a boundary syncs" passes under load
-- [A Kotlin cancel test and a JS late-publisher test hold up](/quest/m0/test-flakes-2/kotlin-js-flakes.md) - the Kotlin flow-cancel test stops failing `unroutable` and the IETF blind-handle test passes under load