From baefaf6dbf0636840da4cc4f4da9a0c18b2a1c31 Mon Sep 17 00:00:00 2001 From: youfak Date: Fri, 31 Jul 2026 14:43:25 +0800 Subject: [PATCH] fix: stop gateway dispatch after snapshot expiry --- docs/api/control-plane.md | 4 ++ docs/design/architecture.md | 3 +- docs/requirements/completion-audit.md | 3 +- internal/gateway/controlplane/watcher.go | 17 +++++++- internal/gateway/controlplane/watcher_test.go | 40 ++++++++++++++++- internal/gateway/dispatch/dispatcher_test.go | 21 +++++++++ internal/gateway/snapshot/store.go | 43 ++++++++++--------- 7 files changed, 107 insertions(+), 24 deletions(-) diff --git a/docs/api/control-plane.md b/docs/api/control-plane.md index bcea8ba..c46aeec 100644 --- a/docs/api/control-plane.md +++ b/docs/api/control-plane.md @@ -88,6 +88,10 @@ Worker 自报的超前版本或 epoch 也必须拒绝。 `usable_until` 作为最后可分配时刻；达到该时间后即使尚未到 `expires_at`， 也不得再为新请求选择该 Proxy。 +Gateway 接收完整快照时必须拒绝缺失、格式错误或已到期的 `valid_until`，并将其 +保存在本地不可变视图。该整体期限到达后，调度直接按无候选处理，不再使用旧视图 +发起新的上游连接，也不查询 Redis 或 PostgreSQL 补偿。 + Delta 声明 `base_version`。Worker 只有在本地版本恰好等于 base 且 checksum 验证成功时才能应用；否则丢弃 Delta 并请求完整 Snapshot。构建在后台完成， 热路径只读取一次原子指针。 diff --git a/docs/design/architecture.md b/docs/design/architecture.md index 956ddd4..825dd1d 100644 --- a/docs/design/architecture.md +++ b/docs/design/architecture.md @@ -383,7 +383,8 @@ flowchart LR `UsableUntil` 和版本校验和。Worker 在 `UsableUntil` 到达后立即停止新分配， 不等待供应商硬过期时间 `ExpiresAt`。 - Worker 断开控制面后在 `maxStaleAge` 内使用最后快照；超限停止接收新流量， - 已有隧道排空。 + 已有隧道排空。Gateway 会把 Controller Snapshot 的整体 `valid_until` 保存在 + 本地视图；该期限到达后本地调度不再为新请求分配 Proxy。 ## 12. 存储与一致性 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 26b74dc..b4cd58a 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -53,7 +53,8 @@ Register/Watch/ACK/Runtime 会话协调已实现。Redis 以 Worker 可下发 ownership 索引构建无凭据引用的已归属 Proxy payload，并以租约收紧可用期；Routing payload、 凭据分发、Outcome 和 Checker 尚未闭环。Snapshot 签发在 Redis 中原子匹配当前 - `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。 + `session_id`，重注册会清除旧引用，迟到旧 Stream 不会覆盖新 session。Gateway 会 + 校验并执行 Snapshot 整体 `valid_until`，过期视图不再分配新 Proxy。 - `PostgreSQL 管理面`：已定义 `adminstate` 事务 seam、并发安全 MemoryStore、 公用契约、100 并发 Routing CAS、租约 Outbox 和只含六张管理表的 Schema；pgx Adapter 已在真实 PostgreSQL 18 上通过同一契约、迁移幂等、审计/Outbox diff --git a/internal/gateway/controlplane/watcher.go b/internal/gateway/controlplane/watcher.go index 7199020..5a6b6bd 100644 --- a/internal/gateway/controlplane/watcher.go +++ b/internal/gateway/controlplane/watcher.go @@ -108,6 +108,10 @@ func (watcher *SnapshotWatcher) applyFull(full *controlplanev1.WorkerSnapshot) e if full == nil || full.GetVersion() == 0 || full.GetOwnershipEpoch() == 0 { return ErrInvalidSnapshotWatcher } + validUntil, err := requiredFutureTimestamp(full.GetValidUntil(), time.Now().UTC()) + if err != nil { + return err + } checksum, err := snapshotwire.Checksum(full) if err != nil { return err @@ -121,7 +125,7 @@ func (watcher *SnapshotWatcher) applyFull(full *controlplanev1.WorkerSnapshot) e } envelope := snapshot.Envelope{ ClusterID: watcher.options.ClusterID, WorkerID: watcher.options.WorkerID, - Epoch: full.GetOwnershipEpoch(), Version: full.GetVersion(), Full: true, Proxies: proxies, + Epoch: full.GetOwnershipEpoch(), Version: full.GetVersion(), Full: true, ValidUntil: validUntil, Proxies: proxies, } envelope.Checksum = snapshot.Checksum(proxies) return watcher.store.Apply(envelope) @@ -192,6 +196,17 @@ func wireTimestamp(value *timestamppb.Timestamp) (*time.Time, error) { return &converted, nil } +func requiredFutureTimestamp(value *timestamppb.Timestamp, now time.Time) (time.Time, error) { + if value == nil || value.CheckValid() != nil || now.IsZero() { + return time.Time{}, ErrInvalidSnapshotWatcher + } + converted := value.AsTime().UTC() + if !converted.After(now) { + return time.Time{}, ErrInvalidSnapshotWatcher + } + return converted, nil +} + type generatedSnapshotRPCClient struct { client controlplanev1.WorkerControlPlaneClient } diff --git a/internal/gateway/controlplane/watcher_test.go b/internal/gateway/controlplane/watcher_test.go index 7f34b34..6d8d0e4 100644 --- a/internal/gateway/controlplane/watcher_test.go +++ b/internal/gateway/controlplane/watcher_test.go @@ -34,7 +34,8 @@ func TestSnapshotWatcherAppliesVerifiedFullSnapshotAndAcknowledges(t *testing.T) t.Fatalf("Watch(): %v", err) } view := store.Current() - if view == nil || view.Version != 1 || view.Epoch != 7 || len(view.Entries) != 1 || view.Entries[0].Proxy.ID != "proxy-a" { + if view == nil || view.Version != 1 || view.Epoch != 7 || len(view.Entries) != 1 || view.Entries[0].Proxy.ID != "proxy-a" || + !view.ValidUntil.Equal(full.GetValidUntil().AsTime()) { t.Fatalf("snapshot view = %+v", view) } if client.watch.GetSessionId() != "session-a" || client.ack.GetVersion() != 1 || !client.ack.GetApplied() || string(client.ack.GetChecksum()) != string(full.GetChecksum()) { @@ -58,6 +59,43 @@ func TestSnapshotWatcherRejectsChecksumAndAcknowledgesFailure(t *testing.T) { } } +func TestSnapshotWatcherRejectsMissingOverallValidityDeadline(t *testing.T) { + store := snapshot.NewStore("cluster-a", "worker-a") + full := &controlplanev1.WorkerSnapshot{Version: 1, OwnershipEpoch: 7, GeneratedAt: timestamppb.New(time.Now())} + setSnapshotChecksum(t, full) + client := &snapshotClientStub{stream: &snapshotStreamStub{values: []*controlplanev1.SnapshotEnvelope{{Payload: &controlplanev1.SnapshotEnvelope_Full{Full: full}}}}} + watcher, err := NewSnapshotWatcher(client, store, SnapshotWatcherOptions{ClusterID: "cluster-a", WorkerID: "worker-a"}) + if err != nil { + t.Fatalf("NewSnapshotWatcher(): %v", err) + } + if err := watcher.Watch(context.Background(), "session-a"); err == nil { + t.Fatal("Watch() error = nil, want missing validity rejection") + } + if client.ack == nil || client.ack.GetApplied() || client.ack.GetErrorCode() != "snapshot_apply_failed" { + t.Fatalf("negative acknowledgement = %+v", client.ack) + } +} + +func TestSnapshotWatcherRejectsExpiredOverallValidityDeadline(t *testing.T) { + store := snapshot.NewStore("cluster-a", "worker-a") + full := &controlplanev1.WorkerSnapshot{ + Version: 1, OwnershipEpoch: 7, GeneratedAt: timestamppb.New(time.Now().Add(-time.Minute)), + ValidUntil: timestamppb.New(time.Now().Add(-time.Second)), + } + setSnapshotChecksum(t, full) + client := &snapshotClientStub{stream: &snapshotStreamStub{values: []*controlplanev1.SnapshotEnvelope{{Payload: &controlplanev1.SnapshotEnvelope_Full{Full: full}}}}} + watcher, err := NewSnapshotWatcher(client, store, SnapshotWatcherOptions{ClusterID: "cluster-a", WorkerID: "worker-a"}) + if err != nil { + t.Fatalf("NewSnapshotWatcher(): %v", err) + } + if err := watcher.Watch(context.Background(), "session-a"); err == nil { + t.Fatal("Watch() error = nil, want expired validity rejection") + } + if client.ack == nil || client.ack.GetApplied() { + t.Fatalf("negative acknowledgement = %+v", client.ack) + } +} + type snapshotClientStub struct { stream SnapshotStream watch *controlplanev1.WatchSnapshotsRequest diff --git a/internal/gateway/dispatch/dispatcher_test.go b/internal/gateway/dispatch/dispatcher_test.go index aa12cef..61bbd62 100644 --- a/internal/gateway/dispatch/dispatcher_test.go +++ b/internal/gateway/dispatch/dispatcher_test.go @@ -50,6 +50,27 @@ func TestAcquireFiltersAndReservesLocalCapacity(t *testing.T) { } } +func TestAcquireRejectsSnapshotAfterOverallValidityDeadline(t *testing.T) { + now := time.Date(2026, 7, 31, 12, 0, 0, 0, time.UTC) + expiresAt := now.Add(time.Hour) + store := snapshot.NewStore("cluster-a", "worker-a") + envelope := snapshot.Envelope{ + ClusterID: "cluster-a", WorkerID: "worker-a", Epoch: 1, Version: 1, Full: true, + ValidUntil: now.Add(time.Second), + Proxies: []proxyDomain.Proxy{{ + ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, State: proxyDomain.StateAvailable, + MaxConcurrency: 1, ExpiresAt: &expiresAt, + }}, + } + envelope.Checksum = snapshot.Checksum(envelope.Proxies) + if err := store.Apply(envelope); err != nil { + t.Fatalf("Apply(): %v", err) + } + if _, err := New(store).Acquire(Request{Now: now.Add(time.Second), Scheme: proxyDomain.SchemeHTTP}); !errors.Is(err, ErrNoCandidate) { + t.Fatalf("Acquire(at validity deadline) error = %v, want ErrNoCandidate", err) + } +} + func TestAcquireNeverOversubscribesSnapshotProxy(t *testing.T) { store := snapshot.NewStore("cluster-a", "worker-a") proxies := []proxyDomain.Proxy{{ID: "p1", Scheme: proxyDomain.SchemeHTTP, State: proxyDomain.StateAvailable, MaxConcurrency: 8}} diff --git a/internal/gateway/snapshot/store.go b/internal/gateway/snapshot/store.go index fe3b3ed..eef21b4 100644 --- a/internal/gateway/snapshot/store.go +++ b/internal/gateway/snapshot/store.go @@ -92,13 +92,14 @@ type runtimeRegistration struct { } type Envelope struct { - ClusterID string - WorkerID string - Epoch uint64 - Version uint64 - Full bool - Checksum string - Proxies []proxyDomain.Proxy + ClusterID string + WorkerID string + Epoch uint64 + Version uint64 + Full bool + Checksum string + ValidUntil time.Time + Proxies []proxyDomain.Proxy } type Entry struct { @@ -107,12 +108,13 @@ type Entry struct { } type View struct { - ClusterID string - WorkerID string - Epoch uint64 - Version uint64 - Checksum string - Entries []Entry + ClusterID string + WorkerID string + Epoch uint64 + Version uint64 + Checksum string + ValidUntil time.Time + Entries []Entry all []int byScheme map[proxyDomain.Scheme][]int @@ -288,12 +290,13 @@ func (s *Store) Apply(envelope Envelope) error { } next := &View{ - ClusterID: envelope.ClusterID, - WorkerID: envelope.WorkerID, - Epoch: envelope.Epoch, - Version: envelope.Version, - Checksum: envelope.Checksum, - Entries: entries, + ClusterID: envelope.ClusterID, + WorkerID: envelope.WorkerID, + Epoch: envelope.Epoch, + Version: envelope.Version, + Checksum: envelope.Checksum, + ValidUntil: envelope.ValidUntil.UTC(), + Entries: entries, } next.buildIndexes() s.current.Store(next) @@ -304,7 +307,7 @@ func (v *View) Select(query Query) Selection { if query.Now.IsZero() { query.Now = time.Now().UTC() } - if v == nil { + if v == nil || (!v.ValidUntil.IsZero() && !v.ValidUntil.After(query.Now)) { return Selection{} }