diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index d8471298336..cc9a3fd6047 100644 --- a/js/net/src/integration.test.ts +++ b/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")); diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index a9742db256e..cad86be4475 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/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(), - ) - assertEquals( - Timed(Status(state = "live2"), Timestamp(2_000uL, 1_000_000uL)), - consumer.next()?.decode(), - ) - assertEquals( - Timed(Status(state = "live4"), Timestamp(4_000uL, 1_000_000uL)), - consumer.latest()?.decode(), - ) + // `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(), + ) + assertEquals( + Timed(Status(state = "live2"), Timestamp(2_000uL, 1_000_000uL)), + consumer.next()?.decode(), + ) + assertEquals( + Timed(Status(state = "live4"), Timestamp(4_000uL, 1_000_000uL)), + consumer.latest()?.decode(), + ) + } } } @@ -573,16 +574,17 @@ 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) {} + } } } @@ -590,17 +592,19 @@ class SmokeTest { 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().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().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().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().first(), + ) + } } } diff --git a/quest/m0/test-flakes-2/README.md b/quest/m0/test-flakes-2/README.md index 7a304e10420..e337b4d92d8 100644 --- a/quest/m0/test-flakes-2/README.md +++ b/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 diff --git a/quest/m0/test-flakes-2/kotlin-js-flakes.md b/quest/m0/test-flakes-2/kotlin-js-flakes.md deleted file mode 100644 index 67a43838498..00000000000 --- a/quest/m0/test-flakes-2/kotlin-js-flakes.md +++ /dev/null @@ -1,29 +0,0 @@ -# [S] A Kotlin cancel test and a JS late-publisher test hold up - -## Goal - -Two intermittent failures are fixed at their cause: the Kotlin test -"cancelling a flow collection cancels its native handle" no longer fails -with `unroutable`, and the JS integration test "ietf blind handle picks up a -publisher that arrives late" passes under a heavily loaded `just check`. - -## Plan - -Decided 2026-10-31 (maintainer, the /quest-complete requeue session): -root-cause both in one quest, since each is small and they surfaced -together in that session's merge runs. - -- **Kotlin.** `kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt:579`. - An `unroutable` answer means no route served the path when the request - resolved. A request made before the publisher's route reached the origin - returns it, and so does one made after that route was withdrawn or the - broadcast closed. Check both a wait on announcement that the test skips - and a route withdrawal or close during the collection it cancels. -- **JS.** `js/net/src/integration.test.ts:1994`, the moq-transport variant - of the lite test at `:1929`. Under load the late publisher's arrival and - the blind handle's wait race some deadline; find which, and replace the - wall-time dependency with an event or a controlled clock. - -Follow the epic's rules: no raised timeout and no retry. - -Public API: none. Wire: none.