fix: preserve outcomes across session replacement

docs/api/control-plane.md

@@ -150,8 +150,9 @@ PostgreSQL Proxy 明细。Gateway 将每次尝试写入进程内有界队列，
 同一 session 内，Controller 仅接受递增 `sequence`；相同序列且摘要一致视为幂等
 重放，相同序列且摘要不同返回 `AlreadyExists`，较小序列返回 `Aborted`。Gateway 在
 未收到确认时保留并重发完全相同的批次。暂态传输错误在当前 session 内按有上限的
-退避重试；会话栅栏错误交由 session supervisor 重建 session。栅栏随 session 替换
+退避重试；会话栅栏错误交由 session supervisor 重建 session。重注册会保留未确认
-或过期清理，因此长期在线 Worker 不会因独立 Outcome TTL 接受旧序列。
+事件，并以新 session 从序列 `1` 重新封装后上报。栅栏随 session 替换或过期清理，
+因此长期在线 Worker 不会因独立 Outcome TTL 接受旧序列。
 
 ## 6. Checker 任务
 

docs/configuration/reference.md

@@ -201,7 +201,8 @@ Checker 同样使用独立的可拨号地址：`proxy-checker` 的 `-control-pla
 `-checker-id`、`-instance-id` 和 `-max-in-flight` 可由对应的
 `PROXY_POOL_*` 环境变量提供。mTLS 模式下该命令读取 `checkerTLS`，明文 fixture
 模式只接受回环 Controller 地址。Checker 只从 gRPC 领取任务并批量上报事实，不读取
-Redis/PostgreSQL；生产 Redis 共享任务队列尚在后续实施范围。
+Redis/PostgreSQL；Controller 在生产启动拓扑中装配 Redis 共享任务队列，当前调度
+HTTP/HTTPS BASIC 检查。EGRESS 和 TARGET 的生产调度仍在后续实施范围。
 
 `maxRuntimeCounters` 同时限制单个 Runtime 报告和单个 Outcome 批次的条目数。Gateway
 在本地维护容量为 `65536` 的非阻塞 Outcome 队列，默认微批上限为 `512`，实际取二者中

internal/gateway/controlplane/outcome_reporter.go

@@ -68,16 +68,14 @@ func NewOutcomeReporter(client OutcomeRPCClient, queue *gatewayOutcome.Queue, op
 
 // RunRegistered continuously batches local observations for a single
 // Controller session. Transient delivery failures retain the exact pending
-// batches and retry them in the same session with bounded backoff.
+// batches and retry them in the same session with bounded backoff. A replaced
+// Controller session reissues the retained events with the new session ID and
+// its sequence space.
 func (reporter *OutcomeReporter) RunRegistered(ctx context.Context, registration Registration) error {
 	if reporter == nil || ctx == nil || !workerruntime.ValidIdentifier(registration.SessionID) {
 		return ErrInvalidOutcomeReporter
 	}
-	reporter.mu.Lock()
+	reporter.startSession(registration.SessionID)
-	reporter.sessionID = registration.SessionID
-	reporter.sequence = 0
-	reporter.pending = nil
-	reporter.mu.Unlock()
 	retryDelay := outcomeRetryInitialDelay
 	for {
 		err := reporter.Report(ctx, registration.SessionID)
@@ -98,6 +96,20 @@ func (reporter *OutcomeReporter) RunRegistered(ctx context.Context, registration
 	}
 }
 
+func (reporter *OutcomeReporter) startSession(sessionID string) {
+	reporter.mu.Lock()
+	defer reporter.mu.Unlock()
+	if reporter.sessionID == sessionID {
+		return
+	}
+	reporter.sessionID = sessionID
+	reporter.sequence = 0
+	for index := range reporter.pending {
+		reporter.pending[index].SessionID = sessionID
+		reporter.pending[index].Sequence = uint64(index + 1)
+	}
+}
+
 func terminalOutcomeError(err error) bool {
 	if errors.Is(err, ErrInvalidOutcomeReporter) || errors.Is(err, ErrNotRegistered) {
 		return true

internal/gateway/controlplane/outcome_reporter_test.go

@@ -91,6 +91,42 @@ func TestOutcomeReporterRetriesTransientFailureWithinRegisteredSession(t *testin
 	}
 }
 
+func TestOutcomeReporterReissuesPendingBatchAfterSessionReplacement(t *testing.T) {
+	queue, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{Capacity: 4, MaxBatch: 2})
+	if err != nil {
+		t.Fatalf("NewQueue() = %v", err)
+	}
+	queue.Record(domain.Event{ProxyID: "proxy-a", Stage: domain.StageDial, Success: true, ObservedAt: time.Now()})
+	client := &outcomeClientStub{closeErr: errors.New("connection dropped")}
+	reporter, err := NewOutcomeReporter(client, queue, OutcomeReporterOptions{WorkerID: "worker-a"})
+	if err != nil {
+		t.Fatalf("NewOutcomeReporter() = %v", err)
+	}
+	reporter.sessionID = "session-a"
+	if err := reporter.Report(context.Background(), "session-a"); !errors.Is(err, client.closeErr) {
+		t.Fatalf("Report(session-a) = %v, want transport failure", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
+	defer cancel()
+	client.closeErr = nil
+	client.onConfirmed = cancel
+	if err := reporter.RunRegistered(ctx, Registration{SessionID: "session-b"}); !errors.Is(err, context.Canceled) {
+		t.Fatalf("RunRegistered(session-b) = %v, want context canceled", err)
+	}
+	if len(client.batches) != 2 {
+		t.Fatalf("sent batch count = %d, want 2", len(client.batches))
+	}
+	first, replay := client.batches[0], client.batches[1]
+	if replay.GetSessionId() != "session-b" || replay.GetSequence() != 1 ||
+		len(replay.GetOutcomes()) != 1 || replay.GetOutcomes()[0].GetProxyId() != "proxy-a" {
+		t.Fatalf("replacement batch = %+v, want proxy-a in session-b sequence 1", replay)
+	}
+	if first.GetSessionId() != "session-a" || first.GetSequence() != 1 {
+		t.Fatalf("original batch = %+v, want session-a sequence 1", first)
+	}
+}
+
 func TestOutcomeReporterRejectsUnboundedStreamOptions(t *testing.T) {
 	queue, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{Capacity: 1, MaxBatch: 1})
 	if err != nil {
@@ -107,6 +143,7 @@ type outcomeClientStub struct {
 	batches     []*controlplanev1.OutcomeBatch
 	closeErr    error
 	accepted    uint64
+	onConfirmed func()
 }
 
 func (client *outcomeClientStub) ReportOutcomes(context.Context) (OutcomeStream, error) {
@@ -128,6 +165,9 @@ func (stream *outcomeStreamStub) CloseAndRecv() (*controlplanev1.ReportOutcomesR
 	if accepted == 0 && len(stream.client.batches) > 0 {
 		accepted = stream.client.batches[len(stream.client.batches)-1].GetSequence()
 	}
+	if stream.client.onConfirmed != nil {
+		stream.client.onConfirmed()
+	}
 	return &controlplanev1.ReportOutcomesResponse{AcceptedThroughSequence: accepted}, nil
 }