diff --git a/pkg/jobcontroller/jobcontroller.go b/pkg/jobcontroller/jobcontroller.go index e41d77585c..bf82e76e1f 100644 --- a/pkg/jobcontroller/jobcontroller.go +++ b/pkg/jobcontroller/jobcontroller.go @@ -71,6 +71,21 @@ const JobOutputFileName = "term" const AutoReconnectDelay = 1 * time.Second const AutoReconnectCooldown = 30 * time.Second +// ReconnectJobTimeout bounds a single job reconnect. ReconnectJobConcurrency +// bounds how many run at once when a connection comes back up. +const ReconnectJobTimeout = 60 * time.Second +const ReconnectJobConcurrency = 8 + +// RemoteJobRpcTimeoutMs bounds a single RPC to a remote connection's job +// machinery, and ReconnectRouteWaitTimeout bounds waiting for the job route +// to register after a successful reconnect. Each reconnect is several +// round trips (reconnect -> socket connect -> job manager auth -> token +// verify back to wavesrv), so on a 300ms+ RTT link with packet loss the +// previous 5s/2s budgets were routinely exceeded and the job was marked +// failed even though the job manager was healthy. +const RemoteJobRpcTimeoutMs = 20000 +const ReconnectRouteWaitTimeout = 10 * time.Second + type connState struct { actual bool processed bool @@ -410,7 +425,7 @@ func attemptAutoReconnect(jobId string, connName string) { } log.Printf("[job:%s] connection %s still up after route down, attempting auto-reconnect to determine job manager status", jobId, connName) - ctx, cancelFn := context.WithTimeout(context.Background(), 10*time.Second) + ctx, cancelFn := context.WithTimeout(context.Background(), ReconnectJobTimeout) defer cancelFn() err = ReconnectJob(ctx, jobId, nil) if err != nil { @@ -481,6 +496,10 @@ func handleBlockCloseEvent(event *wps.WaveEvent) { } } +// onConnectionUp reconnects every durable job attached to connName once its +// connection has been re-established. Each job is reconnected under its own +// deadline, and the batch runs with bounded concurrency, so a slow or +// unreachable job cannot consume the time budget of the jobs behind it. func onConnectionUp(connName string) { log.Printf("[conn:%s] connection became connected, reconnecting jobs", connName) ctx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second) @@ -501,15 +520,37 @@ func onConnectionUp(connName string) { log.Printf("[conn:%s] found %d jobs to reconnect", connName, len(jobsToReconnect)) + // Reconnect each job under its own deadline, with bounded concurrency. + // Previously every ReconnectJob call shared the single 5s context created + // above and ran serially, so on a high-latency connection only the first + // few jobs finished before the deadline elapsed and every remaining job + // failed at once with "context deadline exceeded", leaving those blocks + // permanently stalled until the connection was torn down by hand. + var wg sync.WaitGroup + var successMu sync.Mutex successCount := 0 + sem := make(chan struct{}, ReconnectJobConcurrency) for _, job := range jobsToReconnect { - err = ReconnectJob(ctx, job.OID, nil) - if err != nil { - log.Printf("[job:%s] error reconnecting: %v", job.OID, err) - } else { + wg.Add(1) + go func(job *waveobj.Job) { + defer func() { + panichandler.PanicHandler("onConnectionUp:ReconnectJob", recover()) + wg.Done() + }() + sem <- struct{}{} + defer func() { <-sem }() + jobCtx, jobCancelFn := context.WithTimeout(context.Background(), ReconnectJobTimeout) + defer jobCancelFn() + if rerr := ReconnectJob(jobCtx, job.OID, nil); rerr != nil { + log.Printf("[job:%s] error reconnecting: %v", job.OID, rerr) + return + } + successMu.Lock() successCount++ - } + successMu.Unlock() + }(job) } + wg.Wait() log.Printf("[conn:%s] finished reconnecting jobs: %d/%d successful", connName, successCount, len(jobsToReconnect)) } @@ -970,7 +1011,7 @@ func DisconnectJob(ctx context.Context, jobId string) error { bareRpc := wshclient.GetBareRpcClient() rpcOpts := &wshrpc.RpcOpts{ Route: wshutil.MakeConnectionRouteId(job.Connection), - Timeout: 5000, + Timeout: RemoteJobRpcTimeoutMs, } disconnectData := wshrpc.CommandRemoteDisconnectFromJobManagerData{ @@ -1011,7 +1052,7 @@ func remoteTerminateJobManager(ctx context.Context, job *waveobj.Job) error { rpcOpts := &wshrpc.RpcOpts{ Route: wshutil.MakeConnectionRouteId(job.Connection), - Timeout: 5000, + Timeout: RemoteJobRpcTimeoutMs, } err := wshclient.RemoteTerminateJobManagerCommand(bareRpc, terminateData, rpcOpts) @@ -1108,7 +1149,7 @@ func doReconnectJob(ctx context.Context, jobId string, rtOpts *waveobj.RuntimeOp rpcOpts := &wshrpc.RpcOpts{ Route: wshutil.MakeConnectionRouteId(job.Connection), - Timeout: 5000, + Timeout: RemoteJobRpcTimeoutMs, } log.Printf("[job:%s] sending RemoteReconnectToJobManagerCommand to connection %s", jobId, job.Connection) @@ -1148,7 +1189,7 @@ func doReconnectJob(ctx context.Context, jobId string, rtOpts *waveobj.RuntimeOp log.Printf("[job:%s] RemoteReconnectToJobManagerCommand succeeded, waiting for route", jobId) routeId := wshutil.MakeJobRouteId(jobId) - waitCtx, cancelFn := context.WithTimeout(ctx, 2*time.Second) + waitCtx, cancelFn := context.WithTimeout(ctx, ReconnectRouteWaitTimeout) defer cancelFn() err = wshutil.DefaultRouter.WaitForRegister(waitCtx, routeId) if err != nil { @@ -1257,7 +1298,7 @@ func restartStreaming(ctx context.Context, jobId string, knownConnected bool, rt rpcOpts := &wshrpc.RpcOpts{ Route: wshutil.MakeJobRouteId(jobId), - Timeout: 5000, + Timeout: RemoteJobRpcTimeoutMs, } log.Printf("[job:%s] sending JobPrepareConnectCommand with seq=%d (fileSize=%d, totalGap=%d)", jobId, currentSeq, waveFile.Size, totalGap) diff --git a/pkg/remote/conncontroller/connmonitor.go b/pkg/remote/conncontroller/connmonitor.go index dc14561484..0bf127ca93 100644 --- a/pkg/remote/conncontroller/connmonitor.go +++ b/pkg/remote/conncontroller/connmonitor.go @@ -25,11 +25,46 @@ type ConnMonitor struct { LastInputTime atomic.Int64 KeepAliveSentTime atomic.Int64 KeepAliveInFlight bool + RttEwmaMs atomic.Int64 // smoothed keepalive round-trip time, 0 until first sample ctx context.Context cancelFunc context.CancelFunc inputNotifyCh chan int64 } +// Health thresholds scale with the measured keepalive RTT so that a slow but +// working link (e.g. a 300ms+ cross-continent path with some packet loss) is +// not constantly flagged as degraded/stalled. The floors match or exceed the +// original fixed values, so low-latency links behave as before. +const ( + degradedDelayFloorMs = 2000 + degradedDelayRttFactor = 4 + stalledFloorMs = 15000 + stalledUrgentFloorMs = 8000 + stalledRttFactor = 10 + rttEwmaWeightPct = 25 // weight of a new sample, in percent +) + +// healthThresholds returns how long to wait for a response after user input +// before reporting "degraded", and how long a keepalive may be in flight +// before reporting "stalled", given the smoothed RTT in milliseconds. +func healthThresholds(rttMs int64, urgent bool) (degradedDelayMs int64, stalledMs int64) { + degradedDelayMs = max(int64(degradedDelayFloorMs), degradedDelayRttFactor*rttMs) + stalledFloor := int64(stalledFloorMs) + if urgent { + stalledFloor = stalledUrgentFloorMs + } + stalledMs = max(stalledFloor, stalledRttFactor*rttMs) + return degradedDelayMs, stalledMs +} + +// updateRttEwma folds a new RTT sample into the running average. +func updateRttEwma(prevMs int64, sampleMs int64) int64 { + if prevMs <= 0 { + return sampleMs + } + return (prevMs*(100-rttEwmaWeightPct) + sampleMs*rttEwmaWeightPct) / 100 +} + func MakeConnMonitor(conn *SSHConn, client *ssh.Client) *ConnMonitor { if conn == nil { panic("conn cannot be nil") @@ -125,6 +160,7 @@ func (cm *ConnMonitor) SendKeepAlive() error { log.Printf("[conncontroller] conn:%s keepalive error (duration=%dms): %v", cm.Conn.GetName(), duration, err) return } + cm.RttEwmaMs.Store(updateRttEwma(cm.RttEwmaMs.Load(), time.Since(startTime).Milliseconds())) cm.UpdateLastActivityTime() }() return nil @@ -146,10 +182,7 @@ func (cm *ConnMonitor) checkConnection() { cm.SendKeepAlive() } - stalledThreshold := int64(10000) - if urgent { - stalledThreshold = 5000 - } + _, stalledThreshold := healthThresholds(cm.RttEwmaMs.Load(), urgent) timeSinceKeepAlive := cm.getTimeSinceKeepAlive() if timeSinceKeepAlive > stalledThreshold { cm.setConnHealthStatus(ConnHealthStatus_Stalled) @@ -174,8 +207,9 @@ func (cm *ConnMonitor) keepAliveMonitor() { cm.checkConnection() case inputTime := <-cm.inputNotifyCh: + degradedDelayMs, _ := healthThresholds(cm.RttEwmaMs.Load(), true) select { - case <-time.After(1 * time.Second): + case <-time.After(time.Duration(degradedDelayMs) * time.Millisecond): if cm.LastActivityTime.Load() >= inputTime { break } diff --git a/pkg/remote/conncontroller/connmonitor_test.go b/pkg/remote/conncontroller/connmonitor_test.go new file mode 100644 index 0000000000..f3a3719c1d --- /dev/null +++ b/pkg/remote/conncontroller/connmonitor_test.go @@ -0,0 +1,54 @@ +// Copyright 2026, Command Line Inc. +// SPDX-License-Identifier: Apache-2.0 + +package conncontroller + +import "testing" + +func TestHealthThresholds(t *testing.T) { + tests := []struct { + name string + rttMs int64 + urgent bool + wantDegraded int64 + wantStalled int64 + }{ + {"no sample yet, idle", 0, false, 2000, 15000}, + {"no sample yet, typing", 0, true, 2000, 8000}, + {"lan", 5, true, 2000, 8000}, + {"floor boundary", 500, false, 2000, 15000}, + {"cross-pacific typing", 700, true, 2800, 8000}, + {"very slow link", 2000, true, 8000, 20000}, + {"very slow link idle", 2000, false, 8000, 20000}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + gotDegraded, gotStalled := healthThresholds(tc.rttMs, tc.urgent) + if gotDegraded != tc.wantDegraded || gotStalled != tc.wantStalled { + t.Fatalf("healthThresholds(%d, %v) = (%d, %d), want (%d, %d)", + tc.rttMs, tc.urgent, gotDegraded, gotStalled, tc.wantDegraded, tc.wantStalled) + } + }) + } +} + +func TestUpdateRttEwma(t *testing.T) { + if got := updateRttEwma(0, 300); got != 300 { + t.Fatalf("first sample: got %d, want 300", got) + } + if got := updateRttEwma(-1, 300); got != 300 { + t.Fatalf("negative prev treated as empty: got %d, want 300", got) + } + // 75% of 400 + 25% of 800 = 500 + if got := updateRttEwma(400, 800); got != 500 { + t.Fatalf("got %d, want 500", got) + } + // converges toward a steady sample + v := int64(100) + for i := 0; i < 50; i++ { + v = updateRttEwma(v, 600) + } + if v < 590 || v > 600 { + t.Fatalf("did not converge: got %d", v) + } +} diff --git a/pkg/wshutil/wshrpc.go b/pkg/wshutil/wshrpc.go index 8a29e204bd..574c12b709 100644 --- a/pkg/wshutil/wshrpc.go +++ b/pkg/wshutil/wshrpc.go @@ -25,7 +25,7 @@ import ( "github.com/wavetermdev/waveterm/pkg/wshrpc" ) -const DefaultTimeoutMs = 5000 +const DefaultTimeoutMs = 15000 const RespChSize = 32 const DefaultMessageChSize = 32 const CtxDoneChSize = 10