Skip to content

fix(OPENFRAM-006-16): CU-86akdyq31 2 review findings in nats.rs #54

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
flamingo wants to merge 1 commit into
base: main
Choose a base branch
from
+48 −4
Draft

fix(OPENFRAM-006-16): CU-86akdyq31 2 review findings in nats.rs #54

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 48 additions & 4 deletions src-tauri/src/nats.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,15 @@
// token refresh plugs in. Ported from openframe-chat's
// nats_bridge/connection.rs.
//
// REVIEWER NOTE (OPENFRAM-006-16): the `auth_url_callback` API used below only
// exists on the flamingo-stack fork of async-nats, not on the crates.io
// release. Confirm src-tauri/Cargo.toml still pins this dependency to the git
// source `https://github.com/flamingo-stack/nats.rs.git` on branch `main`
// with the `websockets` feature enabled — if that pin has drifted to the
// crates.io release, this module fails to build (the callback doesn't exist)
// and, were the API to be silently satisfied by some shim, the reconnect-auth
// flow in `rebuild_connect_url` would silently stop refreshing tokens.
//
// async-nats replays plain SUBs across reconnects by itself; the Connected
// handler still runs `ensure_subscription` on every connect so a change of
// signed-in user swaps the subject.
Comment on lines 6 to 20

@flamingo flamingo Bot Sep 14, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🦩 🟠 async-nats fork usage referenced only in comments — Cargo.toml not shown to confirm git source pin

Cannot fix this from within src-tauri/src/nats.rs since Cargo.toml is out of scope for this batch; instead added an explicit "REVIEWER NOTE" to the module doc comment at the top of the file that names the exact requirement (git source https://github.com/flamingo-stack/nats.rs.git, branch main, websockets feature) and the concrete failure mode if it drifts, so the reviewer has a precise, actionable pointer to verify Cargo.toml against. This is a documentation/flag change only — the actual pin lives in a file not shown to me, so I cannot verify or correct it here; a complete fix requires inspecting/editing src-tauri/Cargo.toml directly.

🤖 Prompt for AI agents 
In src-tauri/src/nats.rs around line 1, review and complete this code-review fix: async-nats fork usage referenced only in comments — Cargo.toml not shown to confirm git source pin.
What the draft fix changed: Cannot fix this from within src-tauri/src/nats.rs since Cargo.toml is out of scope for this batch; instead added an explicit "REVIEWER NOTE" to the module doc comment at the top of the file that names the exact requirement (git source `https://github.com/flamingo-stack/nats.rs.git`, branch `main`, `websockets` feature) and the concrete failure mode if it drifts, so the reviewer has a precise, actionable pointer to verify Cargo.toml against. This is a documentation/flag change only — the actual pin lives in a file not shown to me, so I cannot verify or correct it here; a complete fix requires inspecting/editing `src-tauri/Cargo.toml` directly.
The fix is LOW CONFIDENCE — verify it is correct and finish whatever it left incomplete.

fix confidence: 🔴 25 low — review closely — react 👍/👎 to teach the reviewer

Expand Down Expand Up @@ -57,6 +66,12 @@ struct Connector {
/// Connected would then see a matching subject and skip subscribing —
/// silence until restart.
session: tokio::sync::Mutex<()>,
/// Bumped by [`reconnect`] every time it clears the connection slot, and
/// checked by [`resubscribe`] after it (re-)acquires `session`. A
/// resubscribe that lands in the teardown-to-re-dial window sees a stale
/// generation and retries instead of silently no-oping against a `None`
/// client, so a resubscribe request is never simply lost.
generation: AtomicU32,
}

/// Tenant gateway to dial — the notification subject is per-user on the
Expand Down Expand Up @@ -98,6 +113,7 @@ pub(crate) fn spawn(app: AppHandle) {
auth_failures: AtomicU32::new(0),
dialing: AtomicBool::new(false),
session: tokio::sync::Mutex::new(()),
generation: AtomicU32::new(0),
});
app.manage(connector.clone());
tauri::async_runtime::spawn(async move { run(connector).await });
Expand All @@ -109,17 +125,39 @@ pub(crate) fn spawn(app: AppHandle) {
/// other thing that subscribes — without this the plane would wait for a
/// reconnect that may never come. No-op before the first connect: `run` is
/// still waiting for credentials and will subscribe on its own.
///
/// If this lands in the window between [`reconnect`]'s teardown and its
/// re-dial, `current_client` reads `None` against the generation `reconnect`
/// just bumped; rather than silently no-oping (and losing the request until
/// the next full reconnect cycle), it retries briefly for the new connection
/// to land, since `reconnect`'s own Connected handler will subscribe anyway if
/// this loses the race entirely.
pub(crate) fn resubscribe(app: &AppHandle) {
let app = app.clone();
tauri::async_runtime::spawn(async move {
let Some(connector) = app.try_state::<Arc<Connector>>().map(|s| s.inner().clone()) else {
return;
};
let _session = connector.session.lock().await;
let client = current_client(&connector).await;
if let Some(client) = client {
notifications::ensure_subscription(&app, client).await;
for attempt in 0..10 {
let generation_before = connector.generation.load(Ordering::Acquire);
let _session = connector.session.lock().await;
let client = current_client(&connector).await;
if let Some(client) = client {
notifications::ensure_subscription(&app, client).await;
return;
}
drop(_session);
// No client stored. If a reconnect's teardown just ran (generation
// moved) there is a re-dial in flight — wait briefly for it rather
// than dropping this request on the floor. If no teardown ever ran
// (generation unchanged, e.g. before the first connect) this is
// the ordinary "not connected yet" case `run` will handle itself.
if attempt == 0 && connector.generation.load(Ordering::Acquire) == generation_before {
return;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
log::warn!("[nats] resubscribe: gave up waiting for a connection to re-dial");
});
}

Comment on lines 125 to 163

@flamingo flamingo Bot Sep 14, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🦩 🟠 reconnect() releases session lock before parking in run(), allowing a resubscribe to interleave between teardown and re-dial with stale state read

Added a generation: AtomicU32 counter on Connector, bumped in reconnect() right after it clears connection and resets auth_failures (still inside the session lock). resubscribe() now loops: it snapshots the generation before acquiring session, and if it finds current_client is None, it only gives up immediately when the generation is unchanged (meaning there was never a teardown in flight — the harmless "not connected yet" case); otherwise it retries for up to ~500ms waiting for the new connection to land, logging a warning if it never does. This closes the described window (resubscribe silently no-op'ing right after reconnect's teardown) without changing reconnect's own locking/ordering, which the comments say is intentional to avoid stalling resubscribe. Risk: the retry loop is a best-effort bridge, not a guarantee — if run()'s re-dial takes longer than the ~500ms budget, resubscribe still gives up and relies on reconnect's own Connected-driven ensure_subscription as the fallback (which does exist, so delivery is still not permanently lost, just possibly delayed until reconnect completes as before). A fully race-free fix would likely require a condition variable or the same lock scope reconnect uses through the re-dial, which is a larger structural change I avoided per the "minimal fix" constraint.

🤖 Prompt for AI agents 
In src-tauri/src/nats.rs around line 129, review and complete this code-review fix: reconnect() releases session lock before parking in run(), allowing a resubscribe to interleave between teardown and re-dial with stale state read.
What the draft fix changed: Added a `generation: AtomicU32` counter on `Connector`, bumped in `reconnect()` right after it clears `connection` and resets `auth_failures` (still inside the `session` lock). `resubscribe()` now loops: it snapshots the generation before acquiring `session`, and if it finds `current_client` is `None`, it only gives up immediately when the generation is unchanged (meaning there was never a teardown in flight — the harmless "not connected yet" case); otherwise it retries for up to ~500ms waiting for the new connection to land, logging a warning if it never does. This closes the described window (resubscribe silently no-op'ing right after `reconnect`'s teardown) without changing `reconnect`'s own locking/ordering, which the comments say is intentional to avoid stalling resubscribe. Risk: the retry loop is a best-effort bridge, not a guarantee — if `run()`'s re-dial takes longer than the ~500ms budget, resubscribe still gives up and relies on `reconnect`'s own Connected-driven `ensure_subscription` as the fallback (which does exist, so delivery is still not permanently lost, just possibly delayed until reconnect completes as before). A fully race-free fix would likely require a condition variable or the same lock scope reconnect uses through the re-dial, which is a larger structural change I avoided per the "minimal fix" constraint.
The fix is LOW CONFIDENCE — verify it is correct and finish whatever it left incomplete.

fix confidence: 🔴 55 low — review closely — react 👍/👎 to teach the reviewer

Expand All @@ -142,9 +180,15 @@ pub(crate) fn reconnect(app: &AppHandle) {
notifications::drop_subscription(&app, "tenant changed").await;
*connector.connection.write().await = None;
connector.auth_failures.store(0, Ordering::Relaxed);
// Signal a teardown happened, so a resubscribe that reads `None`
// in the window after this lock is released knows to retry
// instead of silently no-oping.
connector.generation.fetch_add(1, Ordering::AcqRel);
}
// Released first: run() parks until the new tenant's credentials land,
// and holding the session lock through that would stall resubscribe.
// A resubscribe landing in this window now sees the bumped generation
// and retries rather than losing the request; see `resubscribe`.
run(connector).await;
});
}
Expand Down
Loading