diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index d11bc0a..109831c 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -150,8 +150,9 @@ PostgreSQL Proxy 明细。Gateway 将每次尝试写入进程内有界队列, 同一 session 内,Controller 仅接受递增 `sequence`;相同序列且摘要一致视为幂等 重放,相同序列且摘要不同返回 `AlreadyExists`,较小序列返回 `Aborted`。Gateway 在 未收到确认时保留并重发完全相同的批次。暂态传输错误在当前 session 内按有上限的 -退避重试;会话栅栏错误交由 session supervisor 重建 session。栅栏随 session 替换 -或过期清理,因此长期在线 Worker 不会因独立 Outcome TTL 接受旧序列。 +退避重试;会话栅栏错误交由 session supervisor 重建 session。重注册会保留未确认 +事件,并以新 session 从序列 `1` 重新封装后上报。栅栏随 session 替换或过期清理, +因此长期在线 Worker 不会因独立 Outcome TTL 接受旧序列。 ## 6. Checker 任务 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index 40d0345..adbfefa 100644 --- a/docs/configuration/reference.md +++ b/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`,实际取二者中 diff --git a/internal/gateway/controlplane/outcome_reporter.go b/internal/gateway/controlplane/outcome_reporter.go index be5f1ae..fd1897c 100644 --- a/internal/gateway/controlplane/outcome_reporter.go +++ b/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.sessionID = registration.SessionID - reporter.sequence = 0 - reporter.pending = nil - reporter.mu.Unlock() + reporter.startSession(registration.SessionID) 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 diff --git a/internal/gateway/controlplane/outcome_reporter_test.go b/internal/gateway/controlplane/outcome_reporter_test.go index 7d86e9e..1b1c15d 100644 --- a/internal/gateway/controlplane/outcome_reporter_test.go +++ b/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 { @@ -104,9 +140,10 @@ func TestOutcomeReporterRejectsUnboundedStreamOptions(t *testing.T) { } type outcomeClientStub struct { - batches []*controlplanev1.OutcomeBatch - closeErr error - accepted uint64 + 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 }