fix: recover durable SSH sessions stuck falsely-attached after stream timeout (#3439) - #3460
fix: recover durable SSH sessions stuck falsely-attached after stream timeout (#3439)#3460akademic123 wants to merge 1 commit into
Conversation
… timeout (upstream wavetermdev#3439) Remote jobmanager: - DataSender.SendData now returns error instead of swallowing it - StreamManager senderLoop: on send error, drop client to disconnected (buffering) mode and invoke OnSendError callback - JobManager closes the stale client socket so the main server observes the route drop and auto-reconnects; remote shell + 2MB scrollback survive Main server (wavesrv): - runOutputLoop wrapped with 45s stall watchdog; on stall, force remote disconnect of the dead client and ReconnectJob (fresh attach + stream resume from persisted offset) Tests: TestSendErrorDisconnectsClient covers disconnect + reattach flow
|
|
WalkthroughThe change adds error returns to stream sender interfaces and implementations. Failed sends now disconnect stale clients, invoke asynchronous cleanup, and preserve buffered stream data for reconnection. Job output reads now use a 45-second stall timeout. Stalled streams trigger job and connection validation, stale-client disconnection, job disconnection marking, and reconnection. Estimated code review effort: 4 (Complex) | ~45 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (3)
pkg/jobmanager/streammanager.go (1)
359-366: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winConfirm
OnSendErrorcallers never block the sender loop.
senderLoopcallsOnSendErrorinline. The comment on Line 64 states the call is asynchronous.SetupJobManagersatisfies that by starting a goroutine, but the contract is not enforced here. A blocking callback stops all further stream progress for this stream.Consider invoking the callback in a goroutine inside
senderLoop, so the documented contract holds for every caller.🤖 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 `@pkg/jobmanager/streammanager.go` around lines 359 - 366, Update the OnSendError invocation in senderLoop so the callback always runs asynchronously in a separate goroutine, preventing a blocking handler from stopping sender-loop progress. Preserve the existing nil check, error argument, and client-disconnection handling.pkg/jobcontroller/jobcontroller.go (1)
876-906: 🩺 Stability & Availability | 🔵 Trivial | 🏗️ Heavy liftAvoid one goroutine per read and shared ownership of
buf.
readWithStallTimeoutstarts a goroutine for every read iteration. For a busy stream this creates a goroutine per 4 KB chunk. The goroutine also writes into the caller'sbuf, whichrunOutputLoopowns across iterations. On the timeout path and on thectx.Donepath the function returns while that goroutine is still running, sobufhas two owners and the bytes it later reads are dropped.
runOutputLoopcurrently breaks out of the loop in both paths, sobufis not read again. The ownership rule is still fragile against future changes to the loop.Prefer a single long-lived read goroutine per output loop that sends results over a channel, with its own buffer per read. The loop then applies the stall timer to the channel receive.
🤖 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 `@pkg/jobcontroller/jobcontroller.go` around lines 876 - 906, Refactor runOutputLoop and readWithStallTimeout to use one long-lived read goroutine per output loop instead of spawning a goroutine for each read. Have that goroutine perform sequential reader.Read calls into a private buffer and send read results through a channel, while runOutputLoop applies StreamStallTimeout and context cancellation when receiving results; never let asynchronous reads write into the caller-owned buf or continue using it after timeout/cancellation.pkg/jobmanager/streammanager_test.go (1)
361-368: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the unused
failingWritertype.
TestSendErrorDisconnectsClientusestestWriter.SetSendError, andfailingWriterhas no references elsewhere in the repository. Delete the dead code unless it is needed for another test.🤖 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 `@pkg/jobmanager/streammanager_test.go` around lines 361 - 368, Remove the unused failingWriter type and its SendData method from the test file; preserve TestSendErrorDisconnectsClient’s existing testWriter.SetSendError setup and make no other changes.
🤖 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 `@pkg/jobcontroller/jobcontroller.go`:
- Around line 931-938: Replace the duplicated cooldown check in the
stalled-stream recovery path with a call to shouldAttemptAutoReconnect, reusing
its shared lastAutoReconnectAttempt state and preserving the existing skip
behavior and logging for recently attempted recoveries.
- Around line 960-969: Update handleStalledStream around SetJobConnStatus and
ReconnectJob to wait until the job route is actually unregistered before
reconnecting, ensuring reconnect does not exit early through CheckJobConnected
and restartStreaming is invoked. Alternatively, propagate a reconnect result
that reflects whether streaming restarted, and only log success when
reconnection truly occurred.
In `@pkg/jobmanager/jobmanager.go`:
- Around line 214-229: The asynchronous send-error path must identify the failed
sender before closing a connection. Update handleStreamSendError and its callers
to accept or retain the failed MainServerConn identity, then clear and close
jm.connectedStreamClient only when it still matches that sender; leave a newer
connected client untouched.
In `@pkg/jobmanager/streammanager.go`:
- Around line 63-66: Make OnSendError private and add a lock-protected
setter/accessor for updating and reading the callback. Update senderLoop to
retrieve the callback while holding sm.lock, and update callers such as
TestSendErrorDisconnectsClient to use the setter so assignments cannot race with
the sender loop.
---
Nitpick comments:
In `@pkg/jobcontroller/jobcontroller.go`:
- Around line 876-906: Refactor runOutputLoop and readWithStallTimeout to use
one long-lived read goroutine per output loop instead of spawning a goroutine
for each read. Have that goroutine perform sequential reader.Read calls into a
private buffer and send read results through a channel, while runOutputLoop
applies StreamStallTimeout and context cancellation when receiving results;
never let asynchronous reads write into the caller-owned buf or continue using
it after timeout/cancellation.
In `@pkg/jobmanager/streammanager_test.go`:
- Around line 361-368: Remove the unused failingWriter type and its SendData
method from the test file; preserve TestSendErrorDisconnectsClient’s existing
testWriter.SetSendError setup and make no other changes.
In `@pkg/jobmanager/streammanager.go`:
- Around line 359-366: Update the OnSendError invocation in senderLoop so the
callback always runs asynchronously in a separate goroutine, preventing a
blocking handler from stopping sender-loop progress. Preserve the existing nil
check, error argument, and client-disconnection handling.
🪄 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: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: c143535a-472f-4efa-b14b-70cee6975b37
📒 Files selected for processing (6)
cmd/test-streammanager/main-test-streammanager.gopkg/jobcontroller/jobcontroller.gopkg/jobmanager/jobmanager.gopkg/jobmanager/mainserverconn.gopkg/jobmanager/streammanager.gopkg/jobmanager/streammanager_test.go
| // prevent overlapping recovery attempts for the same job | ||
| lastAttempt, exists := lastAutoReconnectAttempt.GetEx(jobId) | ||
| now := time.Now().Unix() | ||
| if exists && time.Duration(now-lastAttempt)*time.Second < AutoReconnectCooldown { | ||
| log.Printf("[job:%s] stalled stream: recovery attempted recently, skipping", jobId) | ||
| return | ||
| } | ||
| lastAutoReconnectAttempt.Set(jobId, now) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Reuse shouldAttemptAutoReconnect instead of duplicating the cooldown logic.
Lines 932-938 reimplement the body of shouldAttemptAutoReconnect (Lines 387-403) against the same lastAutoReconnectAttempt map. The two paths now suppress each other: a stall recovery blocks a normal auto-reconnect for AutoReconnectCooldown, and the reverse also applies.
If shared suppression is intended, call the existing helper. If stall recovery needs an independent cooldown, use a separate map.
♻️ Proposed fix: call the existing helper
- // prevent overlapping recovery attempts for the same job
- lastAttempt, exists := lastAutoReconnectAttempt.GetEx(jobId)
- now := time.Now().Unix()
- if exists && time.Duration(now-lastAttempt)*time.Second < AutoReconnectCooldown {
- log.Printf("[job:%s] stalled stream: recovery attempted recently, skipping", jobId)
- return
- }
- lastAutoReconnectAttempt.Set(jobId, now)
+ // prevent overlapping recovery attempts for the same job
+ if !shouldAttemptAutoReconnect(jobId) {
+ log.Printf("[job:%s] stalled stream: recovery attempted recently, skipping", jobId)
+ return
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // prevent overlapping recovery attempts for the same job | |
| lastAttempt, exists := lastAutoReconnectAttempt.GetEx(jobId) | |
| now := time.Now().Unix() | |
| if exists && time.Duration(now-lastAttempt)*time.Second < AutoReconnectCooldown { | |
| log.Printf("[job:%s] stalled stream: recovery attempted recently, skipping", jobId) | |
| return | |
| } | |
| lastAutoReconnectAttempt.Set(jobId, now) | |
| // prevent overlapping recovery attempts for the same job | |
| if !shouldAttemptAutoReconnect(jobId) { | |
| log.Printf("[job:%s] stalled stream: recovery attempted recently, skipping", jobId) | |
| return | |
| } |
🤖 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 `@pkg/jobcontroller/jobcontroller.go` around lines 931 - 938, Replace the
duplicated cooldown check in the stalled-stream recovery path with a call to
shouldAttemptAutoReconnect, reusing its shared lastAutoReconnectAttempt state
and preserving the existing skip behavior and logging for recently attempted
recoveries.
| // the route should drop and re-register as the job manager re-attaches; | ||
| // reconnect restarts streaming from the persisted offset | ||
| SetJobConnStatus(jobId, JobConnStatus_Disconnected) | ||
| reconnectCtx, reconnectCancel := context.WithTimeout(context.Background(), 15*time.Second) | ||
| defer reconnectCancel() | ||
| if err := ReconnectJob(reconnectCtx, jobId, nil); err != nil { | ||
| log.Printf("[job:%s] stalled stream: reconnect failed: %v", jobId, err) | ||
| } else { | ||
| log.Printf("[job:%s] stalled stream: reconnect succeeded", jobId) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Inspect CheckJobConnected and the remote job-manager disconnect handler.
rg -n -C 15 'func CheckJobConnected' pkg/jobcontroller
rg -n -C 20 'RemoteDisconnectFromJobManager' --glob '!**/*_test.go'Repository: wavetermdev/waveterm
Length of output: 2160
🏁 Script executed:
#!/bin/bash
set -u
# Locate and inspect the relevant functions and call graph candidates.
printf 'Files containing jobcontroller.go:\n'
fd -a 'jobcontroller\.go$' .
printf '\nRelevant CheckJobConnected section:\n'
sed -n '520,635p' pkg/jobcontroller/jobcontroller.go 2>/dev/null || true
printf '\nRelevant doReconnectJob/handleStalledStream section:\n'
sed -n '900,1220p' pkg/jobcontroller/jobcontroller.go 2>/dev/null || true
printf '\nSearch ReconnectJob / restartStreaming / jobconn status / disconnect command:\n'
rg -n -C 12 'func doReconnectJob|handleStalledStream|SetJobConnStatus|GetJobConnStatus|restartStreaming|ReconnectJob|RestartStreaming|RemoteDisconnectFromJobManager|DisconnectFromJobManager|RegisterJobRoute|UnregisterJobRoute' pkg --glob '!**/*_test.go'Repository: wavetermdev/waveterm
Length of output: 50376
🏁 Script executed:
#!/bin/bash
set -u
printf 'Candidates for RemoteDisconnectFromJobManager definition:\n'
rg -n -C 12 'RemoteDisconnectFromJobManagerCommand|remotedisconnectfromjobmanager|Remote Disconnect|DisconnectFromJobManager' pkg --glob '!**/*_test.go' --glob '!pkg/wshrpc/wshclient/wshclient.go' --glob '!pkg/wshrpc/wshserver/api_handler.go' --max-count 50
printf '\nCheckJobManagerStatus update candidates:\n'
rg -n -C lo 'CheckJobManagerStatus|JobManagerStatus|stream.*loop|output.*loop|runOutputLoop|runStream' pkg --glob '!**/*_test.go' --max-count 80
printf '\nInspect likely handler files:\n'
fd -a '.*\.go$' pkg | rg 'job|manager|wshserver|stream|output' | sed -n '1,120p'Repository: wavetermdev/waveterm
Length of output: 12988
🏁 Script executed:
#!/bin/bash
set -u
printf 'Inspect remote disconnect and cleanup definitions:\n'
sed -n '200,350p' pkg/wshrpc/wshremote/wshremote_job.go
printf '\nInspect JobManagerConnection and CleanupFn definitions/usages:\n'
rg -n -C 12 'type .*JobManagerConnection|JobManagerConnection|CleanupFn|OnSessionRemoved|OnRoute*|RemoveRoute|Unregister|WshRpcAddRoute|WaitForRegister|Register' pkg/wshrpc pkg/wshutil pkg/jobcontroller --glob '!**/*_test.go'
printf '\nInspect doReconnectJob complete implementation:\n'
sed -n '1210,1278p' pkg/jobcontroller/jobcontroller.goRepository: wavetermdev/waveterm
Length of output: 41882
Wait for the route to unregister before reconnecting.
handleStalledStream clears ConnStatus_Connected immediately after dispatching RemoteDisconnectFromJobManagerCommand, but that RPC only calls CleanupFn() on the remote side. If ReconnectJob runs before the route unregisters, CheckJobConnected sees ConnStatus_Connected, doReconnectJob returns nil, and restartStreaming is never called. Wait for the job route to be unregistered or return a reconnect result instead of logging success.
🤖 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 `@pkg/jobcontroller/jobcontroller.go` around lines 960 - 969, Update
handleStalledStream around SetJobConnStatus and ReconnectJob to wait until the
job route is actually unregistered before reconnecting, ensuring reconnect does
not exit early through CheckJobConnected and restartStreaming is invoked.
Alternatively, propagate a reconnect result that reflects whether streaming
restarted, and only log success when reconnection truly occurred.
| // handleStreamSendError is invoked (async) when the stream manager fails to | ||
| // push data to the main server. The attached client connection is stale at | ||
| // this point: keep the job + buffered stream alive, but close the stale | ||
| // socket so the main server observes the route drop and auto-reconnects. | ||
| func (jm *JobManager) handleStreamSendError(sendErr error) { | ||
| jm.lock.Lock() | ||
| client := jm.connectedStreamClient | ||
| if client != nil { | ||
| jm.connectedStreamClient = nil | ||
| } | ||
| jm.lock.Unlock() | ||
| if client != nil { | ||
| log.Printf("handleStreamSendError: closing stale client connection after send error: %v\n", sendErr) | ||
| client.Close() | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Locate every read/write of connectedStreamClient and the stream attach handler.
rg -n -C 8 'connectedStreamClient' pkg/jobmanager
rg -n -C 10 'ClientConnected\(' pkg/jobmanagerRepository: wavetermdev/waveterm
Length of output: 18730
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect StreamManager lock ordering, send error callback, and JobManager callback wiring without executing repo code.
sed -n '1,240p' pkg/jobmanager/streammanager.go
printf '\n--- jobmanager relevant sections ---\n'
sed -n '160,240p' pkg/jobmanager/jobmanager.go
printf '%s\n' '--- OnSendError callsigns ---'
rg -n -C 5 'OnSendError|HandleStreamSendError|handleStreamSendError|ClientConnected\(' pkg/jobmanager -g '*.go'Repository: wavetermdev/waveterm
Length of output: 20512
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect senderLoop context around the asynchronous OnSendError callback.
sed -n '330,375p' pkg/jobmanager/streammanager.goRepository: wavetermdev/waveterm
Length of output: 1291
Pass the failed sender identity to handleStreamSendError.
handleStreamSendError runs asynchronously after sm.ClientDisconnected() and then closes any non-nil jm.connectedStreamClient. If the stream reconnects during that window, jm.connectedStreamClient points to the new MainServerConn, so the handler clears and closes the healthy session. Keep the failed MainServerConn or sender identity and close only if it still matches the current jm.connectedStreamClient.
🤖 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 `@pkg/jobmanager/jobmanager.go` around lines 214 - 229, The asynchronous
send-error path must identify the failed sender before closing a connection.
Update handleStreamSendError and its callers to accept or retain the failed
MainServerConn identity, then clear and close jm.connectedStreamClient only when
it still matches that sender; leave a newer connected client untouched.
|
|
||
| // OnSendError, if set, is called (asynchronously) when a SendData call fails. | ||
| // Used to tear down the stale client connection so a fresh attach can proceed. | ||
| OnSendError func(err error) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Guard OnSendError with the existing lock, or document that it must be set before the sender loop starts.
senderLoop reads sm.OnSendError at Line 363 without holding sm.lock. The field is exported, so callers can assign it at any time. TestSendErrorDisconnectsClient assigns sm.OnSendError after ClientConnected has started the sender loop (pkg/jobmanager/streammanager_test.go Lines 405-410), so go test -race on that test can report a data race on this field.
Prefer a setter that takes the lock and a locked read in senderLoop.
🔒 Proposed fix: private field plus locked accessors
- // OnSendError, if set, is called (asynchronously) when a SendData call fails.
- // Used to tear down the stale client connection so a fresh attach can proceed.
- OnSendError func(err error)
+ // onSendError, if set, is called (asynchronously) when a SendData call fails.
+ // Used to tear down the stale client connection so a fresh attach can proceed.
+ onSendError func(err error)
}
+
+func (sm *StreamManager) SetOnSendError(fn func(err error)) {
+ sm.lock.Lock()
+ defer sm.lock.Unlock()
+ sm.onSendError = fn
+}
+
+func (sm *StreamManager) getOnSendError() func(err error) {
+ sm.lock.Lock()
+ defer sm.lock.Unlock()
+ return sm.onSendError
+} err := sender.SendData(*pkt)
if err != nil {
log.Printf("senderLoop: send error (seq=%d): %v -- marking client disconnected\n", pkt.Seq, err)
sm.ClientDisconnected()
- if sm.OnSendError != nil {
- sm.OnSendError(err)
+ if cb := sm.getOnSendError(); cb != nil {
+ cb(err)
}
}Also applies to: 359-366
🤖 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 `@pkg/jobmanager/streammanager.go` around lines 63 - 66, Make OnSendError
private and add a lock-protected setter/accessor for updating and reading the
callback. Update senderLoop to retrieve the callback while holding sm.lock, and
update callers such as TestSendErrorDisconnectsClient to use the setter so
assignments cannot race with the sender loop.
2 participants
Summary
Fixes #3439 (and the root cause behind #2985): a durable SSH terminal intermittently becomes completely non-interactive while the shield keeps showing Durable Session (Attached). Recovery previously required restarting Wave (or Force Restart Controller).
Root cause (two-sided)
1. Remote job manager swallows stream send errors.
routedDataSender.SendData(pkg/jobmanager/mainserverconn.go) loggedSendData: error sending stream data: timeout sending requestand then discarded the error. TheStreamManagertherefore kept the stream inconnectedstate forever:2. Main server has no stall detection.
jobcontroller.runOutputLoopblocks inreader.Readindefinitely. The only auto-reconnect trigger is theRouteDownevent (handleRouteDownEvent) — but in this failure mode the job route never goes down, so no reconnect is ever attempted. The block stays frozen until the whole app (or the connection controller) is restarted.Fix
Remote job manager (
pkg/jobmanager):DataSender.SendDatanow returns anerrorinstead of swallowing it.StreamManager.senderLoop: on send error, the stream drops back to disconnected (buffering) mode viaClientDisconnected()— the 2 MB circular buffer keeps capturing PTY output — and invokes a newOnSendErrorcallback.JobManager.handleStreamSendError: closes the stale client socket. This makes the job route drop on the main server, which triggers the existing auto-reconnect path (attemptAutoReconnect→ReconnectJob→ new attach → stream resumes from the buffered offset with no data loss, exactly like after an app restart).Main server (
pkg/jobcontroller/jobcontroller.go):runOutputLoopreads are wrapped in a 45 s stall watchdog (StreamStallTimeout). If no data/EOF arrives within the timeout,handleStalledStreamruns: it asks the remote connection to drop the job manager's dead client (RemoteDisconnectFromJobManagerCommand) and then performs aReconnectJob, which does a fresh attach +PrepareConnect/StartStreamfrom the persisted offset. Remote shell state and scrollback survive. A 30 s cooldown (AutoReconnectCooldown, shared with the existing reconnect path) prevents reconnect flapping.Tests
TestSendErrorDisconnectsClientinpkg/jobmanager/streammanager_test.go: verifies that a send failure (1) drops the stream to disconnected mode, (2) firesOnSendError, and (3) a new client can attach and receives the buffered data.go build ./...clean,go test ./pkg/jobmanager/ ./pkg/streamclient/pass.Notes / trade-offs
totalGap=0) but shows up in logs. Happy to raise the default or make it configurable if preferred.wshbinary must be re-deployed to remote hosts for the job-manager half of the fix to take effect (Wave handles this automatically on connect with the new build).jobmanager.DataSender(implemented byroutedDataSenderand thecmd/test-streammanagerhelper, both updated).Reproduction (for validation)
SendData: timeout sending requestin the remote durable-session log).stream stalled ... forcing job manager reconnect/closing stale client connection), same PIDs, session responsive again.