From 34892d321ef807bb4d65f8bde9749fc21cd7c139 Mon Sep 17 00:00:00 2001
From: ndou <447662456@qq.com>
Date: Thu, 13 Aug 2026 14:19:52 +0800
Subject: [PATCH] fix: fail when data channel cannot start

---
 cmd/ocm-agent/main.go | 51 +++++++++++++++++++++++++++++++++++--------
 1 file changed, 42 insertions(+), 9 deletions(-)

diff --git a/cmd/ocm-agent/main.go b/cmd/ocm-agent/main.go
index 58367b0..25d40b9 100644
--- a/cmd/ocm-agent/main.go
+++ b/cmd/ocm-agent/main.go
@@ -200,7 +200,23 @@ func run() error {
 	// 启动 connection supervisor：管理连接生命周期，断连后自动重连。
 	// 每个连接代际（epoch）启动一组 (ReadLoop, hbReporter, reporter, largeReporter)，
 	// 通过 epoch-scoped context 控制生命周期。
-	go connectionSupervisor(conn, reconnector, agentID, cfg, *configPath, store, quit, log)
+	supervisorErr := make(chan error, 1)
+	supervisorReady := make(chan struct{}, 1)
+	go func() {
+		supervisorErr <- connectionSupervisor(conn, reconnector, agentID, cfg, *configPath, store, quit, log, supervisorReady)
+	}()
+
+	select {
+	case <-supervisorReady:
+	case <-quit:
+		log.Info("agent shutting down before connection epoch became ready")
+		return nil
+	case err := <-supervisorErr:
+		if err != nil {
+			return err
+		}
+		return nil
+	}
 
 	// 任务工作目录清理器：跟随进程生命周期周期性 sweep
 	// <DataDir>/tasks/ 下按 task 划分的子目录，按 mtime 删除超过 TTL
@@ -231,10 +247,17 @@ func run() error {
 
 	log.Info("agent running", zap.Uint64("agent_id", agentID))
 
-	// 主循环：等待关闭信号。
-	<-quit
-	log.Info("agent shutting down")
-	return nil
+	// 主循环：等待关闭信号或关键后台组件失败。
+	select {
+	case <-quit:
+		log.Info("agent shutting down")
+		return nil
+	case err := <-supervisorErr:
+		if err != nil {
+			return err
+		}
+		return nil
+	}
 }
 
 // connectionSupervisor 管理连接的生命周期。每个 epoch 代表一次成功连接
@@ -250,8 +273,10 @@ func connectionSupervisor(
 	store *data.OngoingStore,
 	quit <-chan struct{},
 	log *zap.Logger,
-) {
+	ready chan<- struct{},
+) error {
 	epoch := 0
+	readySent := false
 	for {
 		epoch++
 		log.Info("starting connection epoch", zap.Int("epoch", epoch), zap.Uint64("agent_id", agentID))
@@ -354,7 +379,11 @@ func connectionSupervisor(
 		// 数据通道服务器
 		dataChanServer := dc.NewServer(UnixSocketPath, log, reporter)
 		if err := dataChanServer.Start(); err != nil {
-			log.Error("failed to start data channel server", zap.Error(err))
+			epochCancel()
+			hbReporter.Stop()
+			reporter.Stop()
+			conn.Close()
+			return fmt.Errorf("start data channel server: %w", err)
 		}
 
 		// 插件管理器
@@ -370,6 +399,10 @@ func connectionSupervisor(
 		collector.RunWithContext(epochCtx)
 
 		log.Info("epoch components started", zap.Int("epoch", epoch))
+		if !readySent {
+			ready <- struct{}{}
+			readySent = true
+		}
 
 		// --- 等待断连信号或关闭信号 ---
 		select {
@@ -385,7 +418,7 @@ func connectionSupervisor(
 			if pluginManager != nil {
 				pluginManager.Stop()
 			}
-			return
+			return nil
 		}
 
 		// --- 停止当前 epoch 的所有组件 ---
@@ -403,7 +436,7 @@ func connectionSupervisor(
 		log.Info("starting reconnect loop", zap.Int("epoch", epoch))
 		if !reconnector.ReconnectLoop(quit) {
 			log.Info("shutdown during reconnect")
-			return
+			return nil
 		}
 		conn.ResetForNewEpoch()
 		log.Info("reconnect succeeded, starting new epoch", zap.Int("epoch", epoch+1))
-- 
Gitee