feat(net): add announcedBroadcast, and resolve lite restarts by publisher identity

[kixelated]

Summary

Consuming a broadcast that nobody publishes gets the subscription reset, so a consumer that starts before the publisher stays silent forever unless it retries. js/net had no primitive for "wait for this exact path", only the raw announced(prefix) stream, so every caller that cared drained that stream by hand. There are four such drains now: two in js/watch, one inside Connection.Reload.announced, and one in pipecat-ai/pipecat-client-web-transports#167, which hit this against a bot that starts in response to the client's own announcement and therefore always arrives second.

Rust already has the primitive (origin::Consumer::announced_broadcast); this is the JS counterpart, reshaped as a signal so it reports the broadcast going away as well as coming back.

Announce.Broadcast is a reactive handle to one path. active holds a live Broadcast.Consumer while the path is announced and undefined while nobody publishes it. It owns the edge cases callers kept re-deriving: same-name republish, a relay without discovery (consume blind and warn once, a policy that moves down out of js/watch), and reconnect when built from Connection.Reload.

js/watch's #runBroadcast now mirrors the handle instead of running its own announce loop, which is what proves the primitive covers the real consumer.

Root cause: a restart was resolved blindly

Writing the republish behavior down exposed a real bug one layer below, which the second commit fixes.

Per draft-lcurley-moq-lite an ANNOUNCE_UPDATE means two different things, decided by the advertisement's identity (the Epoch, or while nothing mints one, the first hop of the path):

• Same publisher: the same content over a different route. Cached TRACK_INFO stays valid and in-flight subscriptions resume across it.
• Different publisher: a new generation took the path. Cached TRACK_INFO must be discarded and subscriptions do not carry over.

rs/moq-net's restart_announce implements exactly this: it updates the route in place for the first case (consumers observe nothing) and detaches + reattaches for the second, so downstream sees a real end + start. The JS subscriber implemented neither, flattening both into a bare active: true. A consumer then had to choose between ignoring genuine republishes and tearing down working subscriptions on every reroute, with no information to decide.

So JS now applies the first-hop rule too, covering both the lite-06 ANNOUNCE_UPDATE and the lite-05 duplicate-ANNOUNCE spelling of it. That makes the duplicate active: true that announced.test.ts documents meaningful rather than ambiguous, and it is what lets the new handle keep a live subscription across a reroute while still re-consuming on a republish.

A related bug fell out of review: consumed broadcasts are reference-counted and shared per path, so closing one handle does not release the cache entry. A second holder (the exact shape this PR exists for, a watcher plus a direct consume of the same path for another track) kept the departed publisher's entry alive, and the next consume() cloned its already-reset tracks instead of subscribing to the replacement. Both subscribers now evict the path from the consume cache wherever they surface a retraction, so ended is covered as well as a publisher-change restart, and any caller benefits rather than just this handle. Existing handles are left to their holders; the wire resets whatever they still have open.

Also in review: the IETF subscriber caught a rejected SUBSCRIBE_NAMESPACE, warned, and then let the caller's finally close the announcement stream cleanly. A consumer could not tell discovery failure from "nothing is published under this prefix" and would wait forever. It now aborts the stream with the error, matching the lite subscriber.

Public API changes

Additive, so this targets main:

• js/net: new Announce.Broadcast class (path, active, close()) and its BroadcastProps. The constructor takes a props object so a later option stays additive; the connection methods are the usual entry point, and direct construction covers a caller with its own Getter<Established | undefined>.
• js/net: new announcedBroadcast(path) on Connection.Established and Connection.Reload.

Established is an interface, so a hypothetical external implementor would need the new method. Both implementors are in-tree (lite, ietf) and the interface exists to dispatch between them, so nothing consumers compile against breaks.

The restart change alters observable announcement behavior on announced(): a pure reroute no longer emits an event, and a republish now emits active: false before active: true. That is the behavior rs/moq-net and the draft already specify, so this is JS catching up rather than a new contract.

No wire format change, so no draft update; the draft already specifies the rule being implemented.

Known limitations

Eviction is unconditional, so two announcement streams watching one path can retract out of order and drop the replacement's cache entry, costing a duplicate subscription (both carrying correct content). Telling a stale retraction from a live one needs a generation id on the advertisement, which is what moq-lite's Epoch is for; neither the JS nor the Rust subscriber decodes it yet. Until then this errs toward a duplicate subscription rather than risk handing out a dead generation, and evict's doc comment records why.

On a relay without discovery there is nothing to wait for, so the handle consumes blind and active means assumed present rather than known live: a subscribe to a missing broadcast is how a caller finds out. That is not a dead end, because a consumed broadcast is scoped to the path rather than to one publisher, so a subscribe made after a publisher finally appears succeeds on the same handle. Both are covered by tests.

If discovery fails on a live session, the handle goes offline and stays there, because nothing reopens an announcement stream within a connection. Connection.Reload recovers on the next connection. That gap predates this PR and applies to Reload.announced() equally, so reopen-with-backoff is left as follow-up rather than invented inside a primitive; the class documents the behavior.

Out of scope

Unpublishing over IETF draft-14 throws unknown namespace out of the control-stream adapter and tears the session down rather than retracting. That reproduces independently of this PR (js/net/src/ietf/adapter.ts is untouched here), so it is left alone. It does mean the IETF consume-cache eviction has no test: the lite equivalent covers the logic, and an IETF one needs that bug fixed first.

Test plan

• bun test js/ (746 pass, full suite green), just js check, just js fix.
• New integration tests: a late publisher (stays offline instead of resetting, then resolves), unannounce, same-name republish resolving to a different consumer, close() releasing the broadcast, and the no-discovery blind fallback.
• New reload.test.ts case: the handle drops on session death and re-resolves on the reconnect.
• New integration test: a republish with a second holder of the path reads the new publisher's content, not the previous generation's cached tracks. Reads "old" without the eviction fix.
• New lite/subscriber.test.ts cases driving forged announcements into the subscriber's own stream: a same-publisher restart emits nothing, a different-publisher restart emits end + start, on both the lite-06 and lite-05 spellings. Both fail without the fix.
• Browser, via just dev and the watch demo, run before and after the restart change: bbb.hang plays at 1280x720 / 24-32fps with no console errors; starting a second publisher while the page is open makes a tos.hang tile appear live; killing it retracts the tile after the relay's linger, with bbb.hang unaffected.

(written by Opus 5)

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

The change adds reactive announcedBroadcast handles to established, reload, lite, and IETF connections. Handles follow announcements, reconnects, publication changes, discovery failures, and cleanup. Subscriber logic distinguishes same-publisher reroutes from publisher changes and evicts stale cached consumers. Reload-enabled watch subscriptions use the new handle. Tests cover lifecycle changes, reconnection, restart semantics, cache isolation, and discovery-disabled consumption. A wait example demonstrates the API.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Title check	✅ Passed	The title accurately summarizes the main changes: adding announcedBroadcast functionality and fixing restart resolution by publisher identity in the lite subscriber.
Description check	✅ Passed	The description is comprehensive and directly related to the changeset, explaining the motivation, implementation, and implications of each major change.

✨ Finishing Touches

✨ Simplify code

• Create PR with simplified code
• Commit simplified code in branch claude/moq-broadcast-notifications-5febc5

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 94fc9742b5

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 4de3f289aa

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[coderabbitai]

Actionable comments posted: 4

🧹 Nitpick comments (6)

js/net/src/lite/subscriber.ts (1)

209-213: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Align the comment with the actual map key.

advertised is keyed by suffix (lines 272, 284-286, 304), but the comment states that "the path is the key". path is also a distinct local variable in the same scope (line 263) holding the joined value. Say suffix in the comment to avoid confusion.

♻️ Proposed comment tweak

 			// The publisher behind each path we currently advertise, so a restart can tell a
 			// route change (same publisher, subscriptions resume) from a replacement (a new
 			// generation took the path, nothing carries over). At most one advertisement per
-			// path is current, so the path is the key.
+			// path is current, and every announce on this stream shares `prefix`, so the
+			// suffix is the key.
 			const advertised = new Map<Path.Valid, Origin | undefined>();

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/net/src/lite/subscriber.ts` around lines 209 - 213, Update the comment
immediately above the advertised map declaration to state that suffix is the map
key, replacing the reference to path while preserving the explanation of
publisher tracking and advertisement uniqueness.

js/net/src/lite/subscriber.test.ts (2)

4-6: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Reuse the existing transport mock instead of duplicating it.

announceHarness re-declares the same createBidirectionalStream mock that already exists at lines 14-17. Extract one factory and call it from both places.

As per coding guidelines: "avoid duplicated one-off helpers".

Also applies to: 39-70

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/net/src/lite/subscriber.test.ts` around lines 4 - 6, Consolidate the
duplicated createBidirectionalStream mock used by announceHarness and the
existing setup into a single factory helper. Define the factory once near the
shared test setup, then call it from both locations while preserving each test’s
current behavior and mock configuration.

Source: Coding guidelines

---

76-125: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Add coverage for the reflected-restart branch.

Both tests cover the same-publisher and different-publisher restart paths. The other new branch in #runAnnounced is the reflected active/restart handling at lines 267-277 of js/net/src/lite/subscriber.ts, which deletes the advertisement and emits active: false. No test exercises it. Add a case where the hop chain contains the harness origin after an active announcement, and assert that active: false follows.

As per coding guidelines: "add tests for changes where practical".

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/net/src/lite/subscriber.test.ts` around lines 76 - 125, The announcement
tests need coverage for the reflected active/restart branch in `#runAnnounced`.
Add a test that announces an active path whose hop chain includes the harness
origin, then assert that the announced stream emits the same path with active:
false after the reflected handling. Reuse the existing announceHarness setup and
cleanup patterns.

Source: Coding guidelines

js/net/src/ietf/subscriber.ts (1)

205-210: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low value

Fix the producer-close comment.

Line 210 closes the announcement producer with the subscription error, not the stream: the stream is aborted earlier in the inner catch. Since Producer.close() guards with state.closed.peek() !== undefined, the later clean close cannot overwrite this error.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/net/src/ietf/subscriber.ts` around lines 205 - 210, Update the comment
above announced.close(e) to accurately state that it closes the announcement
producer with the subscription error after the stream was already aborted, and
that Producer.close() preserves this error when the later clean close occurs.

Source: Coding guidelines

js/net/src/connection/reload.test.ts (1)

121-127: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Name the polling limit.

Line 123 uses 500 as an unnamed retry budget. Define a named constant and rename pred to predicate. This makes the test timeout policy clear.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/net/src/connection/reload.test.ts` around lines 121 - 127, Update
waitUntil by introducing a descriptive named constant for the 500-iteration
polling limit and use it in the loop condition; rename the pred parameter to
predicate and update its invocation while preserving the existing timeout
behavior.

Source: Coding guidelines

js/net/examples/wait.ts (1)

1-3: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Add the entrypoint module block.

This file executes main() at Line 41. Add a /** ... @module */ block before Line 1.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/net/examples/wait.ts` around lines 1 - 3, Add a JSDoc module declaration
block at the beginning of the file, before the imports, documenting the
entrypoint module for the `main()` execution flow. Keep the existing `Moq` and
`Effect` imports unchanged.

Source: Coding guidelines

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@js/net/examples/wait.ts`:
- Around line 35-38: Wrap the await of connection.closed in a try/finally block
so effect.close(), broadcast.close(), and connection.close() execute on every
exit path, including when the reload wait rejects. Keep the normal interruption
behavior while ensuring all three resources are released in the finally block.

In `@js/net/src/announced.ts`:
- Around line 213-241: Guard the trailing offline() in the spawned task
associated with the announcement subscription so a stale run cannot clear a
newer run’s active consumer. Use the current effect generation/state via
effect.set() or another run-validity check before calling offline(), while
preserving cleanup for the current run after the stream ends or discovery fails.

In `@js/net/src/ietf/connection.ts`:
- Around line 177-185: Update announcedBroadcast in
js/net/src/ietf/connection.ts (lines 177-185) and the corresponding direct
handle in js/net/src/lite/connection.ts (lines 180-182) so blind consumption
observes the consumer’s closed state in js/net/src/announced.ts and clears
active only when it still references that consumer. Preserve the direct IETF and
Lite handle contracts after the blind consumer closes.

In `@js/net/src/integration.test.ts`:
- Around line 894-946: Ensure both integration tests wrap their bodies in
try/finally blocks so cleanup runs when assertions or awaits fail. In
js/net/src/integration.test.ts lines 894-946, move cleanup for watched, both
producers, both serving tasks, client, and server into finally; in lines
948-976, likewise clean up watched, the producer, serving task, client, and
server in finally, preserving the existing test assertions and flow.

---

Nitpick comments:
In `@js/net/examples/wait.ts`:
- Around line 1-3: Add a JSDoc module declaration block at the beginning of the
file, before the imports, documenting the entrypoint module for the `main()`
execution flow. Keep the existing `Moq` and `Effect` imports unchanged.

In `@js/net/src/connection/reload.test.ts`:
- Around line 121-127: Update waitUntil by introducing a descriptive named
constant for the 500-iteration polling limit and use it in the loop condition;
rename the pred parameter to predicate and update its invocation while
preserving the existing timeout behavior.

In `@js/net/src/ietf/subscriber.ts`:
- Around line 205-210: Update the comment above announced.close(e) to accurately
state that it closes the announcement producer with the subscription error after
the stream was already aborted, and that Producer.close() preserves this error
when the later clean close occurs.

In `@js/net/src/lite/subscriber.test.ts`:
- Around line 4-6: Consolidate the duplicated createBidirectionalStream mock
used by announceHarness and the existing setup into a single factory helper.
Define the factory once near the shared test setup, then call it from both
locations while preserving each test’s current behavior and mock configuration.
- Around line 76-125: The announcement tests need coverage for the reflected
active/restart branch in `#runAnnounced`. Add a test that announces an active path
whose hop chain includes the harness origin, then assert that the announced
stream emits the same path with active: false after the reflected handling.
Reuse the existing announceHarness setup and cleanup patterns.

In `@js/net/src/lite/subscriber.ts`:
- Around line 209-213: Update the comment immediately above the advertised map
declaration to state that suffix is the map key, replacing the reference to path
while preserving the explanation of publisher tracking and advertisement
uniqueness.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 91c03bcb-aeb7-403e-8f89-0790c2f295e7

📥 Commits

Reviewing files that changed from the base of the PR and between e319c6f and 4de3f28.

📒 Files selected for processing (14)

• js/net/README.md
• js/net/examples/wait.ts
• js/net/src/announced.test.ts
• js/net/src/announced.ts
• js/net/src/connection/established.ts
• js/net/src/connection/reload.test.ts
• js/net/src/connection/reload.ts
• js/net/src/ietf/connection.ts
• js/net/src/ietf/subscriber.ts
• js/net/src/integration.test.ts
• js/net/src/lite/connection.ts
• js/net/src/lite/subscriber.test.ts
• js/net/src/lite/subscriber.ts
• js/watch/src/broadcast.ts

[coderabbitai]

🧹 Nitpick comments (1)

js/net/src/announced.ts (1)

177-182: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Document the public Broadcast type.

Broadcast is a public API type returned by Established.announcedBroadcast(), but it has no class-level API documentation or @public annotation. Add a concise class doc.

Proposed change

+/**
+ * A reactive handle for consuming one broadcast path.
+ *
+ * `@public`
+ */
 export class Broadcast {

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/net/src/announced.ts` around lines 177 - 182, Add concise class-level API
documentation to the public Broadcast class, including the `@public` annotation,
while preserving the existing path and active member documentation.

Source: Coding guidelines

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@js/net/src/announced.ts`:
- Around line 177-182: Add concise class-level API documentation to the public
Broadcast class, including the `@public` annotation, while preserving the existing
path and active member documentation.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 85839f0a-6c15-49c4-b983-d96b21bd4bbe

📥 Commits

Reviewing files that changed from the base of the PR and between 4de3f28 and c3a3739.

📒 Files selected for processing (5)

• js/net/examples/wait.ts
• js/net/src/announced.ts
• js/net/src/connection/reload.ts
• js/net/src/ietf/connection.ts
• js/net/src/lite/connection.ts

🚧 Files skipped from review as they are similar to previous changes (4)

• js/net/src/connection/reload.ts
• js/net/src/ietf/connection.ts
• js/net/src/lite/connection.ts
• js/net/examples/wait.ts

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: c3a373914a

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: b7ef2c28c7

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 4f5128f92a

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 64b137b571

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".