From 78593c99385b7dd59c1d3061fe6a34eacb3862e1 Mon Sep 17 00:00:00 2001
From: HuaiYJ <huaiyujiao@foxmail.com>
Date: Mon, 24 Aug 2026 16:21:14 +0800
Subject: [PATCH] fix: scope resume work to connection epoch

---
 cmd/ocm-agent/main.go                |  2 +-
 internal/data/large_reporter.go      |  4 ++--
 internal/data/large_reporter_test.go | 26 ++++++++++++++++++++++++++
 3 files changed, 29 insertions(+), 3 deletions(-)

diff --git a/cmd/ocm-agent/main.go b/cmd/ocm-agent/main.go
index 936f119..3ca68ac 100644
--- a/cmd/ocm-agent/main.go
+++ b/cmd/ocm-agent/main.go
@@ -353,7 +353,7 @@ func connectionSupervisor(
 
 		// 恢复大数据进行中会话
 		if largeReporter != nil {
-			go largeReporter.Resume(quit)
+			go largeReporter.Resume(epochCtx.Done())
 		}
 
 		// 数据通道服务器
diff --git a/internal/data/large_reporter.go b/internal/data/large_reporter.go
index 9e24361..92152d6 100644
--- a/internal/data/large_reporter.go
+++ b/internal/data/large_reporter.go
@@ -772,7 +772,7 @@ func (r *LargeReporter) onError(frame *protocol.Frame) {
 //
 // Resume 是尽力而为的：单条失败仅以 WARN 记录，不会中断扫描。它应当
 // 在 cmd/ocm-agent/main.go 中、connector 完全建立之后调用一次。
-func (r *LargeReporter) Resume(agentDone <-chan struct{}) {
+func (r *LargeReporter) Resume(epochDone <-chan struct{}) {
 	if r.store == nil {
 		return
 	}
@@ -783,7 +783,7 @@ func (r *LargeReporter) Resume(agentDone <-chan struct{}) {
 	}
 	for _, row := range rows {
 		select {
-		case <-agentDone:
+		case <-epochDone:
 			return
 		default:
 		}
diff --git a/internal/data/large_reporter_test.go b/internal/data/large_reporter_test.go
index 941a3ed..643a9ec 100644
--- a/internal/data/large_reporter_test.go
+++ b/internal/data/large_reporter_test.go
@@ -7,6 +7,7 @@ import (
 	"crypto/md5"
 	"encoding/hex"
 	"errors"
+	"os"
 	"path/filepath"
 	"sync"
 	"testing"
@@ -165,6 +166,31 @@ func TestLargeReporter_HappyPath_TwoFragments(t *testing.T) {
 	}
 }
 
+func TestLargeReporter_ResumeStopsWhenEpochEnds(t *testing.T) {
+	cfg := DefaultLargeReporterConfig()
+	reporter, sender, store := newTestLargeReporter(t, cfg)
+	source := filepath.Join(t.TempDir(), "resume.bin")
+	if err := os.WriteFile(source, []byte("resume payload"), 0o600); err != nil {
+		t.Fatalf("write source: %v", err)
+	}
+	if err := store.Upsert(OngoingTask{
+		TaskID: 1, DataID: 2, TotalSize: 14, MD5: "unused", SourcePath: source,
+	}); err != nil {
+		t.Fatalf("upsert ongoing task: %v", err)
+	}
+	epochDone := make(chan struct{})
+	close(epochDone)
+
+	reporter.Resume(epochDone)
+
+	sender.mu.Lock()
+	sent := len(sender.sent)
+	sender.mu.Unlock()
+	if sent != 0 {
+		t.Fatalf("Resume sent %d frames after epoch cancellation", sent)
+	}
+}
+
 func TestLargeReporter_DirectRequireFragAfterUpload(t *testing.T) {
 	cfg := DefaultLargeReporterConfig()
 	cfg.PerFragmentTimeout = 100 * time.Millisecond
-- 
Gitee