From 97a4f2b236fcf394129afe5691a0e762ac9f29d4 Mon Sep 17 00:00:00 2001 From: RuoJi6 Date: Sat, 31 Oct 2026 18:23:40 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BB=A3=E7=90=86=E6=9A=82=E5=81=9C/?= =?UTF-8?q?=E6=81=A2=E5=A4=8D=E5=B1=8F=E9=9A=9C=E5=9C=A8=E9=94=81=E5=86=85?= =?UTF-8?q?=E5=AE=89=E8=A3=85=E5=B9=B6=E9=98=B2=E6=AD=A2=E5=A4=8D=E6=B4=BB?= =?UTF-8?q?=E5=B7=B2=E8=80=97=E5=B0=BD=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/manager.go | 23 ++++-- server/proxy_autoresume.go | 48 +++++++----- server/proxy_autoresume_test.go | 131 ++++++++++++++++++++++++++++++++ server/task_control.go | 18 ++++- 4 files changed, 196 insertions(+), 24 deletions(-) diff --git a/server/manager.go b/server/manager.go index 30d4ec2..7342c54 100644 --- a/server/manager.go +++ b/server/manager.go @@ -1580,6 +1580,16 @@ func (m *Manager) ApplyTaskPauseOriginReason(id, reason string, scheduled bool) // ApplyProxyPause atomically pauses a running task and marks it eligible for // reconnect auto-resume. Already-paused tasks retain their original reason. func (m *Manager) ApplyProxyPause(id string) error { + _, err := m.ApplyProxyPauseWithCallback(id, nil) + return err +} + +// ApplyProxyPauseWithCallback is the locked proxy-pause transition. When the +// task actually changes from running to paused, callback runs while the +// Server's task-control mutex is still held, after the DB and live lifecycle +// state have committed. This lets callers install the Engine barrier before a +// concurrent user/orchestrator pause can observe the new paused state. +func (m *Manager) ApplyProxyPauseWithCallback(id string, callback func()) (transitioned bool, err error) { if m.controlMu != nil { m.controlMu.Lock() defer m.controlMu.Unlock() @@ -1588,7 +1598,7 @@ func (m *Manager) ApplyProxyPause(id string) error { defer m.taskStateMu.Unlock() n, err := strconv.ParseInt(id, 10, 64) if err != nil { - return err + return false, err } var mode string var active int64 @@ -1598,10 +1608,10 @@ func (m *Manager) ApplyProxyPause(id string) error { run_segment_started_at=NULL WHERE id=$1 AND deleted_at IS NULL AND paused=false AND status NOT IN ('done','failed','timeout') RETURNING COALESCE(queue_mode,''), active_run_seconds, run_segment_started_at`, n).Scan(&mode, &active, &started) if errors.Is(err, sql.ErrNoRows) { - return nil + return false, nil } if err != nil { - return err + return false, err } m.mu.RLock() t := m.tasks[id] @@ -1617,7 +1627,10 @@ func (m *Manager) ApplyProxyPause(id string) error { s.RunSegmentStartedAt = 0 }) } - return nil + if callback != nil { + callback() + } + return true, nil } // ClearPauseReason removes auto-resume eligibility without changing lifecycle. @@ -1633,7 +1646,7 @@ func (m *Manager) ClearPauseReason(id string) error { s.PauseReason = "" s.ReconnectAttempt, s.ReconnectMax, s.ReconnectNextAt = 0, 0, 0 s.ReconnectExhausted, s.ReconnectLastError = false, "" - }) + }) } return nil } diff --git a/server/proxy_autoresume.go b/server/proxy_autoresume.go index f4f33ff..a6ee6ef 100644 --- a/server/proxy_autoresume.go +++ b/server/proxy_autoresume.go @@ -108,24 +108,29 @@ func (s *Server) pauseProxyTask(taskID string, cause error) { if strings.TrimSpace(s.m.GlobalProxy()) == "" { return } - err := s.m.ApplyProxyPause(taskID) + transitioned, err := s.m.ApplyProxyPauseWithCallback(taskID, func() { + if s.engine != nil { + s.engine.Pause(taskID, agent.AbortPausedByProxy) + } + }) if err != nil { log.Printf("[proxy] pause task %s: %v", taskID, err) return } - if s.engine != nil { - if t, ok := s.m.Task(taskID); ok && t.lifecycleSnapshot().PauseReason == "paused_by_proxy" { - s.engine.Pause(taskID, agent.AbortPausedByProxy) - } + if !transitioned { + // A no-op means the task was already paused (possibly by a user, + // schedule, or an exhausted proxy attempt). Preserve its reason and + // reconnect state exactly as-is. + return } s.cancelTaskChat(taskID, agent.AbortChatPausedWithTask) max, interval := s.proxyReconnectConfig() now := time.Now().Add(interval).Unix() - for _, t := range s.m.List() { - if t.ID != taskID && t.lifecycleSnapshot().PauseReason != "paused_by_proxy" { - continue - } + if t, ok := s.m.Task(taskID); ok { t.updateLifecycle(func(st *taskLifecycleState) { + // Only the task that transitioned in this call starts a new + // reconnect round. A task that exhausted a prior round remains + // untouched until it is manually resumed and paused again. st.ReconnectMax = max st.ReconnectNextAt = now st.ReconnectExhausted = false @@ -238,6 +243,14 @@ func (s *Server) proxyReconnectTick() { } func (s *Server) proxyReconnectSuccess() { + s.proxyReconnectSuccessWithHook(nil) +} + +// proxyReconnectSuccessWithHook is kept small and private so tests can +// deterministically place a competing pause immediately after admission has +// released concMu. Production callers use proxyReconnectSuccess without a +// hook. +func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) { max, interval := s.proxyReconnectConfig() s.scheduleMu.Lock() defer s.scheduleMu.Unlock() @@ -258,12 +271,12 @@ func (s *Server) proxyReconnectSuccess() { continue } } - var ( - queued bool - err error - ) + var err error if s.engine != nil { - queued, err = s.admitTaskWhenScheduledWithReason(t, "resume", true, false, "paused_by_proxy", true) + _, err = s.admitTaskWhenScheduledWithReason(t, "resume", true, false, "paused_by_proxy", true) + if afterAdmission != nil { + afterAdmission() + } } else { err = s.m.ClearPauseReason(t.ID) } @@ -281,9 +294,10 @@ func (s *Server) proxyReconnectSuccess() { log.Printf("[proxy] resume task %s: %v", t.ID, err) continue } - if s.engine != nil && !queued { - s.engine.Resume(t) - } + // The non-queued path's startAdmittedTask already resumes the Engine + // while concMu is held. Do not Resume again after releasing the lock: + // a concurrent user/orchestrator pause could have installed a newer + // barrier in the meantime. t.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectAttempt = 0 v.ReconnectMax = 0 diff --git a/server/proxy_autoresume_test.go b/server/proxy_autoresume_test.go index 43f1c6b..67683a5 100644 --- a/server/proxy_autoresume_test.go +++ b/server/proxy_autoresume_test.go @@ -192,3 +192,134 @@ func TestProxyAdmissionPreservesCalendarOverride(t *testing.T) { t.Fatalf("proxy admission changed calendar state: paused=%v until=%v", schedulePaused, manualUntil) } } + +func TestProxyReconnectSuccessKeepsCompetingUserPauseBarrier(t *testing.T) { + m := proxyTestManager(t) + task, err := m.CreateTask("proxy success race", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }() + s := newAdmissionTestServer(m, nil) + m.SetTaskControlMutex(&s.concMu) + m.globalProxy = "http://proxy.invalid" + if _, err := m.ApplyProxyPauseWithCallback(task.ID, func() { s.engine.Pause(task.ID, agent.AbortPausedByProxy) }); err != nil { + t.Fatal(err) + } + + var pauseErr error + s.proxyReconnectSuccessWithHook(func() { + _, pauseErr = s.applyTaskControl(task, "pause") + }) + if pauseErr != nil { + t.Fatalf("competing user pause failed: %v", pauseErr) + } + state := task.lifecycleSnapshot() + if state.PauseReason != "paused_by_user" || !state.Paused { + t.Fatalf("user pause was overwritten: %+v", state) + } + if !s.engine.IsPaused(task.ID) { + t.Fatal("user pause barrier was cleared by a post-admission Resume") + } +} + +func TestProxyPauseBarrierInstalledBeforeControlUnlock(t *testing.T) { + m := proxyTestManager(t) + task, err := m.CreateTask("proxy barrier", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { _, _ = m.DeleteTask(task.ID, DeleteTaskOptions{}) }() + s := newAdmissionTestServer(m, nil) + m.SetTaskControlMutex(&s.concMu) + m.globalProxy = "http://proxy.invalid" + + userDone := make(chan error, 1) + transitioned, err := m.ApplyProxyPauseWithCallback(task.ID, func() { + s.engine.Pause(task.ID, agent.AbortPausedByProxy) + if !s.engine.IsPaused(task.ID) { + t.Fatal("proxy barrier was not installed in the locked callback") + } + go func() { + _, pauseErr := s.applyTaskControl(task, "pause") + userDone <- pauseErr + }() + select { + case <-userDone: + t.Fatal("user pause acquired concMu before proxy callback returned") + case <-time.After(20 * time.Millisecond): + } + }) + if err != nil || !transitioned { + t.Fatalf("proxy transition=%v err=%v", transitioned, err) + } + if err := <-userDone; err != nil { + t.Fatalf("concurrent user pause failed: %v", err) + } + if !s.engine.IsPaused(task.ID) { + t.Fatal("concurrent user pause left the engine running") + } + if got := task.lifecycleSnapshot().PauseReason; got != "paused_by_user" { + t.Fatalf("concurrent user pause reason=%q", got) + } +} + +func TestProxyPauseDoesNotReviveExhaustedOrNonProxyTasks(t *testing.T) { + m := proxyTestManager(t) + exhausted, err := m.CreateTask("exhausted proxy", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + userPaused, err := m.CreateTask("user paused", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + other, err := m.CreateTask("new proxy failure", "goal", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + defer func() { + _, _ = m.DeleteTask(exhausted.ID, DeleteTaskOptions{}) + _, _ = m.DeleteTask(userPaused.ID, DeleteTaskOptions{}) + _, _ = m.DeleteTask(other.ID, DeleteTaskOptions{}) + }() + s := newAdmissionTestServer(m, nil) + m.SetTaskControlMutex(&s.concMu) + m.globalProxy = "http://proxy.invalid" + if err := m.ApplyProxyPause(exhausted.ID); err != nil { + t.Fatal(err) + } + if err := m.ApplyTaskPauseOriginReason(userPaused.ID, "paused_by_user", false); err != nil { + t.Fatal(err) + } + exhausted.updateLifecycle(func(v *taskLifecycleState) { + v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 5, 5, 12345 + v.ReconnectExhausted, v.ReconnectLastError = true, "proxy down" + }) + userPaused.updateLifecycle(func(v *taskLifecycleState) { + v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 3, 7, 23456 + v.ReconnectExhausted, v.ReconnectLastError = true, "manual pause state" + }) + exhaustedBefore := exhausted.lifecycleSnapshot() + userBefore := userPaused.lifecycleSnapshot() + + // A no-op proxy pause on an already paused task must not reset either state. + s.pauseProxyTask(exhausted.ID, errors.New("another proxy failure")) + s.pauseProxyTask(userPaused.ID, errors.New("proxy failure while user paused")) + if got := exhausted.lifecycleSnapshot(); got.ReconnectNextAt != exhaustedBefore.ReconnectNextAt || !got.ReconnectExhausted || got.ReconnectLastError != exhaustedBefore.ReconnectLastError { + t.Fatalf("exhausted task was revived: before=%+v after=%+v", exhaustedBefore, got) + } + if got := userPaused.lifecycleSnapshot(); got.PauseReason != "paused_by_user" || got.ReconnectNextAt != userBefore.ReconnectNextAt || !got.ReconnectExhausted { + t.Fatalf("user-paused task was mutated: before=%+v after=%+v", userBefore, got) + } + + // A genuinely new proxy pause of another running task initializes only that + // task's reconnect round. + s.pauseProxyTask(other.ID, errors.New("new proxy failure")) + if got := exhausted.lifecycleSnapshot(); got.ReconnectNextAt != exhaustedBefore.ReconnectNextAt || !got.ReconnectExhausted { + t.Fatalf("subsequent proxy pause revived exhausted task: %+v", got) + } + if got := other.lifecycleSnapshot(); got.PauseReason != "paused_by_proxy" || got.ReconnectMax == 0 || got.ReconnectExhausted { + t.Fatalf("new proxy task did not initialize reconnect state: %+v", got) + } +} diff --git a/server/task_control.go b/server/task_control.go index 060b1a4..005e2f8 100644 --- a/server/task_control.go +++ b/server/task_control.go @@ -116,17 +116,31 @@ func (s *Server) applyTaskControlWithCause(t *Task, action string, pauseCause er } if lifecycle.Paused { if pauseCause != agent.AbortPausedBySchedule { - if _, err := s.m.pg.Exec(`UPDATE tasks SET schedule_paused=false, pause_reason='paused_by_user' WHERE id=$1`, t.ID); err != nil { + if pauseCause == nil { + pauseCause = agent.AbortPausedByUser + } + reason := "paused_by_user" + if pauseCause == agent.AbortPausedByOrchestrator { + reason = "paused_by_orchestrator" + } + if pauseCause == agent.AbortPausedByProxy { + reason = "paused_by_proxy" + } + if _, err := s.m.pg.Exec(`UPDATE tasks SET schedule_paused=false, pause_reason=$2 WHERE id=$1`, t.ID, reason); err != nil { return out, err } t.updateLifecycle(func(state *taskLifecycleState) { - state.PauseReason = "paused_by_user" + state.PauseReason = reason state.ReconnectAttempt = 0 state.ReconnectMax = 0 state.ReconnectNextAt = 0 state.ReconnectExhausted = false state.ReconnectLastError = "" }) + // The lifecycle row may already have been paused by another + // actor. Reinstall the requested cause barrier while holding + // concMu instead of assuming the existing barrier is sufficient. + s.engine.Pause(t.ID, pauseCause) out.Paused = true out.Status = "paused" return out, nil