Skip to content

fix(streaming): polling 稼働中の connect が WebSocket を復活させないようにする #59

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

Merged
hitalin merged 4 commits into from
Aug 7, 2026
Merged

fix(streaming): polling 稼働中の connect が WebSocket を復活させないようにする #59

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b051a7f
Merge pull request #54 from notedeck-dev/develop
hitalin Aug 3, 2026
272edae
Merge pull request #56 from notedeck-dev/develop
hitalin Aug 4, 2026
4d61fb2
fix(streaming): polling 稼働中の connect が WebSocket を復活させないようにする
hitalin Aug 7, 2026
1164ca1
Merge branch 'develop' into fix/connect-respects-polling-mode
hitalin Aug 7, 2026
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
49 changes: 49 additions & 0 deletions src/streaming.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -472,6 +472,23 @@ impl StreamingManager {
host: &str,
token: &str,
) -> Result<(), NoteDeckError> {
// polling がこのアカウントのストリームを供給中なら WS を張らない。
// connect は「ストリームを現在のモードで確保する」であって「WS を
// 強制する」ではない。カラムのマウントや復帰は connect を無条件に
// 呼ぶため、ここで弾かないと polling モードが黙って崩れる
// (notedeck#1004)。WS へ戻す唯一の経路は set_mode("realtime")。
{
let polls = self.poll_connections.lock().await;
if polls.contains_key(account_id) {
self.emitter
.emit(StreamEvent::Status(Box::new(StreamStatusEvent {
account_id: account_id.to_string(),
state: StreamConnectionState::Connected,
})));
Comment on lines +483 to +487

@coderabbitai coderabbitai Bot Aug 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/usr/bin/env bash
set -euo pipefail

ast-grep outline src/streaming.rs --items all --match 'StreamingManager|EventBus'
fd -i '^event_bus\.rs$' . -x ast-grep outline {} --items all
rg -n -C 4 --glob '*.rs' 'StreamEvent::Status|StreamStatusEvent|EventBus|FrontendEmitter' .

Repository: notedeck-dev/notecli

Length of output: 24227

🏁 Script executed:

#!/usr/bin/env bash
set -euo pipefail

# Read focused streaming/event_bus/server sections and run read-only behavioral checks.
sed -n '1,90p' src/streaming.rs
sed -n '440,700p' src/streaming.rs
sed -n '1720,1940p' src/streaming.rs
sed -n '720,815p' src/http_server.rs
rg -n --glob '*.rs' 'stream-status|s:.*stream-status|sse_event_type|payload_value|event_bus\.send|EventBusEmitter|StreamStatusEvent' .
python3 - <<'PY'
from pathlib import Path
import re
p=Path('src/streaming.rs')
s=p.read_text()
checks={
    'line483_emitter_status': bool(re.search(r'(?ms)$', re.sub(r'//.*','',s)[460:490] )), # placeholder
    'emit_both_contains_polling_capture_comment': 'status / polling 由来 capture は従来どおり emitter のみ' in s,
    'event_bus_send_is_sse_json_payload': 'let data = serde_json::to_string(&event)?' in p.read_text(),
    'status_event_type': 'stream-status' in s and 'Status(Box<StreamStatusEvent>)' in s,
}
for k,v in checks.items():
    print(f'{k}={v}')
PY

PY

Repository: notedeck-dev/notecli

Length of output: 28534

Connected status を EventBus 経由で配信してください。

このコードパスは self.emitter.emit(...) のみを実行し、EventBus へ送信しません。SSE/イベントバス消費者は StreamEvent::Status を受信できません。Status は emit_both(emitter, &event_bus, ...) に移してください。

🤖 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 `@src/streaming.rs` around lines 483 - 487, Update the Connected status path to
dispatch the StreamEvent::Status through EventBus as well as self.emitter by
using the existing emit_both helper with the relevant emitter and event_bus
references, ensuring SSE and event-bus consumers receive the status.

Source: Coding guidelines

return Ok(());
}
}

let mut conns = self.connections.lock().await;
if let Some(handle) = conns.get(account_id) {
// 冪等 return でも現在の実状態を emit する。フロントは復帰時に
Expand Down Expand Up @@ -2457,6 +2474,38 @@ mod tests {
manager.disconnect("acc-1").await;
}

#[tokio::test]
async fn connect_does_not_resurrect_websocket_while_polling() {
// フロントはカラムのマウントや復帰のたびに connect() を無条件に呼ぶ。
// polling 中に WS が復活すると、永続化されたモードと実動作が乖離する
// (notedeck#1004)。connect は「ストリームを現在のモードで確保する」であって
// 「WS を強制する」ではない。
let (_dir, manager, mut rx) = manager_with_dead_connection(&["acc-1"]).await;
manager
.set_mode("acc-1", "127.0.0.1:1", "token", "polling", Some(60_000))
.await
.unwrap();
drain_status(&mut rx);

manager
.connect("acc-1", "127.0.0.1:1", "token")
.await
.unwrap();

assert!(
manager.connections.lock().await.is_empty(),
"polling 中の connect は WS を張らない"
);
assert!(manager.poll_connections.lock().await.contains_key("acc-1"));
// polling がストリームを供給中なので Connected として報告する
assert!(
drain_status(&mut rx).contains(&StreamConnectionState::Connected),
"冪等 return でも現在状態を emit すること"
);

manager.disconnect("acc-1").await;
}

#[tokio::test]
async fn unsubscribe_drops_the_subscription_in_both_modes() {
let (_dir, manager, _rx) = manager_with_dead_connection(&["acc-1"]).await;
Expand Down
Loading