fix(runtime): don't drop a crash during recovery unwind (fixes CI hang)

changelog/unreleased/02b-runtime-recovery.md

@@ -0,0 +1,7 @@
+### English
+
+- **Fix** · A Qoder worker that crashes right after an automatic restart is now restarted again instead of staying dead.
+
+### 中文
+
+- **修复** · Qoder worker 在自动重启后立即再次崩溃时会继续重启，不再一直处于 dead 状态。

internal/runtime/manager.go

@@ -78,11 +78,14 @@ type Manager struct {
 	restarts       map[string]int
 	restartBackoff map[string]int
 	recovering     map[string]bool
-	recoverDone    sync.WaitGroup
-	nextPort       int
-	httpClient     *http.Client
-	runCtx         context.Context
-	cancel         context.CancelFunc
+	recoverPending map[string]string
+	// recoverUnwindHook runs (tests only) as a recovery goroutine starts unwinding.
+	recoverUnwindHook func(id string)
+	recoverDone       sync.WaitGroup
+	nextPort          int
+	httpClient        *http.Client
+	runCtx            context.Context
+	cancel            context.CancelFunc
 	// The persistence path is one mutex-guarded goroutine. The dirty set
 	// is keyed by account ID and merged on enqueue: a stale snapshot (older
 	// StateVersion) that arrives after a newer one is discarded, so the final
@@ -164,6 +167,7 @@ func NewManager(config ManagerConfig, store AccountStore, starter ProcessStarter
 		restarts:          map[string]int{},
 		restartBackoff:    map[string]int{},
 		recovering:        map[string]bool{},
+		recoverPending:    map[string]string{},
 		nextPort:          config.BasePort,
 		httpClient:        &http.Client{Timeout: 2 * time.Second},
 		runCtx:            runCtx,
@@ -524,6 +528,22 @@ func (m *Manager) TestRecovering(id string) bool {
 	return m.recovering[id]
 }
 
+// TestSetRecoverUnwindHook installs a hook run when a recovery goroutine
+// starts unwinding, before it clears the recovering guard. Tests only.
+func (m *Manager) TestSetRecoverUnwindHook(hook func(id string)) {
+	m.mu.Lock()
+	m.recoverUnwindHook = hook
+	m.mu.Unlock()
+}
+
+// TestRecoverPending reports whether a failure is queued behind an active recovery.
+func (m *Manager) TestRecoverPending(id string) bool {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	_, ok := m.recoverPending[id]
+	return ok
+}
+
 func (m *Manager) TestRestartDelay(level int) time.Duration {
 	return m.restartDelay(level)
 }

internal/runtime/manager_recovery.go

@@ -32,6 +32,10 @@ func (m *Manager) beginAccountRecovery(id string, startErr error) {
 		return
 	}
 	if m.recovering[id] {
+		// A recovery goroutine is still unwinding (e.g. its startAccount just
+		// registered a process that already exited). Remember the failure so
+		// the goroutine re-enters recovery instead of dropping the crash.
+		m.recoverPending[id] = message
 		m.mu.Unlock()
 		return
 	}
@@ -50,9 +54,20 @@ func (m *Manager) beginAccountRecovery(id string, startErr error) {
 
 func (m *Manager) recoverAccount(id string, restarts, backoffLevel int, message string) {
 	defer func() {
+		m.mu.Lock()
+		hook := m.recoverUnwindHook
+		m.mu.Unlock()
+		if hook != nil {
+			hook(id)
+		}
 		m.mu.Lock()
 		delete(m.recovering, id)
+		pending, again := m.recoverPending[id]
+		delete(m.recoverPending, id)
 		m.mu.Unlock()
+		if again {
+			m.beginAccountRecovery(id, errors.New(pending))
+		}
 		m.recoverDone.Done()
 	}()
 

internal/runtime/manager_test.go

@@ -884,6 +884,51 @@ func TestManagerEscalatesConsecutiveRestartBackoffSeparately(t *testing.T) {
 	}
 }
 
+// A process that exits while the recovery goroutine that started it is still
+// unwinding must not be dropped by the recovering guard.
+func TestManagerRecoversCrashDuringRecoveryUnwind(t *testing.T) {
+	ctx := context.Background()
+	store, err := sqlstore.OpenStore(filepath.Join(t.TempDir(), "qoder.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+	account, err := store.Create(ctx, accounts.CreateAccount{Name: "Unwind", Enabled: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	starter := &fakeStarter{started: make(chan *fakeProcess, 3)}
+	manager := accountruntime.NewManager(accountruntime.ManagerConfig{DataDir: t.TempDir(), RestartDelay: time.Millisecond}, store, starter)
+	defer manager.Close()
+	release := make(chan struct{})
+	var once sync.Once
+	manager.TestSetRecoverUnwindHook(func(string) { once.Do(func() { <-release }) })
+	if err := manager.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	first := <-starter.started
+	first.done <- errors.New("first crash")
+	second := <-starter.started
+	second.done <- errors.New("second crash")
+	deadline := time.Now().Add(5 * time.Second)
+	for !manager.TestRecoverPending(account.ID) {
+		if time.Now().After(deadline) {
+			close(release)
+			t.Fatal("second crash was not queued behind the unwinding recovery")
+		}
+		time.Sleep(time.Millisecond)
+	}
+	close(release)
+	select {
+	case third := <-starter.started:
+		if third == second {
+			t.Fatal("manager reused exited process")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("crash during recovery unwind was dropped; account never restarted")
+	}
+}
+
 func TestManagerRestartDelayIsBounded(t *testing.T) {
 	manager := accountruntime.NewManager(accountruntime.ManagerConfig{RestartDelay: 2 * time.Second, RestartMaxDelay: 5 * time.Second}, nil, &fakeStarter{})
 	if got := manager.TestRestartDelay(1); got != 2*time.Second {