libmoq: auto-reconnect sessions; conducer-based Reconnect notifications

[kixelated]

Summary

moq_session_connect established a single session that died permanently when the connection dropped. It now drives moq_native's Reconnect helper (the same one used by moq-cli, moq-boy, and the hang/moq-native examples) so libmoq sessions reconnect with exponential backoff. Origins outlive the connection, so published broadcasts are re-announced and consumers re-subscribed automatically on each reconnect.

Reconnect rebuilt on kio

Reconnect stores its state (current status + terminal error) in a kio channel instead of tokio::sync::watch, matching the rest of moq-net. It exposes:

• status(&mut self) -> Result<Status> / poll_status — returns the current Status (Connected/Disconnected) whenever it differs from what this handle last reported. Level-triggered: it tracks the current state, so a flip-and-flip-back between polls collapses to a single report. Err is the terminal signal (give-up error, or generic once the handle is dropped).
• closed() -> Result<()> / poll_closed — for callers that only want to await termination (unchanged for existing consumers).

kio is now re-exported from moq-net (pub use kio). Its public poll_* API already exposes kio::Waiter, so downstream callers need to name the type; moq-native uses moq_net::kio rather than a direct dep. (Making Reconnect cloneable for multiple handles is deferred to a follow-up PR.)

libmoq callback contract

moq_session_connect's on_status reports connection state via the sign of the code:

• positive — connected; the value is the connection epoch (1 = first connect, 2+ = reconnect).
• negative — reconnection permanently gave up (backoff timeout exceeded). Terminal.

Transient disconnects are not reported (a separate change reserves 0 for "closed"). The status loop lives in report(), which returns anyhow::Result<()> so ? flows; connect_run maps the terminal error to Connect once. The callback is copied out before the global lock is released, so the C callback never runs while the lock is held. The function signature is unchanged (no wire/ABI break), so this targets main.

Follow-up (not in this PR)

rs/moq-ffi/src/session.rs has the identical single-shot .connect() gap for the Python/Swift/Kotlin/Go bindings. The same .connect() -> .reconnect() swap applies there.

Test plan

• cargo clippy --all-targets --workspace -- -D warnings clean
• cargo test -p libmoq -p moq-native -p moq-net pass
• RUSTDOCFLAGS="-D warnings" cargo doc --no-deps --workspace clean
• moq-cli/moq-boy/hang/moq-native consumers + examples still build
• Manual: drop/restore a relay mid-session and confirm the callback reports a reconnect (epoch++) and the publisher re-announces / subscriber resumes

🤖 Generated with Claude Code

(Written by Claude)

[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

This PR updates moq_session_connect docs for automatic exponential-backoff reconnect and re-announcement/resubscription semantics; refactors Session::connect_run to drive a reconnect(url) loop that reports positive epoch-based Connected codes and surfaces terminal failures as negative codes; moves reconnect lifecycle signaling to a conducer-shared State with a new public Status and Reconnect::status polling API; ensures session on_status callbacks are invoked outside the global lock; and re-exports conducer from moq-net.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title accurately summarizes the two main changes: auto-reconnect sessions in libmoq and the Reconnect refactor to use conducer-based notifications.
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.
Description check	✅ Passed	The PR description clearly explains the changes: auto-reconnect sessions in libmoq using moq-native's Reconnect helper, rebuilt on conducer channels with a new level-triggered API, and updated callback contract with signed status codes.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

✨ Simplify code

• Create PR with simplified code
• Commit simplified code in branch claude/compassionate-wozniak-88b7a7

---

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 and usage tips.}

[coderabbitai]

Actionable comments posted: 2

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

rs/libmoq/src/session.rs (1)

43-45: ⚠️ Potential issue | 🔴 Critical | ⚡ Quick win

Snapshot the callback so State::lock() is dropped before invoking user code

State::lock() in the if let scrutinee stays alive through the if body, so entry.callback.call(res) can run while holding the global lock (contrary to the “lock is dropped before the callback is invoked” comment), risking deadlock if the callback re-enters libmoq.

Suggested fix

-			if let Some(entry) = State::lock().session.task.remove(id).flatten() {
-				entry.callback.call(res);
-			}
+			let callback = State::lock()
+				.session
+				.task
+				.remove(id)
+				.flatten()
+				.map(|entry| entry.callback);
+
+			if let Some(callback) = callback {
+				callback.call(res);
+			}

🤖 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 `@rs/libmoq/src/session.rs` around lines 43 - 45, The code currently holds the
global lock across the `if` body so `entry.callback.call(res)` can run while the
lock is still held; change the logic to snapshot (move) the callback out while
the lock is held and then drop the lock before invoking user code: use
`State::lock().session.task.remove(id).flatten()` only to extract the entry,
immediately move `entry.callback` into a local variable (e.g., `callback`) while
still under the lock, let the lock scope end, then call `callback.call(res)`
outside that lock scope so `State::lock()` is not held during the callback
invocation.

🤖 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 `@rs/libmoq/src/session.rs`:
- Around line 73-87: The current tokio::select! on reconnect.connected(connects)
and reconnect.disconnected(disconnects) can yield disconnected(0) before the
matching connected epoch because the two counters are independent; replace the
dual-branch await with a single ordered event stream so events are serialized.
Concretely, add or use a single producer method on the reconnect state (e.g.
reconnect.next_event() or reconnect.events() returning an async Stream/async fn
that yields an enum like ReconnectEvent::Connected(epoch) |
ReconnectEvent::Disconnected(epoch)), then await that single source inside the
loop and call Self::notify(task_id, epoch_or_0) based on the event; this ensures
ordering (use State.connects / State.disconnects internally in reconnect to
produce ordered events) and removes the separate reconnect.connected(...) and
reconnect.disconnected(...) branches from the tokio::select!.

In `@rs/moq-native/src/reconnect.rs`:
- Around line 153-161: poll_connected (and similarly poll_disconnected)
currently maps Poll::Ready(Err(_)) to Pending, which drops the last recorded
epoch if the reconnect task terminated; instead, on Poll::Ready(Err(_)) read the
last stored epoch from the shared state and return Poll::Ready(epoch) when that
epoch > since (otherwise return Pending). Concretely, inside
poll_connected/poll_disconnected change the Poll::Ready(Err(_)) arm to fetch the
final state (via the same shared state accessor used in the closure—e.g.,
inspect self.state's stored connects/disconnects) and compare to since,
returning Poll::Ready(last_epoch) when newer, otherwise Poll::Pending; do the
same mirrored change for poll_disconnected.

---

Outside diff comments:
In `@rs/libmoq/src/session.rs`:
- Around line 43-45: The code currently holds the global lock across the `if`
body so `entry.callback.call(res)` can run while the lock is still held; change
the logic to snapshot (move) the callback out while the lock is held and then
drop the lock before invoking user code: use
`State::lock().session.task.remove(id).flatten()` only to extract the entry,
immediately move `entry.callback` into a local variable (e.g., `callback`) while
still under the lock, let the lock scope end, then call `callback.call(res)`
outside that lock scope so `State::lock()` is not held during the callback
invocation.

🪄 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

Run ID: f893df7f-8952-4df3-af88-8f234bd67001

📥 Commits

Reviewing files that changed from the base of the PR and between 8b8f093 and d735876.

📒 Files selected for processing (4)

• rs/libmoq/src/api.rs
• rs/libmoq/src/session.rs
• rs/moq-native/src/reconnect.rs
• rs/moq-net/src/lib.rs

✅ Files skipped from review due to trivial changes (1)

• rs/libmoq/src/api.rs

[coderabbitai]

Actionable comments posted: 1

🤖 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 `@rs/moq-native/src/reconnect.rs`:
- Around line 178-196: The code returns the terminal error as soon as
consumer.poll returns Err(_) and thus may drop pending transitions; modify the
polling closure used in conducer::wait (the closure around consumer.poll /
state.epoch / since) so that on Poll::Ready(Err(_)) you still inspect the latest
state. If state.epoch > since treat it as a successful advance (return
Poll::Ready(true) / equivalent) so pending transitions are surfaced, otherwise
return the terminal error as before (the code path that currently calls
self.outcome().err().unwrap_or_else(...)). Update the logic in reconnect.rs
around the consumer.poll match and the advanced handling so channel-close errors
only short-circuit when no new epoch > since exists.

🪄 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

Run ID: da62c84e-29a7-4372-9ad6-9e0018b37389

📥 Commits

Reviewing files that changed from the base of the PR and between dd7137d and d083c8c.

📒 Files selected for processing (2)

• rs/libmoq/src/session.rs
• rs/moq-native/src/reconnect.rs

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

• rs/libmoq/src/session.rs