fix: use session-scoped context to prevent stuck reconnects
When the WebSocket breaks (broken pipe), the main loop can get stuck in handleMessage -> dispatchJob -> submit if the worker pool is full. submit blocks on ctx.Done() but ctx is the parent context which only cancels on SIGINT/SIGTERM, not on connection errors. Create a session-scoped context that cancels when read/write errors occur. Pass it to handleMessage so blocked submit calls unblock immediately, allowing the main loop to receive the write error and reconnect.
This commit is contained in:
parent
3d98105853
commit
1b12a1bdd8
1 changed files with 8 additions and 1 deletions
9
agent.go
9
agent.go
|
|
@ -100,6 +100,11 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
|
|
||||||
slog.Info("connected", "agent_id", agentID)
|
slog.Info("connected", "agent_id", agentID)
|
||||||
|
|
||||||
|
// Session-scoped context: cancelled on write/read errors so blocked
|
||||||
|
// pool submits unblock immediately instead of waiting for workers to finish.
|
||||||
|
sessionCtx, sessionCancel := context.WithCancel(ctx)
|
||||||
|
defer sessionCancel()
|
||||||
|
|
||||||
// Channel for serializing WebSocket writes
|
// Channel for serializing WebSocket writes
|
||||||
writeCh := make(chan []byte, 10000)
|
writeCh := make(chan []byte, 10000)
|
||||||
|
|
||||||
|
|
@ -180,6 +185,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
data, _, err := ws.ReadMessage()
|
data, _, err := ws.ReadMessage()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
errCh <- err
|
errCh <- err
|
||||||
|
sessionCancel() // Unblock any stuck pool submits
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
msgCh <- data
|
msgCh <- data
|
||||||
|
|
@ -236,6 +242,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
case writeErrCh <- err:
|
case writeErrCh <- err:
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
|
sessionCancel() // Unblock any stuck pool submits
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -301,7 +308,7 @@ func runSession(ctx context.Context, baseURL, token string) error {
|
||||||
slog.Warn("invalid message", "error", err)
|
slog.Warn("invalid message", "error", err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if handleMessage(ctx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh, lldpTopologyResultCh) {
|
if handleMessage(sessionCtx, msg, pools, snmpResultCh, mikrotikResultCh, credTestResultCh, monitoringCheckCh, checkResultCh, lldpTopologyResultCh) {
|
||||||
flushSnmpBatch()
|
flushSnmpBatch()
|
||||||
return errRestartRequested
|
return errRestartRequested
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue