From 255d8d59756a0cf975a581944d7ecfdda6289b5f Mon Sep 17 00:00:00 2001 From: hitalin Date: Fri, 3 Jul 2026 09:44:20 +0900 Subject: [PATCH 1/3] =?UTF-8?q?fix:=20=E5=88=9D=E5=9B=9E=E6=8E=A5=E7=B6=9A?= =?UTF-8?q?=E5=A4=B1=E6=95=97=E3=82=92=E5=86=8D=E6=8E=A5=E7=B6=9A=E3=83=AB?= =?UTF-8?q?=E3=83=BC=E3=83=97=E3=81=AB=E5=A7=94=E3=81=AD=E3=81=A6=E7=84=A1?= =?UTF-8?q?=E9=9F=B3=E6=81=92=E4=B9=85=E5=8C=96=E3=82=92=E9=98=B2=E3=81=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit connect() は初回ハンドシェイク失敗で Err を返すだけで、再接続ループは 一度セッションが確立した connection_task 内にしか存在しなかった。 起動直後のネットワーク未確立などで初回だけ失敗すると、再試行の責務が 宙に浮き「永久にストリームなし」になっていた (notedeck#506 で顕在化 した無音恒久化の構造側)。 初回失敗時も connection_task を spawn し、既存の Equal Jitter バック オフループに最初から任せる。接続の生死は stream-status イベント (connected / reconnecting) でフロントへ伝わる。 --- src/streaming.rs | 93 +++++++++++++++++++++++++++++++++++------------- 1 file changed, 69 insertions(+), 24 deletions(-) diff --git a/src/streaming.rs b/src/streaming.rs index 2f1db7d..f5db242 100644 --- a/src/streaming.rs +++ b/src/streaming.rs @@ -296,14 +296,29 @@ impl StreamingManager { let url = format!("wss://{host}/streaming?i={token}"); - // Verify initial connection (with timeout to prevent hang DoS) - let (ws_stream, _) = tokio::time::timeout( + // Attempt the initial connection (with timeout to prevent hang DoS). + // 失敗しても Err を返さず connection_task の再接続ループに委ねる: + // 初回失敗を呼び出し側へ返すだけだと再試行の責務が宙に浮き、 + // 「起動直後のネットワーク未確立 = 永久にストリームなし」になる。 + // 接続の生死は stream-status イベント (connected / reconnecting) で + // フロントへ伝わる。 + let initial_ws = match tokio::time::timeout( WS_CONNECT_TIMEOUT, tokio_tungstenite::connect_async_with_config(&url, Some(ws_config()), false), ) .await - .map_err(|_| NoteDeckError::WebSocket("Connection timeout".to_string()))? - .map_err(|e| NoteDeckError::WebSocket(e.to_string()))?; + { + Ok(Ok((ws_stream, _))) => Some(ws_stream), + Ok(Err(e)) => { + tracing::warn!(account_id, error = %e, "initial connect failed; retrying in background"); + None + } + Err(_) => { + tracing::warn!(account_id, "initial connect timed out; retrying in background"); + None + } + }; + let connected = initial_ws.is_some(); let (cmd_tx, cmd_rx) = mpsc::unbounded_channel(); @@ -328,7 +343,7 @@ impl StreamingManager { host_owned, token_owned, url_owned, - ws_stream, + initial_ws, cmd_rx, subscriptions, captured_notes, @@ -350,7 +365,7 @@ impl StreamingManager { "stream-status", StreamStatusEvent { account_id: account_id.to_string(), - state: "connected".to_string(), + state: if connected { "connected" } else { "reconnecting" }.to_string(), } ); @@ -917,30 +932,34 @@ async fn connection_task( host: String, token: String, url: String, - initial_ws: WsStream, + initial_ws: Option, mut cmd_rx: mpsc::UnboundedReceiver, subscriptions: Arc>>, captured_notes: Arc>>>, ) { let mut backoff_secs: u64 = 1; - // Run the first session with the already-connected WebSocket - let reason = run_ws_session( - &emitter, - &event_bus, - &db, - &api_client, - &account_id, - &host, - &token, - initial_ws, - &mut cmd_rx, - &subscriptions, - &captured_notes, - ) - .await; - if matches!(reason, WsExitReason::Shutdown) { - return; + // Run the first session with the already-connected WebSocket. + // None = 初回接続に失敗している (connect() が委譲してきた)。 + // その場合は最初から下の再接続ループに任せる。 + if let Some(initial_ws) = initial_ws { + let reason = run_ws_session( + &emitter, + &event_bus, + &db, + &api_client, + &account_id, + &host, + &token, + initial_ws, + &mut cmd_rx, + &subscriptions, + &captured_notes, + ) + .await; + if matches!(reason, WsExitReason::Shutdown) { + return; + } } // Reconnection loop @@ -1800,4 +1819,30 @@ mod tests { // 3x gives margin for two consecutive lost pongs. assert!(WS_READ_IDLE_TIMEOUT >= WS_PING_INTERVAL * 3); } + + #[tokio::test] + async fn connect_failure_hands_off_to_reconnect_loop() { + // 初回接続に失敗しても Err にせず、再接続ループ入りのハンドルを + // 残すこと。旧実装は Err を返して再試行の責務が宙に浮き、 + // 「起動直後のネットワーク未確立 = 永久にストリームなし」だった。 + let dir = tempfile::tempdir().unwrap(); + let db = Arc::new(crate::db::Database::open(&dir.path().join("test.db")).unwrap()); + let manager = StreamingManager::new( + Arc::new(NoopEmitter), + Arc::new(EventBus::new()), + db, + ); + + // 127.0.0.1:1 は即 connection refused になる + manager + .connect("acc-1", "127.0.0.1:1", "token") + .await + .expect("connect must not error on initial failure"); + + // ハンドルが生きていてコマンドを受け付ける (= ループが存在する) + manager.sub_note("acc-1", "note-1").await.unwrap(); + + // disconnect が backoff 中の Shutdown を届けて task を終了できる + manager.disconnect("acc-1").await; + } } From b06477c5bf9df8e46736c0b29d034c1ee4169e84 Mon Sep 17 00:00:00 2001 From: hitalin Date: Fri, 3 Jul 2026 09:56:59 +0900 Subject: [PATCH 2/3] =?UTF-8?q?test:=20=E5=86=8D=E6=8E=A5=E7=B6=9A?= =?UTF-8?q?=E3=83=AB=E3=83=BC=E3=83=97=E7=AA=81=E5=85=A5=E3=82=92=20stream?= =?UTF-8?q?-status=20=E3=81=AE=E8=A6=B3=E6=B8=AC=E3=81=A7=E6=A4=9C?= =?UTF-8?q?=E8=A8=BC=E3=81=99=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 旧テストは cmd channel の生存しか見ておらず、「ループに入らず return する」ミューテーションでも通ってしまう vacuous なものだった。 ChannelEmitter で reconnecting の 2 回目 (= ループ冒頭の emit) を待つ 形にし、ミューテーション注入で FAILED になることを確認済み。 --- src/streaming.rs | 29 ++++++++++++++++++++++++++--- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/src/streaming.rs b/src/streaming.rs index f5db242..8763710 100644 --- a/src/streaming.rs +++ b/src/streaming.rs @@ -1820,6 +1820,15 @@ mod tests { assert!(WS_READ_IDLE_TIMEOUT >= WS_PING_INTERVAL * 3); } + /// stream-status イベントを channel に流すテスト用 emitter。 + struct ChannelEmitter(mpsc::UnboundedSender<(String, Value)>); + + impl FrontendEmitter for ChannelEmitter { + fn emit(&self, event: &str, payload: Value) { + let _ = self.0.send((event.to_string(), payload)); + } + } + #[tokio::test] async fn connect_failure_hands_off_to_reconnect_loop() { // 初回接続に失敗しても Err にせず、再接続ループ入りのハンドルを @@ -1827,8 +1836,9 @@ mod tests { // 「起動直後のネットワーク未確立 = 永久にストリームなし」だった。 let dir = tempfile::tempdir().unwrap(); let db = Arc::new(crate::db::Database::open(&dir.path().join("test.db")).unwrap()); + let (tx, mut rx) = mpsc::unbounded_channel(); let manager = StreamingManager::new( - Arc::new(NoopEmitter), + Arc::new(ChannelEmitter(tx)), Arc::new(EventBus::new()), db, ); @@ -1839,8 +1849,21 @@ mod tests { .await .expect("connect must not error on initial failure"); - // ハンドルが生きていてコマンドを受け付ける (= ループが存在する) - manager.sub_note("acc-1", "note-1").await.unwrap(); + // "reconnecting" を 2 回観測する: 1 回目は connect() 自身の emit、 + // 2 回目は connection_task の再接続ループ冒頭の emit。2 回目が + // 来ること = ループが実際に生きて再試行していることの証明で、 + // 「ループに入らず return する」ミューテーションはここで落ちる + // (ミューテーション注入で FAILED になることを確認済み)。 + let mut reconnecting_count = 0; + while reconnecting_count < 2 { + let (event, payload) = tokio::time::timeout(Duration::from_secs(10), rx.recv()) + .await + .expect("timed out waiting for reconnect loop to emit stream-status") + .expect("emitter channel closed before loop emitted"); + if event == "stream-status" && payload["state"] == "reconnecting" { + reconnecting_count += 1; + } + } // disconnect が backoff 中の Shutdown を届けて task を終了できる manager.disconnect("acc-1").await; From 218edb0510aeb4d6e6306f9e42d2fece43280d1f Mon Sep 17 00:00:00 2001 From: hitalin Date: Fri, 3 Jul 2026 09:59:56 +0900 Subject: [PATCH 3/3] =?UTF-8?q?feat:=20=E6=8E=A5=E7=B6=9A=E7=8A=B6?= =?UTF-8?q?=E6=85=8B=E3=82=92=E8=BF=BD=E8=B7=A1=E3=81=97=E5=86=AA=E7=AD=89?= =?UTF-8?q?=20connect=20=E3=81=A7=E3=82=82=E7=8F=BE=E5=9C=A8=E7=8A=B6?= =?UTF-8?q?=E6=85=8B=E3=82=92=20emit=20=E3=81=99=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit フロントは復帰時にリスナーを張り直してから connect を呼び直すが、 既存接続への冪等 return は status を emit しないため、背景化中に 取り逃した状態遷移を補正できず、JS 側が「resolve = connected」と 楽観的に見なすロジックを持たざるを得なかった (初回失敗を Err に しなくなったことでこの楽観が誤りになる)。ConnectionHandle に実状態の AtomicBool を持たせ、冪等 return 時に connected / reconnecting を 正しく emit する。フロントは status イベントだけを信じればよくなる。 --- src/streaming.rs | 33 ++++++++++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/src/streaming.rs b/src/streaming.rs index 8763710..f41a3c7 100644 --- a/src/streaming.rs +++ b/src/streaming.rs @@ -221,6 +221,9 @@ struct ConnectionHandle { cmd_tx: mpsc::UnboundedSender, task: tokio::task::JoinHandle<()>, host: String, + /// connection_task が維持する実際の接続状態。connect() の冪等 return 時に + /// 現在状態を emit するために持つ (フロントは status イベントだけを信じる)。 + connected: Arc, } /// Handle for a polling task (non-WebSocket mode). @@ -290,7 +293,24 @@ impl StreamingManager { token: &str, ) -> Result<(), NoteDeckError> { let mut conns = self.connections.lock().await; - if conns.contains_key(account_id) { + if let Some(handle) = conns.get(account_id) { + // 冪等 return でも現在の実状態を emit する。フロントは復帰時に + // リスナーを張り直してから connect を呼び直すため、背景化中に + // 取り逃した状態遷移をここで補正できる (楽観的に connected と + // 見なす JS 側ロジックを置かないための供給側の責務)。 + let state = if handle.connected.load(std::sync::atomic::Ordering::Relaxed) { + "connected" + } else { + "reconnecting" + }; + emit_or_log!( + self.emitter, + "stream-status", + StreamStatusEvent { + account_id: account_id.to_string(), + state: state.to_string(), + } + ); return Ok(()); } @@ -319,6 +339,7 @@ impl StreamingManager { } }; let connected = initial_ws.is_some(); + let connected_flag = Arc::new(std::sync::atomic::AtomicBool::new(connected)); let (cmd_tx, cmd_rx) = mpsc::unbounded_channel(); @@ -333,6 +354,7 @@ impl StreamingManager { let api_client = self.api_client.clone(); let host_owned = host.to_string(); let token_owned = token.to_string(); + let connected_flag_task = connected_flag.clone(); let task = tokio::spawn(async move { connection_task( emitter, @@ -347,6 +369,7 @@ impl StreamingManager { cmd_rx, subscriptions, captured_notes, + connected_flag_task, ) .await; }); @@ -357,6 +380,7 @@ impl StreamingManager { cmd_tx, task, host: host.to_string(), + connected: connected_flag, }, ); @@ -936,7 +960,10 @@ async fn connection_task( mut cmd_rx: mpsc::UnboundedReceiver, subscriptions: Arc>>, captured_notes: Arc>>>, + connected_flag: Arc, ) { + use std::sync::atomic::Ordering; + let mut backoff_secs: u64 = 1; // Run the first session with the already-connected WebSocket. @@ -957,6 +984,7 @@ async fn connection_task( &captured_notes, ) .await; + connected_flag.store(false, Ordering::Relaxed); if matches!(reason, WsExitReason::Shutdown) { return; } @@ -964,6 +992,7 @@ async fn connection_task( // Reconnection loop loop { + connected_flag.store(false, Ordering::Relaxed); emit_or_log!( emitter, "stream-status", @@ -1009,6 +1038,7 @@ async fn connection_task( match ws_result { Ok(Ok((ws_stream, _))) => { backoff_secs = 1; // Reset backoff on success + connected_flag.store(true, Ordering::Relaxed); emit_or_log!( emitter, @@ -1033,6 +1063,7 @@ async fn connection_task( &captured_notes, ) .await; + connected_flag.store(false, Ordering::Relaxed); if matches!(reason, WsExitReason::Shutdown) { return;