From 9b29e19f24a90c330ec62ae38747bfb3e9a83f2f Mon Sep 17 00:00:00 2001 From: hitalin Date: Thu, 2 Jul 2026 10:46:29 +0900 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20WebSocket=20=E3=83=A2=E3=83=BC?= =?UTF-8?q?=E3=83=89=E3=81=AE=20subNote=20=E8=B3=BC=E8=AA=AD=E3=82=92?= =?UTF-8?q?=E5=86=8D=E6=8E=A5=E7=B6=9A=E6=99=82=E3=81=AB=20replay=20?= =?UTF-8?q?=E3=81=99=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WebSocket モードの sub_note は WsCommand を送るだけで captured_notes に 記録していなかったため、再接続 (idle watchdog / ネットワークエラー) で サーバー側の subNote 購読が失われたまま二度と復元されず、捕捉中ノートの noteUpdated (リアクション等) がサイレントに止まっていた。 - sub_note / unsub_note は両モード共通で captured_notes を更新する - run_ws_session がチャンネル購読の replay 後に captured_notes からも subNote を再送する - disconnect はアカウントの captured_notes も破棄する (subscriptions と 同じライフサイクル) - 副次効果: realtime→polling のモード切替でも捕捉が引き継がれる --- src/streaming.rs | 82 ++++++++++++++++++++++++++++++++++++------------ 1 file changed, 62 insertions(+), 20 deletions(-) diff --git a/src/streaming.rs b/src/streaming.rs index 2975ede..37bbc6d 100644 --- a/src/streaming.rs +++ b/src/streaming.rs @@ -310,6 +310,7 @@ impl StreamingManager { let account_id_owned = account_id.to_string(); let url_owned = url.clone(); let subscriptions = self.subscriptions.clone(); + let captured_notes = self.captured_notes.clone(); let emitter = self.emitter.clone(); let event_bus = self.event_bus.clone(); let db = self.db.clone(); @@ -330,6 +331,7 @@ impl StreamingManager { ws_stream, cmd_rx, subscriptions, + captured_notes, ) .await; }); @@ -379,6 +381,12 @@ impl StreamingManager { // Remove all subscriptions for this account let mut subs = self.subscriptions.write().await; subs.retain(|_, info| info.account_id != account_id); + drop(subs); + + // Note captures follow the same lifecycle as subscriptions + let mut captured = self.captured_notes.write().await; + captured.remove(account_id); + drop(captured); emit_or_log!( self.emitter, @@ -780,7 +788,19 @@ impl StreamingManager { } pub async fn sub_note(&self, account_id: &str, note_id: &str) -> Result<(), NoteDeckError> { - // WebSocket mode: send subNote command + // Record in captured_notes in BOTH modes. Without this, WebSocket-mode + // captures are not in any table, so the reconnect replay in + // run_ws_session never restores them and noteUpdated events silently + // stop after the first reconnect (idle watchdog, network error, ...). + { + let mut captured = self.captured_notes.write().await; + captured + .entry(account_id.to_string()) + .or_default() + .insert(note_id.to_string()); + } + + // WebSocket mode: also send subNote now (polling mode reads the table) let conns = self.connections.lock().await; if let Some(handle) = conns.get(account_id) { return handle @@ -790,19 +810,21 @@ impl StreamingManager { }) .map_err(|_| NoteDeckError::ConnectionClosed); } - drop(conns); - - // Polling mode: add to captured_notes set for batch polling - let mut captured = self.captured_notes.write().await; - captured - .entry(account_id.to_string()) - .or_default() - .insert(note_id.to_string()); Ok(()) } pub async fn unsub_note(&self, account_id: &str, note_id: &str) -> Result<(), NoteDeckError> { - // WebSocket mode + { + let mut captured = self.captured_notes.write().await; + if let Some(set) = captured.get_mut(account_id) { + set.remove(note_id); + if set.is_empty() { + captured.remove(account_id); + } + } + } + + // WebSocket mode: also send unsubNote now let conns = self.connections.lock().await; if let Some(handle) = conns.get(account_id) { return handle @@ -812,13 +834,6 @@ impl StreamingManager { }) .map_err(|_| NoteDeckError::ConnectionClosed); } - drop(conns); - - // Polling mode: remove from captured_notes - let mut captured = self.captured_notes.write().await; - if let Some(set) = captured.get_mut(account_id) { - set.remove(note_id); - } Ok(()) } @@ -905,6 +920,7 @@ async fn connection_task( initial_ws: WsStream, mut cmd_rx: mpsc::UnboundedReceiver, subscriptions: Arc>>, + captured_notes: Arc>>>, ) { let mut backoff_secs: u64 = 1; @@ -920,6 +936,7 @@ async fn connection_task( initial_ws, &mut cmd_rx, &subscriptions, + &captured_notes, ) .await; if matches!(reason, WsExitReason::Shutdown) { @@ -950,9 +967,10 @@ async fn connection_task( cmd = cmd_rx.recv() => { match cmd { Some(WsCommand::Shutdown) | None => break true, - // Subscribe/Unsubscribe: safe to drop here because the - // subscriptions table is already updated by the caller. - // run_ws_session will re-subscribe from that table. + // Subscribe/Unsubscribe/SubNote/UnsubNote: safe to drop + // here because the subscriptions and captured_notes + // tables are already updated by the caller. + // run_ws_session replays from those tables. _ => {} } } @@ -993,6 +1011,7 @@ async fn connection_task( ws_stream, &mut cmd_rx, &subscriptions, + &captured_notes, ) .await; @@ -1025,6 +1044,7 @@ async fn run_ws_session( ws_stream: WsStream, cmd_rx: &mut mpsc::UnboundedReceiver, subscriptions: &Arc>>, + captured_notes: &Arc>>>, ) -> WsExitReason { let (write, read) = ws_stream.split(); let write = Arc::new(Mutex::new(write)); @@ -1053,6 +1073,28 @@ async fn run_ws_session( } } + // Replay note captures. Unlike channel subscriptions these have no + // server-side persistence either, so a reconnect without this replay + // permanently loses noteUpdated (reactions, edits) for captured notes. + let to_recapture: Vec = { + let captured = captured_notes.read().await; + captured + .get(account_id) + .map(|set| set.iter().cloned().collect()) + .unwrap_or_default() + }; + + if !to_recapture.is_empty() { + let mut w = write.lock().await; + for note_id in &to_recapture { + let msg = json!({ "type": "subNote", "body": { "id": note_id } }); + if let Err(e) = w.send(Message::Text(msg.to_string().into())).await { + tracing::warn!(error = %e, "subNote replay send failed"); + break; + } + } + } + ws_loop( emitter, event_bus, From 247d29e161c6c4801be37adce47b6bb1156303f8 Mon Sep 17 00:00:00 2001 From: hitalin Date: Thu, 2 Jul 2026 11:21:46 +0900 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20=E3=82=BB=E3=83=83=E3=82=B7=E3=83=A7?= =?UTF-8?q?=E3=83=B3=E5=86=85=20subNote=20=E9=80=81=E4=BF=A1=E3=82=92=20de?= =?UTF-8?q?dup=20=E3=81=97=E3=81=A6=E4=BA=8C=E9=87=8D=E8=B3=BC=E8=AA=AD?= =?UTF-8?q?=E3=82=92=E9=98=B2=E3=81=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Misskey の subNote は refcount 式で冪等でないため、replay と切断中に queue された SubNote コマンドが二重送信されると、unsubNote 1 回では サーバー側購読が残ってしまう。セッション内で送信済みのノート id を 追跡し、重複送信をスキップする。 --- src/streaming.rs | 36 +++++++++++++++++++++++++----------- 1 file changed, 25 insertions(+), 11 deletions(-) diff --git a/src/streaming.rs b/src/streaming.rs index 37bbc6d..2f1db7d 100644 --- a/src/streaming.rs +++ b/src/streaming.rs @@ -1076,20 +1076,24 @@ async fn run_ws_session( // Replay note captures. Unlike channel subscriptions these have no // server-side persistence either, so a reconnect without this replay // permanently loses noteUpdated (reactions, edits) for captured notes. - let to_recapture: Vec = { + // + // session_note_subs tracks what this session has already sent subNote + // for. Misskey refcounts subNote (it is NOT idempotent), so a queued + // WsCommand::SubNote that raced with this replay must not be sent twice + // or a single unsubNote would leave a dangling server-side subscription. + let session_note_subs: HashSet = { let captured = captured_notes.read().await; - captured - .get(account_id) - .map(|set| set.iter().cloned().collect()) - .unwrap_or_default() + captured.get(account_id).cloned().unwrap_or_default() }; - if !to_recapture.is_empty() { + if !session_note_subs.is_empty() { let mut w = write.lock().await; - for note_id in &to_recapture { + for note_id in &session_note_subs { let msg = json!({ "type": "subNote", "body": { "id": note_id } }); if let Err(e) = w.send(Message::Text(msg.to_string().into())).await { tracing::warn!(error = %e, "subNote replay send failed"); + // 送信失敗分は未購読のまま残るが、直後に接続自体が落ちて + // 再接続 replay でやり直すので個別追跡はしない break; } } @@ -1107,6 +1111,7 @@ async fn run_ws_session( write, cmd_rx, subscriptions, + session_note_subs, ) .await } @@ -1124,6 +1129,10 @@ async fn ws_loop( write: WsWrite, cmd_rx: &mut mpsc::UnboundedReceiver, subscriptions: &Arc>>, + // このセッションで subNote 済みのノート id。replay 分を含む。 + // Misskey の subNote は refcount 式で冪等でないため、replay と + // 切断中に queue された SubNote コマンドの二重送信をここで排除する。 + mut session_note_subs: HashSet, ) -> WsExitReason { let mut ping_interval = tokio::time::interval(WS_PING_INTERVAL); ping_interval.tick().await; // skip the first immediate tick @@ -1191,13 +1200,18 @@ async fn ws_loop( } } Some(WsCommand::SubNote { id }) => { - let msg = json!({ "type": "subNote", "body": { "id": id } }); - let mut w = write.lock().await; - if let Err(e) = w.send(Message::Text(msg.to_string().into())).await { - tracing::warn!(error = %e, "subNote send failed"); + // insert が false = このセッションで購読済み (replay 済み + // or 二重コマンド)。重複送信すると refcount が狂う。 + if session_note_subs.insert(id.clone()) { + let msg = json!({ "type": "subNote", "body": { "id": id } }); + let mut w = write.lock().await; + if let Err(e) = w.send(Message::Text(msg.to_string().into())).await { + tracing::warn!(error = %e, "subNote send failed"); + } } } Some(WsCommand::UnsubNote { id }) => { + session_note_subs.remove(&id); let msg = json!({ "type": "unsubNote", "body": { "id": id } }); let mut w = write.lock().await; if let Err(e) = w.send(Message::Text(msg.to_string().into())).await {