diff --git a/server/goals.go b/server/goals.go index 32c9cd4..9421e1f 100644 --- a/server/goals.go +++ b/server/goals.go @@ -121,6 +121,13 @@ func (s *Server) admitTaskWhenScheduled(t *Task, mode string, requirePaused, sch // a user while the probe was in flight. preserveSchedule keeps calendar state // and never creates schedule_manual_until (used by proxy recovery). func (s *Server) admitTaskWhenScheduledWithReason(t *Task, mode string, requirePaused, scheduled bool, requiredReason string, preserveSchedule bool) (queued bool, err error) { + return s.admitTaskWhenScheduledWithGeneration(t, mode, requirePaused, scheduled, requiredReason, preserveSchedule, 0) +} + +// admitTaskWhenScheduledWithGeneration extends the proxy recovery admission +// gate with a reconnect-round check. A delayed probe must not admit a task from +// a newer proxy pause round that happens to have the same attempt number. +func (s *Server) admitTaskWhenScheduledWithGeneration(t *Task, mode string, requirePaused, scheduled bool, requiredReason string, preserveSchedule bool, requiredGeneration uint64) (queued bool, err error) { if t == nil { return false, fmt.Errorf("task not found") } @@ -167,6 +174,9 @@ func (s *Server) admitTaskWhenScheduledWithReason(t *Task, mode string, requireP if requiredReason != "" && lifecycle.PauseReason != requiredReason { return false, fmt.Errorf("任务暂停原因已改变") } + if requiredReason == "paused_by_proxy" && requiredGeneration != 0 && lifecycle.ReconnectGeneration != requiredGeneration { + return false, fmt.Errorf("任务代理重连轮次已改变") + } mode = s.resumeAdmissionMode(t) } var manualUntil *time.Time diff --git a/server/manager.go b/server/manager.go index 7342c54..973fe03 100644 --- a/server/manager.go +++ b/server/manager.go @@ -28,25 +28,26 @@ import ( // sharing the process-wide asset store. ID is the PG task id as a string; ExpID // is the exploration the task owns. type Task struct { - ID string `json:"id"` - ExpID int64 `json:"exploration_id"` - Remark string `json:"remark"` - Name string `json:"name"` // 可选任务名称;空=未命名 - CategoryID *int64 `json:"category_id,omitempty"` - CategoryName string `json:"category_name,omitempty"` - PinnedAt int64 `json:"pinned_at,omitempty"` - Description string `json:"description"` - Goal string `json:"goal"` - CreatedAt int64 `json:"created_at"` - CompletedAt int64 `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成 - Paused bool `json:"paused"` - PauseReason string `json:"pause_reason,omitempty"` - ReconnectAttempt int `json:"-"` - ReconnectMax int `json:"-"` - ReconnectNextAt int64 `json:"-"` - ReconnectExhausted bool `json:"-"` - ReconnectLastError string `json:"-"` - Queued bool `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑 + ID string `json:"id"` + ExpID int64 `json:"exploration_id"` + Remark string `json:"remark"` + Name string `json:"name"` // 可选任务名称;空=未命名 + CategoryID *int64 `json:"category_id,omitempty"` + CategoryName string `json:"category_name,omitempty"` + PinnedAt int64 `json:"pinned_at,omitempty"` + Description string `json:"description"` + Goal string `json:"goal"` + CreatedAt int64 `json:"created_at"` + CompletedAt int64 `json:"completed_at,omitempty"` // 进入终态的 unix 秒;0=未完成 + Paused bool `json:"paused"` + PauseReason string `json:"pause_reason,omitempty"` + ReconnectAttempt int `json:"-"` + ReconnectMax int `json:"-"` + ReconnectNextAt int64 `json:"-"` + ReconnectExhausted bool `json:"-"` + ReconnectLastError string `json:"-"` + ReconnectGeneration uint64 `json:"-"` + Queued bool `json:"queued"` // 因并发上限被挂起、等待空位自动启动;true=尚未开跑 // QueuedAt is an internal Unix-nanosecond ordering key. It is deliberately // finer than CreatedAt so several tasks enqueued in the same second retain // their real FIFO order. @@ -100,6 +101,7 @@ type taskLifecycleState struct { ReconnectNextAt int64 ReconnectExhausted bool ReconnectLastError string + ReconnectGeneration uint64 Queued bool QueuedAt int64 QueueMode string @@ -131,7 +133,7 @@ func (t *Task) lifecycleSnapshotLocked() taskLifecycleState { Status: t.Status, Paused: t.Paused, PauseReason: t.PauseReason, - ReconnectAttempt: t.ReconnectAttempt, ReconnectMax: t.ReconnectMax, ReconnectNextAt: t.ReconnectNextAt, ReconnectExhausted: t.ReconnectExhausted, ReconnectLastError: t.ReconnectLastError, + ReconnectAttempt: t.ReconnectAttempt, ReconnectMax: t.ReconnectMax, ReconnectNextAt: t.ReconnectNextAt, ReconnectExhausted: t.ReconnectExhausted, ReconnectLastError: t.ReconnectLastError, ReconnectGeneration: t.ReconnectGeneration, Queued: t.Queued, QueuedAt: t.QueuedAt, QueueMode: t.QueueMode, @@ -165,6 +167,7 @@ func (t *Task) updateLifecycle(update func(*taskLifecycleState)) { t.ReconnectNextAt = state.ReconnectNextAt t.ReconnectExhausted = state.ReconnectExhausted t.ReconnectLastError = state.ReconnectLastError + t.ReconnectGeneration = state.ReconnectGeneration t.Queued = state.Queued t.QueuedAt = state.QueuedAt t.QueueMode = state.QueueMode @@ -1512,6 +1515,11 @@ func (m *Manager) applyTaskAdmissionWithSchedule(id, expectedStatus, status stri if segmentStartedAt.Valid { state.RunSegmentStartedAt = segmentStartedAt.Time.Unix() } + state.ReconnectAttempt = 0 + state.ReconnectMax = 0 + state.ReconnectNextAt = 0 + state.ReconnectExhausted = false + state.ReconnectLastError = "" }) } m.mu.Unlock() diff --git a/server/proxy_autoresume.go b/server/proxy_autoresume.go index a6ee6ef..d848054 100644 --- a/server/proxy_autoresume.go +++ b/server/proxy_autoresume.go @@ -94,6 +94,13 @@ func (s *Server) proxyReconnectConfig() (int, time.Duration) { return max, time.Duration(interval) * time.Second } +func (s *Server) runProxyProbe(ctx context.Context, proxyRaw, caPEMPath string, probes []string, timeout time.Duration) egressProbeResult { + if s != nil && s.proxyProbe != nil { + return s.proxyProbe(ctx, proxyRaw, caPEMPath, probes, timeout) + } + return probeEgress(ctx, proxyRaw, caPEMPath, probes, timeout) +} + func (s *Server) handleProxyFailure(taskID string, cause error) { if strings.TrimSpace(s.m.GlobalProxy()) == "" { return @@ -108,10 +115,26 @@ func (s *Server) pauseProxyTask(taskID string, cause error) { if strings.TrimSpace(s.m.GlobalProxy()) == "" { return } + max, interval := s.proxyReconnectConfig() + now := time.Now().Add(interval).Unix() + lastError := truncateProxyError(cause) transitioned, err := s.m.ApplyProxyPauseWithCallback(taskID, func() { if s.engine != nil { s.engine.Pause(taskID, agent.AbortPausedByProxy) } + if t, ok := s.m.Task(taskID); ok { + t.updateLifecycle(func(st *taskLifecycleState) { + // A successful admission clears the prior round's counters. This + // transition starts a fresh round, so make that reset atomic with + // the persisted proxy pause and install a new generation. + st.ReconnectGeneration++ + st.ReconnectAttempt = 0 + st.ReconnectMax = max + st.ReconnectNextAt = now + st.ReconnectExhausted = false + st.ReconnectLastError = lastError + }) + } }) if err != nil { log.Printf("[proxy] pause task %s: %v", taskID, err) @@ -124,19 +147,6 @@ func (s *Server) pauseProxyTask(taskID string, cause error) { return } s.cancelTaskChat(taskID, agent.AbortChatPausedWithTask) - max, interval := s.proxyReconnectConfig() - now := time.Now().Add(interval).Unix() - 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 - st.ReconnectLastError = truncateProxyError(cause) - }) - } s.ensureProxySupervisor() s.signalProxySupervisor() } @@ -218,32 +228,47 @@ func (s *Server) proxyReconnectTick() { if st.PauseReason != "paused_by_proxy" || st.ReconnectExhausted || (st.ReconnectNextAt > 0 && st.ReconnectNextAt > now) { continue } - st.ReconnectAttempt++ - st.ReconnectMax = max + attempt := st.ReconnectAttempt + 1 + round := st.ReconnectGeneration + claimed := false t.updateLifecycle(func(v *taskLifecycleState) { - v.ReconnectAttempt = st.ReconnectAttempt + if v.PauseReason != "paused_by_proxy" || v.ReconnectGeneration != round || v.ReconnectAttempt != st.ReconnectAttempt { + return + } + v.ReconnectAttempt = attempt v.ReconnectMax = max v.ReconnectNextAt = 0 + claimed = true }) - res := probeEgress(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second) + if !claimed { + continue + } + res := s.runProxyProbe(context.Background(), s.m.GlobalProxy(), "", defaultEgressProbes, 10*time.Second) if res.OK { - s.proxyReconnectSuccess() + s.proxyReconnectSuccessFor(t.ID, round, attempt) return } err := res.Error - exhausted := st.ReconnectAttempt >= max + exhausted := attempt >= max next := time.Now().Add(interval).Unix() t.updateLifecycle(func(v *taskLifecycleState) { + if v.PauseReason != "paused_by_proxy" || v.ReconnectGeneration != round || v.ReconnectAttempt != attempt { + return + } v.ReconnectNextAt = next v.ReconnectExhausted = exhausted v.ReconnectLastError = err }) - log.Printf("[proxy] reconnect attempt %d/%d failed: %s", st.ReconnectAttempt, max, err) + log.Printf("[proxy] reconnect attempt %d/%d failed: %s", attempt, max, err) } } func (s *Server) proxyReconnectSuccess() { - s.proxyReconnectSuccessWithHook(nil) + s.proxyReconnectSuccessWithGuard("", 0, 0, nil) +} + +func (s *Server) proxyReconnectSuccessFor(taskID string, generation uint64, attempt int) { + s.proxyReconnectSuccessWithGuard(taskID, generation, attempt, nil) } // proxyReconnectSuccessWithHook is kept small and private so tests can @@ -251,6 +276,10 @@ func (s *Server) proxyReconnectSuccess() { // released concMu. Production callers use proxyReconnectSuccess without a // hook. func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) { + s.proxyReconnectSuccessWithGuard("", 0, 0, afterAdmission) +} + +func (s *Server) proxyReconnectSuccessWithGuard(expectedID string, expectedGeneration uint64, expectedAttempt int, afterAdmission func()) { max, interval := s.proxyReconnectConfig() s.scheduleMu.Lock() defer s.scheduleMu.Unlock() @@ -258,6 +287,12 @@ func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) { if st := t.lifecycleSnapshot(); st.PauseReason != "paused_by_proxy" { continue } + if expectedID == t.ID { + st := t.lifecycleSnapshot() + if st.PauseReason != "paused_by_proxy" || st.ReconnectGeneration != expectedGeneration || st.ReconnectAttempt != expectedAttempt { + continue + } + } if s.m.pg != nil { inside, scheduleErr := s.proxyResumeInsideSchedule(t.ID, time.Now()) if scheduleErr != nil { @@ -273,7 +308,11 @@ func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) { } var err error if s.engine != nil { - _, err = s.admitTaskWhenScheduledWithReason(t, "resume", true, false, "paused_by_proxy", true) + if expectedID == t.ID { + _, err = s.admitTaskWhenScheduledWithGeneration(t, "resume", true, false, "paused_by_proxy", true, expectedGeneration) + } else { + _, err = s.admitTaskWhenScheduledWithReason(t, "resume", true, false, "paused_by_proxy", true) + } if afterAdmission != nil { afterAdmission() } @@ -285,7 +324,8 @@ func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) { // terminal transition, delete, or archive therefore cannot be revived; // only keep probing if the proxy pause is still eligible. st := t.lifecycleSnapshot() - if st.PauseReason == "paused_by_proxy" && !isTerminalStatus(st.Status) { + if st.PauseReason == "paused_by_proxy" && !isTerminalStatus(st.Status) && + (expectedID == "" || (st.ReconnectGeneration == expectedGeneration && st.ReconnectAttempt == expectedAttempt)) { t.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectMax = max v.ReconnectNextAt = time.Now().Add(interval).Unix() @@ -299,6 +339,9 @@ func (s *Server) proxyReconnectSuccessWithHook(afterAdmission func()) { // a concurrent user/orchestrator pause could have installed a newer // barrier in the meantime. t.updateLifecycle(func(v *taskLifecycleState) { + if expectedID == t.ID && v.ReconnectGeneration != expectedGeneration { + return + } v.ReconnectAttempt = 0 v.ReconnectMax = 0 v.ReconnectNextAt = 0 diff --git a/server/proxy_autoresume_test.go b/server/proxy_autoresume_test.go index 67683a5..6bd0864 100644 --- a/server/proxy_autoresume_test.go +++ b/server/proxy_autoresume_test.go @@ -323,3 +323,105 @@ func TestProxyPauseDoesNotReviveExhaustedOrNonProxyTasks(t *testing.T) { t.Fatalf("new proxy task did not initialize reconnect state: %+v", got) } } + +func TestProxyReconnectRoundResetsAfterManualResume(t *testing.T) { + m := proxyTestManager(t) + task, err := m.CreateTask("proxy retry round", "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.ApplyProxyPause(task.ID); err != nil { + t.Fatal(err) + } + s.engine.Pause(task.ID, agent.AbortPausedByProxy) + task.updateLifecycle(func(v *taskLifecycleState) { + v.ReconnectAttempt, v.ReconnectMax, v.ReconnectNextAt = 5, 5, 12345 + v.ReconnectExhausted, v.ReconnectLastError = true, "proxy exhausted" + }) + if _, err := s.applyTaskControl(task, "resume"); err != nil { + t.Fatal(err) + } + resumed := task.lifecycleSnapshot() + if resumed.PauseReason != "" || resumed.ReconnectAttempt != 0 || resumed.ReconnectMax != 0 || resumed.ReconnectNextAt != 0 || resumed.ReconnectExhausted || resumed.ReconnectLastError != "" { + t.Fatalf("manual resume retained reconnect state: %+v", resumed) + } + s.pauseProxyTask(task.ID, errors.New("proxy failed again")) + paused := task.lifecycleSnapshot() + if paused.PauseReason != "paused_by_proxy" || paused.ReconnectAttempt != 0 || paused.ReconnectExhausted { + t.Fatalf("new proxy round was not reset: %+v", paused) + } + if paused.ReconnectMax == 0 || paused.ReconnectNextAt == 0 || paused.ReconnectLastError == "" { + t.Fatalf("new proxy round did not initialize reconnect state: %+v", paused) + } +} + +func TestProxyReconnectDropsStaleProbeFailureAfterNewRound(t *testing.T) { + m := proxyTestManager(t) + task, err := m.CreateTask("proxy stale probe", "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" + oldMax, hadMax, err := m.pg.GetSetting(settingProxyReconnectMax) + if err != nil { + t.Fatal(err) + } + defer func() { + if hadMax { + _ = m.pg.SetSetting(settingProxyReconnectMax, oldMax) + } else { + _, _ = m.pg.Exec(`DELETE FROM settings WHERE key=$1`, settingProxyReconnectMax) + } + }() + if err := m.pg.SetSetting(settingProxyReconnectMax, "1"); err != nil { + t.Fatal(err) + } + s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult { + return egressProbeResult{Error: "initial probe failure"} + } + s.pauseProxyTask(task.ID, errors.New("initial proxy failure")) + task.updateLifecycle(func(v *taskLifecycleState) { v.ReconnectNextAt = 0 }) + oldRound := task.lifecycleSnapshot().ReconnectGeneration + probeStarted := make(chan struct{}) + releaseProbe := make(chan struct{}) + s.proxyProbe = func(context.Context, string, string, []string, time.Duration) egressProbeResult { + close(probeStarted) + <-releaseProbe + return egressProbeResult{Error: "stale probe failure"} + } + tickDone := make(chan struct{}) + go func() { + s.proxyReconnectTick() + close(tickDone) + }() + select { + case <-probeStarted: + case <-time.After(time.Second): + t.Fatal("reconnect tick did not start its probe") + } + if _, err := s.applyTaskControl(task, "resume"); err != nil { + t.Fatal(err) + } + s.pauseProxyTask(task.ID, errors.New("new proxy failure")) + newRound := task.lifecycleSnapshot() + if newRound.ReconnectGeneration == oldRound || newRound.ReconnectAttempt != 0 || newRound.ReconnectExhausted { + t.Fatalf("new proxy round was not established: old=%d new=%+v", oldRound, newRound) + } + close(releaseProbe) + select { + case <-tickDone: + case <-time.After(time.Second): + t.Fatal("reconnect tick did not finish") + } + after := task.lifecycleSnapshot() + if after.PauseReason != "paused_by_proxy" || after.ReconnectGeneration != newRound.ReconnectGeneration || after.ReconnectAttempt != 0 || after.ReconnectExhausted || after.ReconnectNextAt != newRound.ReconnectNextAt || after.ReconnectLastError != newRound.ReconnectLastError { + t.Fatalf("stale probe mutated the new round: before=%+v after=%+v", newRound, after) + } +} diff --git a/server/server.go b/server/server.go index 8b69043..9807c38 100644 --- a/server/server.go +++ b/server/server.go @@ -129,7 +129,10 @@ type Server struct { archiveWG sync.WaitGroup credentialWake chan struct{} proxyReconnect proxyReconnectSupervisor - side *sideQuestionState + // proxyProbe is injectable for deterministic reconnect race tests. Production + // servers leave it nil and use probeEgress directly. + proxyProbe func(context.Context, string, string, []string, time.Duration) egressProbeResult + side *sideQuestionState } // provEntry is a cached provider + its config for one LLM profile id.