From 779be27bfcf34c6b4e5dfa7096dca1e8bebac6d6 Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:04:22 +0800 Subject: [PATCH] =?UTF-8?q?fix(edge):=20=E6=81=A2=E5=A4=8D=E4=B8=BB?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=20goroutine=20=E4=B8=8E=E8=81=9A=E5=90=88?= =?UTF-8?q?=E5=99=A8=20panic=EF=BC=8C=E5=B9=82=E7=AD=89=20Bus.Close?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - lifecycle：把 run/watchRunProcess/publishOutput/publishStructuredOutput/cancelGrace/resultAggregator/resultAggregatorTimeout 全部裸 go 改走 safeGo，避免任一 panic 崩溃整个 Edge 进程 - events.Bus.Close：用 sync.Once 幂等化，二次调用不再 panic；新增 TestBusCloseIdempotent - finish：清理包级 hubCallbackQueues/hubStreamChunkSeq，避免 panic 路径跳过 fireHubDone/Fail 时条目泄漏 - hubCallbackQueueState.close：新增幂等 close 方法供 finish 与 terminal enqueue 共用 Co-authored-by: Cursor --- edge-server/internal/events/bus.go | 26 +++++++++++++------ .../internal/events/bus_behavior_test.go | 13 +++++++++- .../lifecycle/process_executor_build.go | 10 +++---- .../lifecycle/process_executor_cancel.go | 4 +-- .../lifecycle/process_executor_finish.go | 13 ++++++++++ .../process_executor_hub_callback.go | 14 ++++++++++ .../lifecycle/process_executor_start.go | 7 ++++- .../internal/lifecycle/result_aggregator.go | 6 ++--- 8 files changed, 73 insertions(+), 20 deletions(-) diff --git a/edge-server/internal/events/bus.go b/edge-server/internal/events/bus.go index 0dcf7c4c6..a9d7f31fe 100644 --- a/edge-server/internal/events/bus.go +++ b/edge-server/internal/events/bus.go @@ -96,6 +96,10 @@ type Bus struct { stopCh chan struct{} jobs chan observerJob workersWg sync.WaitGroup + + // closeOnce makes Close idempotent: a second call (e.g. from a shutdown + // hook + a defer in the same process) must not re-close stopCh and panic. + closeOnce sync.Once } // NewBus creates a new event bus with the given maximum history size. @@ -431,13 +435,19 @@ func (b *Bus) DroppedCount() int64 { // Close shuts down the observer worker pool, closes the job channel, and // flushes and closes the underlying event log if one was configured via // WithEventLogPath. It is safe to call Close on a Bus that has no event log. +// +// Close is idempotent: a second call (e.g. from a shutdown hook plus a defer +// in the same process) returns nil without re-closing stopCh and panicking. func (b *Bus) Close() error { - // Shut down the observer worker pool. - close(b.stopCh) - b.workersWg.Wait() - - if b.eventLog != nil { - return b.eventLog.Close() - } - return nil + var eventLogErr error + b.closeOnce.Do(func() { + // Shut down the observer worker pool. + close(b.stopCh) + b.workersWg.Wait() + + if b.eventLog != nil { + eventLogErr = b.eventLog.Close() + } + }) + return eventLogErr } diff --git a/edge-server/internal/events/bus_behavior_test.go b/edge-server/internal/events/bus_behavior_test.go index aa5cb3559..fccf63f20 100644 --- a/edge-server/internal/events/bus_behavior_test.go +++ b/edge-server/internal/events/bus_behavior_test.go @@ -625,7 +625,18 @@ func TestBusCloseShutsDownWorkers(t *testing.T) { // After close, verify the bus is no longer operational. // Worker pool is stopped; observers will not be processed. // Note: Close() closes stopCh and waits for workers, then closes eventLog if any. - // It is NOT idempotent — calling Close() twice panics because stopCh is re-closed. + // Close() is idempotent via sync.Once — calling it twice is safe (see TestBusCloseIdempotent). +} + +func TestBusCloseIdempotent(t *testing.T) { + b := NewBus(10) + if err := b.Close(); err != nil { + t.Fatalf("first Close: %v", err) + } + // A second Close must not panic on re-closing stopCh. + if err := b.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } } // --------------------------------------------------------------------------- diff --git a/edge-server/internal/lifecycle/process_executor_build.go b/edge-server/internal/lifecycle/process_executor_build.go index fb1a27f31..803aef8ea 100644 --- a/edge-server/internal/lifecycle/process_executor_build.go +++ b/edge-server/internal/lifecycle/process_executor_build.go @@ -130,7 +130,7 @@ func (e *ProcessExecutor) buildAndStartProcess( e.mu.Lock() e.processes[run.ID] = cmd.Process e.mu.Unlock() - go e.watchRunProcess(ctx, run.ID, cmd.Process, watchStop) + safeGo("watchRunProcess", func() { e.watchRunProcess(ctx, run.ID, cmd.Process, watchStop) }) } // Eager-close stdin when adapter/decision-loop do not need the pipe. @@ -228,7 +228,7 @@ func (e *ProcessExecutor) collectAndWaitOutput( var wg sync.WaitGroup outputLimiter := newRunOutputLimiter(e.maxRunOutputBytes) wg.Add(1) - go e.publishOutput(&wg, run, outStore, outputLimiter, "stderr", proc.stderr) + safeGo("publishOutput.stderr", func() { e.publishOutput(&wg, run, outStore, outputLimiter, "stderr", proc.stderr) }) // Inject context budget for token tracking in stream parsers. // Also inject RunProcessContext unconditionally — SDK adapters @@ -238,11 +238,11 @@ func (e *ProcessExecutor) collectAndWaitOutput( if proc.buildPlan.UseStructuredParser { wg.Add(1) - go e.publishStructuredOutput(&wg, run, proc.stdout, proc.stdin, adapter, parserCtx, &parseErr) + safeGo("publishStructuredOutput", func() { e.publishStructuredOutput(&wg, run, proc.stdout, proc.stdin, adapter, parserCtx, &parseErr) }) } else { // Raw capture: stdout goes to run.output.batch events wg.Add(1) - go e.publishOutput(&wg, run, outStore, outputLimiter, "stdout", proc.stdout) + safeGo("publishOutput.stdout", func() { e.publishOutput(&wg, run, outStore, outputLimiter, "stdout", proc.stdout) }) } // StdoutPipe/StderrPipe readers must finish before Wait closes the pipe @@ -466,6 +466,6 @@ func (e *ProcessExecutor) handleFaultEscalation( e.bus.Publish("run.fault_escalation.retry", runScope(*run), faultEscalationRetryPayload(run.ID, newCount, e.faultEscalationCfg.MaxRetries)) slog.Warn("process: fault escalation auto-retry", "runId", run.ID, "retryCount", newCount, "maxRetries", e.faultEscalationCfg.MaxRetries) - go e.run(newCtx, *run, runCtx) + safeGo("run.faultEscalation", func() { e.run(newCtx, *run, runCtx) }) return true } diff --git a/edge-server/internal/lifecycle/process_executor_cancel.go b/edge-server/internal/lifecycle/process_executor_cancel.go index de3dc2eb5..318526656 100644 --- a/edge-server/internal/lifecycle/process_executor_cancel.go +++ b/edge-server/internal/lifecycle/process_executor_cancel.go @@ -47,7 +47,7 @@ func (e *ProcessExecutor) Cancel(runID string) CancelResult { e.mu.Lock() e.cancelDone[runID] = done e.mu.Unlock() - go func() { + safeGo("cancelGrace", func() { select { case <-done: return @@ -68,7 +68,7 @@ func (e *ProcessExecutor) Cancel(runID string) CancelResult { if _, err := proc.Wait(); planProcessWaitAfterKill(err).Log { slog.Warn("process wait error after kill", "run_id", runID, "error", err) } - }() + }) } // Cancel the run context after the grace path is armed. This stops diff --git a/edge-server/internal/lifecycle/process_executor_finish.go b/edge-server/internal/lifecycle/process_executor_finish.go index 4ecf41cce..184c77d40 100644 --- a/edge-server/internal/lifecycle/process_executor_finish.go +++ b/edge-server/internal/lifecycle/process_executor_finish.go @@ -58,6 +58,19 @@ func (e *ProcessExecutor) finish(runID string) { delete(e.runOutputs, runID) } e.mu.Unlock() + + // Clean up package-level hub callback state so a run that panicked and + // was recovered by safeGo (skipping fireHubDone/fireHubFail) does not + // leak hubCallbackQueues / hubStreamChunkSeq entries. The consumer + // goroutine's own defer also deletes; sync.Map.Delete is idempotent. + // Closing the queue channel unblocks a consumer stuck on range, letting + // it drain and exit cleanly. + if stateAny, ok := hubCallbackQueues.LoadAndDelete(runID); ok { + if state, ok := stateAny.(*hubCallbackQueueState); ok { + state.close() + } + } + hubStreamChunkSeq.Delete(runID) } // hasActiveChildren reports whether the given run has at least one registered diff --git a/edge-server/internal/lifecycle/process_executor_hub_callback.go b/edge-server/internal/lifecycle/process_executor_hub_callback.go index 9e5a27904..22697352e 100644 --- a/edge-server/internal/lifecycle/process_executor_hub_callback.go +++ b/edge-server/internal/lifecycle/process_executor_hub_callback.go @@ -249,6 +249,20 @@ func newHubCallbackQueueState(runID string) *hubCallbackQueueState { } } +// close idempotently closes the queue channel so the consumer goroutine (if +// running) drains remaining jobs and exits. It is safe to call from finish() +// when a panicked run skipped fireHubDone/fireHubFail and never enqueued a +// terminal job — otherwise the queue entry and consumer would leak. +func (s *hubCallbackQueueState) close() { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return + } + s.closed = true + close(s.ch) +} + // enqueueHubStreamJob appends a stream chunk to the run's FIFO without // blocking. Returns false when the queue is full (chunk dropped) or already // closed (terminal callback decided the run's delivery). diff --git a/edge-server/internal/lifecycle/process_executor_start.go b/edge-server/internal/lifecycle/process_executor_start.go index fc4b0f35c..be735c38c 100644 --- a/edge-server/internal/lifecycle/process_executor_start.go +++ b/edge-server/internal/lifecycle/process_executor_start.go @@ -20,7 +20,12 @@ func (e *ProcessExecutor) Start(run store.Run, runCtx RunProcessContext) error { ctx, cancel := context.WithTimeout(context.Background(), e.runTimeout) e.running[run.ID] = cancel - go e.run(ctx, run, bindRunProcessContext(runCtx, run)) + // Spawn the run lifecycle goroutine through safeGo so a panic inside run() + // (adapter parse, emitter chain, output store) is recovered and logged + // instead of crashing the whole Edge process. run()'s deferred finish() + // still runs during panic unwinding before recover catches, so terminal + // state and package-level sync.Map hygiene are preserved. + safeGo("run", func() { e.run(ctx, run, bindRunProcessContext(runCtx, run)) }) return nil } diff --git a/edge-server/internal/lifecycle/result_aggregator.go b/edge-server/internal/lifecycle/result_aggregator.go index 32e7e0a05..5846cc310 100644 --- a/edge-server/internal/lifecycle/result_aggregator.go +++ b/edge-server/internal/lifecycle/result_aggregator.go @@ -106,12 +106,12 @@ func (ra *ResultAggregator) Start() (stop func()) { ra.subID = subID done := make(chan struct{}) - go func() { + safeGo("resultAggregator", func() { defer close(done) for evt := range ch { ra.handleEvent(evt) } - }() + }) // Timeout fallback goroutine: periodically checks for parents whose // children have exceeded the configured timeout. When found, emits @@ -120,7 +120,7 @@ func (ra *ResultAggregator) Start() (stop func()) { var timeoutDone chan struct{} if ra.collector != nil { timeoutDone = make(chan struct{}) - go ra.runTimeoutCheck(timeoutDone) + safeGo("resultAggregatorTimeout", func() { ra.runTimeoutCheck(timeoutDone) }) } return func() {