diff --git a/cmd/meshd/main.go b/cmd/meshd/main.go index 5eac2c6..7e0f0d9 100644 --- a/cmd/meshd/main.go +++ b/cmd/meshd/main.go @@ -90,7 +90,8 @@ Up flags: --tun [name] Create a real TUN device (default; asks sudo when needed) --no-tun Use userspace mode without OS routes --port WireGuard UDP listen port (default: auto) - --poll-interval DWN poll interval (default: 30s) + --poll-interval + Override refresh cadence (fallback: 30s; healthy: 5m) --wait-timeout How long to wait for dashboard approval (default: 15m) --no-wait Exit immediately when the join request is still pending @@ -3590,7 +3591,7 @@ func parseUpFlags(args []string) upFlags { } case "--poll-interval": if i+1 < len(args) { - if d, err := time.ParseDuration(args[i+1]); err == nil { + if d, err := time.ParseDuration(args[i+1]); err == nil && d > 0 { f.pollInterval = d } i++ @@ -4249,7 +4250,7 @@ func cmdUp(ctx context.Context, args []string, flagProfile string) error { daemonSrv := daemon.NewServer(socketPath, func() daemon.Status { routing := eng.RoutingStatus() meshSnapshot := eng.MeshSnapshot() - selfStatus, peerStatuses, snapshotStatus := daemonStatusesFromMeshSnapshot(meshSnapshot) + selfStatus, peerStatuses, snapshotStatus := daemonStatusesFromMeshSnapshot(meshSnapshot, eng.RefreshHealth()) if meshSnapshot == nil { // Preserve the legacy tray/status view before the first materialized // snapshot attempt. The peer-list fast path requires Snapshot+Self @@ -4321,22 +4322,86 @@ func cmdUp(ctx context.Context, args []string, flagProfile string) error { return nil } -func daemonStatusesFromMeshSnapshot(snapshot *engine.MeshSnapshot) (*daemon.PeerStatus, []daemon.PeerStatus, *daemon.SnapshotStatus) { - if snapshot == nil { +func daemonStatusesFromMeshSnapshot(snapshot *engine.MeshSnapshot, refresh *engine.RefreshCoordinatorHealth) (*daemon.PeerStatus, []daemon.PeerStatus, *daemon.SnapshotStatus) { + if snapshot == nil && refresh == nil { return nil, nil, nil } var self *daemon.PeerStatus - if snapshot.Self != nil { - status := daemonPeerStatus(*snapshot.Self) - self = &status + var peers []daemon.PeerStatus + status := &daemon.SnapshotStatus{} + if snapshot != nil { + if snapshot.Self != nil { + peerStatus := daemonPeerStatus(*snapshot.Self) + self = &peerStatus + } + peers = daemonPeerStatuses(snapshot.Peers) + status.Generation = snapshot.Generation + status.RefreshedAt = daemonSnapshotTimestamp(snapshot.RefreshedAt) + status.LastAttemptAt = daemonSnapshotTimestamp(snapshot.LastAttemptAt) + status.LastError = snapshot.LastError + } + applyDaemonRefreshHealth(status, snapshot, refresh) + return self, peers, status +} + +func applyDaemonRefreshHealth(status *daemon.SnapshotStatus, snapshot *engine.MeshSnapshot, refresh *engine.RefreshCoordinatorHealth) { + if refresh == nil { + return + } + + status.State = daemonRefreshState(snapshot, refresh) + status.Mode = string(refresh.Mode) + status.InFlight = refresh.InFlight + status.Pending = daemonRefreshReasons(refresh.PendingReasons) + status.Paused = refresh.Paused + status.LastReasons = daemonRefreshReasons(refresh.LastReasons) + status.LastDurationMS = refresh.LastDuration.Milliseconds() + status.ConsecutiveFailures = refresh.ConsecutiveFailures + status.LastSuccessAt = daemonSnapshotTimestamp(refresh.LastSuccessAt) + status.RetryNotBefore = daemonSnapshotTimestamp(refresh.RetryNotBefore) + status.NextAttemptAt = daemonSnapshotTimestamp(refresh.NextAttemptAt) + if !refresh.LastAttemptAt.IsZero() { + status.LastAttemptAt = daemonSnapshotTimestamp(refresh.LastAttemptAt) + } + if status.LastError == "" { + status.LastError = refresh.LastError + } + if len(refresh.Streams) != 0 { + status.Streams = make(map[string]daemon.SnapshotStreamStatus, len(refresh.Streams)) + for stream, health := range refresh.Streams { + status.Streams[string(stream)] = daemon.SnapshotStreamStatus{ + Covered: health.Covered, + Live: health.Live, + Repaired: health.Repaired, + } + } + } +} + +func daemonRefreshState(snapshot *engine.MeshSnapshot, refresh *engine.RefreshCoordinatorHealth) string { + hasSuccessfulRefresh := !refresh.LastSuccessAt.IsZero() || (snapshot != nil && snapshot.Generation > 0) + if !hasSuccessfulRefresh { + if refresh.ConsecutiveFailures > 0 || refresh.LastError != "" || refresh.Paused { + return "degraded" + } + return "starting" + } + if !refresh.Running || refresh.Paused || !refresh.StreamsHealthy || refresh.ConsecutiveFailures > 0 || refresh.LastError != "" { + return "degraded" + } + return "healthy" +} + +func daemonRefreshReasons(reasons []engine.RefreshReason) []string { + if len(reasons) == 0 { + return nil } - return self, daemonPeerStatuses(snapshot.Peers), &daemon.SnapshotStatus{ - Generation: snapshot.Generation, - RefreshedAt: daemonSnapshotTimestamp(snapshot.RefreshedAt), - LastAttemptAt: daemonSnapshotTimestamp(snapshot.LastAttemptAt), - LastError: snapshot.LastError, + result := make([]string, len(reasons)) + for i, reason := range reasons { + result[i] = string(reason) } + return result } func daemonPeerStatuses(peers []engine.PeerSnapshot) []daemon.PeerStatus { diff --git a/cmd/meshd/main_test.go b/cmd/meshd/main_test.go index 3d6f8a2..98d6aa3 100644 --- a/cmd/meshd/main_test.go +++ b/cmd/meshd/main_test.go @@ -821,6 +821,17 @@ func TestParseUpFlagsForeground(t *testing.T) { } } +func TestParseUpFlagsPollIntervalOverride(t *testing.T) { + if got := parseUpFlags([]string{"--poll-interval", "2m"}).pollInterval; got != 2*time.Minute { + t.Fatalf("poll interval = %v, want 2m", got) + } + for _, value := range []string{"0s", "-1s", "invalid"} { + if got := parseUpFlags([]string{"--poll-interval", value}).pollInterval; got != 0 { + t.Fatalf("poll interval for %q = %v, want unset", value, got) + } + } +} + func TestShouldReexecWithSudoForTun(t *testing.T) { t.Setenv(trayConnectEnv, "") if !shouldReexecWithSudoForTun(upFlags{}, 501, "darwin") { diff --git a/cmd/meshd/peer_list_local_test.go b/cmd/meshd/peer_list_local_test.go index e61838d..a01b354 100644 --- a/cmd/meshd/peer_list_local_test.go +++ b/cmd/meshd/peer_list_local_test.go @@ -219,7 +219,7 @@ func TestDaemonStatusesFromMeshSnapshot(t *testing.T) { }}, } - self, peers, freshness := daemonStatusesFromMeshSnapshot(snapshot) + self, peers, freshness := daemonStatusesFromMeshSnapshot(snapshot, nil) if self == nil { t.Fatal("self status = nil") } @@ -240,11 +240,121 @@ func TestDaemonStatusesFromMeshSnapshot(t *testing.T) { t.Fatalf("freshness = %+v", freshness) } - if self, peers, freshness := daemonStatusesFromMeshSnapshot(nil); self != nil || peers != nil || freshness != nil { + if self, peers, freshness := daemonStatusesFromMeshSnapshot(nil, nil); self != nil || peers != nil || freshness != nil { t.Fatalf("nil snapshot = (%+v, %+v, %+v), want nils", self, peers, freshness) } } +func TestDaemonRefreshHealthMapping(t *testing.T) { + base := time.Date(2026, 7, 11, 12, 0, 0, 987654321, time.FixedZone("offset", -7*60*60)) + snapshot := &engine.MeshSnapshot{ + Generation: 12, + RefreshedAt: base, + LastAttemptAt: base, + } + health := &engine.RefreshCoordinatorHealth{ + Running: true, + Paused: true, + InFlight: true, + Mode: engine.RefreshModeFallback, + StreamsHealthy: false, + Streams: map[engine.RefreshStream]engine.RefreshStreamHealth{ + engine.RefreshStreamTopology: {Covered: true, Live: true, Repaired: true}, + engine.RefreshStreamDelivery: {Covered: true, Live: false, Repaired: false}, + }, + PendingReasons: []engine.RefreshReason{engine.RefreshReasonDelivery, engine.RefreshReasonEndpoint}, + ConsecutiveFailures: 2, + LastAttemptAt: base.Add(2 * time.Minute), + LastSuccessAt: base.Add(-time.Minute), + LastReasons: []engine.RefreshReason{engine.RefreshReasonTopology}, + LastDuration: 1500*time.Millisecond + 250*time.Microsecond, + LastError: "refresh failed", + RetryNotBefore: base.Add(3 * time.Minute), + NextAttemptAt: base.Add(3 * time.Minute), + } + + _, _, got := daemonStatusesFromMeshSnapshot(snapshot, health) + if got == nil { + t.Fatal("snapshot status = nil") + } + if got.State != "degraded" || got.Mode != "fallback" || !got.InFlight || !got.Paused { + t.Fatalf("scheduler state = %+v", got) + } + if !reflect.DeepEqual(got.Pending, []string{"delivery", "endpoint"}) || + !reflect.DeepEqual(got.LastReasons, []string{"topology"}) { + t.Fatalf("refresh reasons = pending %v, last %v", got.Pending, got.LastReasons) + } + if got.LastDurationMS != 1500 || got.ConsecutiveFailures != 2 || got.LastError != "refresh failed" { + t.Fatalf("refresh failure status = %+v", got) + } + if got.LastAttemptAt != health.LastAttemptAt.UTC().Format(time.RFC3339Nano) || + got.LastSuccessAt != health.LastSuccessAt.UTC().Format(time.RFC3339Nano) || + got.RetryNotBefore != health.RetryNotBefore.UTC().Format(time.RFC3339Nano) || + got.NextAttemptAt != health.NextAttemptAt.UTC().Format(time.RFC3339Nano) { + t.Fatalf("refresh timestamps = %+v", got) + } + if len(got.Streams) != 2 || !got.Streams["topology"].Covered || !got.Streams["topology"].Live || + !got.Streams["topology"].Repaired || !got.Streams["delivery"].Covered || got.Streams["delivery"].Live || + got.Streams["delivery"].Repaired { + t.Fatalf("stream status = %+v", got.Streams) + } +} + +func TestDaemonRefreshState(t *testing.T) { + tests := []struct { + name string + snapshot *engine.MeshSnapshot + health engine.RefreshCoordinatorHealth + want string + }{ + { + name: "starting before first success", + health: engine.RefreshCoordinatorHealth{ + Running: true, + Mode: engine.RefreshModeFallback, + }, + want: "starting", + }, + { + name: "healthy after success with repaired streams", + snapshot: &engine.MeshSnapshot{Generation: 1}, + health: engine.RefreshCoordinatorHealth{ + Running: true, + Mode: engine.RefreshModeHealthy, + StreamsHealthy: true, + }, + want: "healthy", + }, + { + name: "degraded while subscriptions use fallback", + snapshot: &engine.MeshSnapshot{Generation: 1}, + health: engine.RefreshCoordinatorHealth{ + Running: true, + Mode: engine.RefreshModeFallback, + StreamsHealthy: false, + }, + want: "degraded", + }, + { + name: "degraded when initial refresh fails", + health: engine.RefreshCoordinatorHealth{ + Running: true, + ConsecutiveFailures: 1, + LastError: "unavailable", + }, + want: "degraded", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := daemonRefreshState(tc.snapshot, &tc.health); got != tc.want { + t.Fatalf("daemonRefreshState() = %q, want %q", got, tc.want) + } + }) + } +} + func TestCmdPeerListUsesDaemonSnapshotBeforeIdentity(t *testing.T) { stateDir := t.TempDir() t.Setenv("MESHD_STATE_DIR", stateDir) diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index ea1ab58..64b7fda 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -84,10 +84,32 @@ type PeerStatus struct { // SnapshotStatus describes the freshness of the daemon's materialized mesh // snapshot. Timestamp fields are encoded as RFC3339Nano strings. type SnapshotStatus struct { - Generation uint64 `json:"generation"` - RefreshedAt string `json:"refreshedAt,omitempty"` - LastAttemptAt string `json:"lastAttemptAt,omitempty"` - LastError string `json:"lastError,omitempty"` + Generation uint64 `json:"generation"` + RefreshedAt string `json:"refreshedAt,omitempty"` + LastAttemptAt string `json:"lastAttemptAt,omitempty"` + LastError string `json:"lastError,omitempty"` + State string `json:"state,omitempty"` + Mode string `json:"mode,omitempty"` + InFlight bool `json:"inFlight,omitempty"` + Pending []string `json:"pending,omitempty"` + Paused bool `json:"paused,omitempty"` + LastReasons []string `json:"lastReasons,omitempty"` + LastDurationMS int64 `json:"lastDurationMs,omitempty"` + ConsecutiveFailures int `json:"consecutiveFailures,omitempty"` + LastSuccessAt string `json:"lastSuccessAt,omitempty"` + RetryNotBefore string `json:"retryNotBefore,omitempty"` + NextAttemptAt string `json:"nextAttemptAt,omitempty"` + Streams map[string]SnapshotStreamStatus `json:"streams,omitempty"` +} + +// SnapshotStreamStatus describes whether one invalidation stream is +// authorized, connected, and repaired through its end-of-stored-events +// barrier. False values are intentionally retained when the stream is present +// so operators can distinguish missing coverage from a legacy daemon. +type SnapshotStreamStatus struct { + Covered bool `json:"covered"` + Live bool `json:"live"` + Repaired bool `json:"repaired"` } type peerAuthorizedContextKey struct{} diff --git a/internal/daemon/status_snapshot_test.go b/internal/daemon/status_snapshot_test.go index 6da6085..8243160 100644 --- a/internal/daemon/status_snapshot_test.go +++ b/internal/daemon/status_snapshot_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "net" "net/http" + "reflect" "testing" "time" ) @@ -34,10 +35,25 @@ func TestStatusEndpointRichSnapshot(t *testing.T) { LastSeen: "2026-07-11T17:40:06.987654321Z", } wantSnapshot := SnapshotStatus{ - Generation: 42, - RefreshedAt: "2026-07-11T17:40:06.987654321Z", - LastAttemptAt: "2026-07-11T17:40:07.123456789Z", - LastError: "refresh deferred by rate limit", + Generation: 42, + RefreshedAt: "2026-07-11T17:40:06.987654321Z", + LastAttemptAt: "2026-07-11T17:40:07.123456789Z", + LastError: "refresh deferred by rate limit", + State: "degraded", + Mode: "fallback", + InFlight: true, + Pending: []string{"delivery", "endpoint"}, + Paused: true, + LastReasons: []string{"topology"}, + LastDurationMS: 1250, + ConsecutiveFailures: 3, + LastSuccessAt: "2026-07-11T17:39:06.987654321Z", + RetryNotBefore: "2026-07-11T17:40:09.123456789Z", + NextAttemptAt: "2026-07-11T17:40:09.123456789Z", + Streams: map[string]SnapshotStreamStatus{ + "delivery": {Covered: true, Live: false, Repaired: false}, + "topology": {Covered: true, Live: true, Repaired: true}, + }, } srv := NewServer(sock, func() Status { @@ -68,7 +84,7 @@ func TestStatusEndpointRichSnapshot(t *testing.T) { if status.Self == nil || *status.Self != wantSelf { t.Fatalf("Self = %+v, want %+v", status.Self, wantSelf) } - if status.Snapshot == nil || *status.Snapshot != wantSnapshot { + if status.Snapshot == nil || !reflect.DeepEqual(*status.Snapshot, wantSnapshot) { t.Fatalf("Snapshot = %+v, want %+v", status.Snapshot, wantSnapshot) } } @@ -113,6 +129,23 @@ func TestClientGetStatusLegacyResponse(t *testing.T) { } } +func TestSnapshotStatusLegacyJSONShape(t *testing.T) { + snapshot := SnapshotStatus{ + Generation: 7, + RefreshedAt: "2026-07-11T17:40:06Z", + LastAttemptAt: "2026-07-11T17:40:07Z", + LastError: "temporary failure", + } + got, err := json.Marshal(snapshot) + if err != nil { + t.Fatalf("Marshal() error: %v", err) + } + want := `{"generation":7,"refreshedAt":"2026-07-11T17:40:06Z","lastAttemptAt":"2026-07-11T17:40:07Z","lastError":"temporary failure"}` + if string(got) != want { + t.Fatalf("legacy SnapshotStatus JSON = %s, want %s", got, want) + } +} + func TestPeerStatusLegacyJSONShape(t *testing.T) { peer := PeerStatus{Name: "peer-a", MeshIP: "10.200.0.2", Online: true} got, err := json.Marshal(peer) diff --git a/internal/engine/convert.go b/internal/engine/convert.go index 528c543..c5b46d7 100644 --- a/internal/engine/convert.go +++ b/internal/engine/convert.go @@ -11,6 +11,7 @@ import ( "log/slog" "net/netip" "sort" + "time" "github.com/enboxorg/meshd/internal/control" @@ -183,6 +184,19 @@ func (c *Converter) convertNode(n *control.Node) (*tailcfg.Node, error) { node.LastSeen = &lastSeen } + if n.ExpiresAt != "" { + keyExpiry, err := time.Parse(time.RFC3339Nano, n.ExpiresAt) + if err != nil { + c.logger.Debug("skipping invalid node expiry", + slog.String("node", n.Name), + slog.String("expiresAt", n.ExpiresAt), + slog.Any("error", err), + ) + } else { + node.KeyExpiry = keyExpiry + } + } + // Parse WireGuard public key. The key is stored as base64 in DWN records. if n.Key != "" { nodeKey, err := parseWireGuardKey(n.Key) diff --git a/internal/engine/convert_test.go b/internal/engine/convert_test.go index 7081448..fd97f8c 100644 --- a/internal/engine/convert_test.go +++ b/internal/engine/convert_test.go @@ -4,11 +4,51 @@ import ( "encoding/base64" "net/netip" "testing" + "time" "github.com/enboxorg/meshd/internal/control" "github.com/enboxorg/meshnet/types/key" ) +func TestConvertNodeKeyExpiry(t *testing.T) { + standard := time.Date(2026, 8, 1, 0, 0, 0, 0, time.UTC) + nanosecond := time.Date(2026, 8, 1, 2, 30, 0, 123456789, time.FixedZone("UTC+02:30", 2*60*60+30*60)) + + tests := map[string]struct { + expiresAt string + want time.Time + }{ + "RFC3339": { + expiresAt: "2026-08-01T00:00:00Z", + want: standard, + }, + "RFC3339Nano": { + expiresAt: "2026-08-01T02:30:00.123456789+02:30", + want: nanosecond, + }, + "empty": {}, + "invalid": { + expiresAt: "not-a-timestamp", + }, + } + + converter := NewConverter("test") + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + node, err := converter.convertNode(&control.Node{ + Name: "peer", + ExpiresAt: tc.expiresAt, + }) + if err != nil { + t.Fatalf("convertNode: %v", err) + } + if !node.KeyExpiry.Equal(tc.want) { + t.Fatalf("KeyExpiry = %v, want %v", node.KeyExpiry, tc.want) + } + }) + } +} + // testWireGuardKey returns a valid base64-encoded 32-byte key for testing. func testWireGuardKey() string { // Generate a real key pair so we have a valid 32-byte public key. diff --git a/internal/engine/dwncontrol.go b/internal/engine/dwncontrol.go index adb1b95..746c34a 100644 --- a/internal/engine/dwncontrol.go +++ b/internal/engine/dwncontrol.go @@ -12,8 +12,10 @@ package engine import ( "context" + "fmt" "log" "math/rand/v2" + "slices" "sync" "sync/atomic" "time" @@ -35,10 +37,7 @@ type DWNControlConfig struct { // The DWN control integration implements this by reading DWN records // and building a NetworkMap. // - // This is called: - // - Once on Login (initial state load) - // - Periodically at PollInterval - // - When explicitly triggered via Notify() + // Calls are serialized and coalesced by RefreshCoordinator. MapResponseFunc func(ctx context.Context) (*netmap.NetworkMap, error) // OnMapResult observes every completed map load. It is called after the @@ -55,10 +54,21 @@ type DWNControlConfig struct { // If nil, endpoint updates are not published. EndpointUpdateFunc func(ctx context.Context, endpoints []tailcfg.Endpoint) - // PollInterval is how often to re-read DWN state. - // Default: 30 seconds. + // PollInterval is the fallback reconciliation interval while authoritative + // subscription coverage is incomplete or unhealthy. Default: 30 seconds. PollInterval time.Duration + // HealthyPollInterval is the slow anti-entropy interval while topology and + // delivery streams are both live and repaired. Default: 5 minutes. + HealthyPollInterval time.Duration + + // RefreshTimeout bounds one complete DWN state rebuild. Default: 2 minutes. + RefreshTimeout time.Duration + + // StartupSubscriptionWait gives asynchronous subscription handshakes a + // bounded chance to establish before the initial full rebuild. + StartupSubscriptionWait time.Duration + // NodePrivateKey, if set, overrides the auto-generated WireGuard node // key with the given key. This is essential when the WireGuard key has // already been published to a coordination store (e.g. DWN records) and @@ -118,13 +128,12 @@ type DWNControl struct { logf logger.Logf clientID int64 - mu sync.Mutex - disco key.DiscoPublic - paused atomic.Bool - shutdown chan struct{} - shutdownOnce sync.Once - cancel context.CancelFunc - notify chan struct{} // signal to re-poll immediately + mu sync.Mutex + disco key.DiscoPublic + shutdownOnce sync.Once + cancel context.CancelFunc + coordinator *RefreshCoordinator + initialMapApplied atomic.Bool } // NewDWNControlFactory returns a factory function suitable for @@ -139,6 +148,15 @@ func NewDWNControlFactory(config *DWNControlConfig) func(controlclient.Options) // NewDWNControl creates a new DWN-backed control client. func NewDWNControl(config *DWNControlConfig, opts controlclient.Options) (*DWNControl, error) { + if config == nil { + return nil, fmt.Errorf("DWN control config is required") + } + if config.MapResponseFunc == nil { + return nil, fmt.Errorf("DWN control map response function is required") + } + if config.RefreshTimeout < 0 || config.StartupSubscriptionWait < 0 { + return nil, fmt.Errorf("DWN control timeouts must not be negative") + } logf := config.Logf if logf == nil { logf = log.Printf @@ -148,6 +166,14 @@ func NewDWNControl(config *DWNControlConfig, opts controlclient.Options) (*DWNCo if pollInterval == 0 { pollInterval = 30 * time.Second } + healthyPollInterval := config.HealthyPollInterval + if healthyPollInterval == 0 { + healthyPollInterval = 5 * time.Minute + } + refreshTimeout := config.RefreshTimeout + if refreshTimeout == 0 { + refreshTimeout = 2 * time.Minute + } p := opts.Persist.Clone() if p == nil { @@ -167,22 +193,37 @@ func NewDWNControl(config *DWNControlConfig, opts controlclient.Options) (*DWNCo cc := &DWNControl{ config: DWNControlConfig{ - MapResponseFunc: config.MapResponseFunc, - OnMapResult: config.OnMapResult, - EndpointUpdateFunc: config.EndpointUpdateFunc, - PollInterval: pollInterval, - NodePrivateKey: config.NodePrivateKey, - DiscoKeyRegistry: config.DiscoKeyRegistry, - Logf: logf, + MapResponseFunc: config.MapResponseFunc, + OnMapResult: config.OnMapResult, + EndpointUpdateFunc: config.EndpointUpdateFunc, + PollInterval: pollInterval, + HealthyPollInterval: healthyPollInterval, + RefreshTimeout: refreshTimeout, + StartupSubscriptionWait: config.StartupSubscriptionWait, + NodePrivateKey: config.NodePrivateKey, + DiscoKeyRegistry: config.DiscoKeyRegistry, + Logf: logf, }, observer: opts.Observer, persist: p, logf: logf, clientID: rand.Int64(), - shutdown: make(chan struct{}), cancel: cancel, - notify: make(chan struct{}, 1), } + coordinator, err := NewRefreshCoordinator(RefreshCoordinatorConfig{ + Refresh: cc.refreshControlState, + FallbackInterval: pollInterval, + HealthyInterval: healthyPollInterval, + RetryBackoff: time.Second, + Debounce: 250 * time.Millisecond, + MaxDebounce: time.Second, + MaxRetryBackoff: time.Minute, + }) + if err != nil { + cancel() + return nil, err + } + cc.coordinator = coordinator cc.disco = opts.DiscoPublicKey @@ -208,85 +249,88 @@ func NewDWNControl(config *DWNControlConfig, opts controlclient.Options) (*DWNCo return cc, nil } -// pollLoop periodically reads DWN state and pushes NetworkMap updates. +// pollLoop gives the LocalBackend event bus time to initialize, then starts +// the single-owner refresh coordinator. func (cc *DWNControl) pollLoop(ctx context.Context) { - // Brief startup delay to let the LocalBackend's event bus fully - // initialize before we send the first NetworkMap. Without this, the - // first SetControlClientStatus triggers authReconfigLocked which calls - // Reconfig → ParseEndpoint before magicsock has processed the - // NodeViewsUpdate event, causing "unknown peer" errors. select { case <-ctx.Done(): return case <-time.After(50 * time.Millisecond): } + cc.waitForStartupSubscriptions(ctx, cc.config.StartupSubscriptionWait) + if err := cc.coordinator.Start(ctx); err != nil && ctx.Err() == nil { + cc.logf("dwn-control: starting refresh coordinator: %v", err) + } +} - // Initial login: load state and report logged-in. - cc.loadAndPush(ctx) - - // Allow the event bus to propagate the initial NodeViewsUpdate to - // magicsock before any subsequent Reconfig from the notify feedback - // loop. The first loadAndPush publishes the netmap to the event bus - // (via setNetMapLocked), but authReconfigLocked runs before magicsock - // processes it. This sleep gives the event bus time to deliver. - select { - case <-ctx.Done(): +func (cc *DWNControl) waitForStartupSubscriptions(ctx context.Context, maximum time.Duration) { + if maximum <= 0 { return - case <-time.After(200 * time.Millisecond): - } - // Drain any queued notifications from the startup burst. - select { - case <-cc.notify: - default: } - // Re-push now that magicsock knows about our peers. - cc.loadAndPush(ctx) - - ticker := time.NewTicker(cc.config.PollInterval) - defer ticker.Stop() - - // Minimum interval between polls to avoid a feedback loop where - // SetControlClientStatus triggers Login/Notify, causing another - // loadAndPush immediately. - const minPollInterval = 250 * time.Millisecond - lastPoll := time.Now() - + deadline := time.NewTimer(maximum) + defer deadline.Stop() + poll := time.NewTicker(25 * time.Millisecond) + defer poll.Stop() for { + if refreshStreamsReadyForStartup(cc.coordinator.Health()) { + return + } select { case <-ctx.Done(): return - case <-cc.shutdown: + case <-deadline.C: return - case <-ticker.C: - if cc.paused.Load() { - continue - } - cc.loadAndPush(ctx) - lastPoll = time.Now() - case <-cc.notify: - if cc.paused.Load() { - continue - } - // Debounce: ensure minimum interval between polls. - if elapsed := time.Since(lastPoll); elapsed < minPollInterval { - time.Sleep(minPollInterval - elapsed) - } - // Drain any queued notifications that accumulated during the sleep. - select { - case <-cc.notify: - default: - } - cc.loadAndPush(ctx) - lastPoll = time.Now() + case <-poll.C: + } + } +} + +func refreshStreamsReadyForStartup(health RefreshCoordinatorHealth) bool { + covered := 0 + for _, stream := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} { + state := health.Streams[stream] + if !state.Covered { + continue } + covered++ + if !state.Live { + return false + } + } + return covered > 0 +} + +// refreshControlState performs one bounded remote rebuild. The first usable +// map is applied locally a second time after the event bus has observed its peer +// views; this preserves the startup ordering workaround without a second DWN load. +func (cc *DWNControl) refreshControlState(ctx context.Context, _ RefreshBatch) error { + refreshCtx := ctx + cancel := func() {} + if cc.config.RefreshTimeout > 0 { + refreshCtx, cancel = context.WithTimeout(ctx, cc.config.RefreshTimeout) + } + defer cancel() + + replay, err := cc.loadAndPush(refreshCtx) + if err != nil { + return err + } + if replay == nil || cc.observer == nil || !cc.initialMapApplied.CompareAndSwap(false, true) { + return nil } + if !waitForRefreshContext(refreshCtx, 200*time.Millisecond) { + return refreshCtx.Err() + } + cc.pushNetMap(replay) + return nil } -// loadAndPush reads DWN state and pushes a Status to the observer. -func (cc *DWNControl) loadAndPush(ctx context.Context) { +// loadAndPush reads DWN state, applies it once, and returns an observer-safe +// clone for the one-time startup replay. +func (cc *DWNControl) loadAndPush(ctx context.Context) (*netmap.NetworkMap, error) { if cc.config.MapResponseFunc == nil { cc.logf("dwn-control: no MapResponseFunc configured") - return + return nil, nil } nm, err := cc.config.MapResponseFunc(ctx) @@ -301,69 +345,94 @@ func (cc *DWNControl) loadAndPush(ctx context.Context) { if cc.config.OnMapResult != nil { cc.config.OnMapResult(ctx, nil, err) } - return + return nil, err } if nm == nil { if cc.config.OnMapResult != nil { cc.config.OnMapResult(ctx, nil, nil) } - return + return nil, nil } - // Inject our node key into the NetworkMap. - nm.NodeKey = cc.persist.PrivateNodeKey.Public() + cc.prepareNetMap(nm) + replay := cloneNetMapForReplay(nm) + cc.pushNetMap(nm) + if cc.config.OnMapResult != nil { + cc.config.OnMapResult(ctx, nm, nil) + } + return replay, nil +} - // If a disco key registry is available, inject disco keys into the - // network map. This replaces the role of the Tailscale control server - // which normally distributes disco keys to peers. - if cc.config.DiscoKeyRegistry != nil { - // Inject our own disco key into SelfNode. - cc.mu.Lock() - selfDisco := cc.disco - cc.mu.Unlock() - if !selfDisco.IsZero() && nm.SelfNode.Valid() { - sn := nm.SelfNode.AsStruct() - sn.DiscoKey = selfDisco - nm.SelfNode = sn.View() +func (cc *DWNControl) prepareNetMap(nm *netmap.NetworkMap) { + nm.NodeKey = cc.persist.PrivateNodeKey.Public() + if cc.config.DiscoKeyRegistry == nil { + return + } + cc.mu.Lock() + selfDisco := cc.disco + cc.mu.Unlock() + if !selfDisco.IsZero() && nm.SelfNode.Valid() { + sn := nm.SelfNode.AsStruct() + sn.DiscoKey = selfDisco + nm.SelfNode = sn.View() + } + for i, peer := range nm.Peers { + if peer.Key().IsZero() { + continue } - // Inject peer disco keys. - for i, p := range nm.Peers { - if !p.Key().IsZero() { - dk := cc.config.DiscoKeyRegistry.GetDisco(p.Key()) - if !dk.IsZero() { - ps := p.AsStruct() - ps.DiscoKey = dk - nm.Peers[i] = ps.View() - } - } + disco := cc.config.DiscoKeyRegistry.GetDisco(peer.Key()) + if disco.IsZero() { + continue } + copy := peer.AsStruct() + copy.DiscoKey = disco + nm.Peers[i] = copy.View() } +} - if cc.observer != nil { - s := controlclient.Status{ - LoggedIn: true, - InMapPoll: true, - NetMap: nm, - Persist: cc.persist.View(), - } - cc.observer.SetControlClientStatus(cc, s) +func (cc *DWNControl) pushNetMap(nm *netmap.NetworkMap) { + if cc.observer == nil || nm == nil { + return } - if cc.config.OnMapResult != nil { - cc.config.OnMapResult(ctx, nm, nil) + cc.observer.SetControlClientStatus(cc, controlclient.Status{ + LoggedIn: true, + InMapPoll: true, + NetMap: nm, + Persist: cc.persist.View(), + }) +} + +func cloneNetMapForReplay(nm *netmap.NetworkMap) *netmap.NetworkMap { + if nm == nil { + return nil } + clone := *nm + clone.Peers = slices.Clone(nm.Peers) + return &clone } -// Notify triggers an immediate re-read of DWN state. -// This is called by the DWN subscription handler when records change. -func (cc *DWNControl) Notify() { +func waitForRefreshContext(ctx context.Context, delay time.Duration) bool { + timer := time.NewTimer(delay) + defer timer.Stop() select { - case cc.notify <- struct{}{}: - default: - // Already a pending notification. + case <-ctx.Done(): + return false + case <-timer.C: + return true } } +// Notify records a manual invalidation without blocking the caller. +func (cc *DWNControl) Notify() { + cc.coordinator.Notify(RefreshReasonManual) +} + +// RefreshHealth returns a deep-copied view of refresh scheduling and stream health. +func (cc *DWNControl) RefreshHealth() RefreshCoordinatorHealth { + return cc.coordinator.Health() +} + // // --- controlclient.Client interface --- // @@ -371,14 +440,14 @@ func (cc *DWNControl) Notify() { func (cc *DWNControl) Shutdown() { cc.shutdownOnce.Do(func() { cc.cancel() - close(cc.shutdown) + cc.coordinator.Stop() }) } func (cc *DWNControl) Login(flags controlclient.LoginFlags) { - // DWN doesn't need interactive login — the DID is the identity. - // Trigger an immediate state load which will report LoggedIn. - cc.Notify() + // DWN has no interactive login. The coordinator always queues startup, so + // treating LocalBackend login feedback as another invalidation would cause + // a redundant rebuild after every successful map application. } func (cc *DWNControl) Logout(ctx context.Context) error { @@ -392,11 +461,7 @@ func (cc *DWNControl) Logout(ctx context.Context) error { } func (cc *DWNControl) SetPaused(paused bool) { - cc.paused.Store(paused) - if !paused { - // Unpause: trigger immediate refresh. - cc.Notify() - } + cc.coordinator.SetPaused(paused) } func (cc *DWNControl) AuthCantContinue() bool { @@ -422,8 +487,8 @@ func (cc *DWNControl) UpdateEndpoints(endpoints []tailcfg.Endpoint) { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() cc.config.EndpointUpdateFunc(ctx, endpoints) - // Trigger re-poll so peers pick up the new endpoints. - cc.Notify() + // Coalesce the local write with its subscription echo. + cc.coordinator.Notify(RefreshReasonEndpoint) }() } } diff --git a/internal/engine/dwncontrol_coordinator_test.go b/internal/engine/dwncontrol_coordinator_test.go new file mode 100644 index 0000000..1d72601 --- /dev/null +++ b/internal/engine/dwncontrol_coordinator_test.go @@ -0,0 +1,601 @@ +package engine + +import ( + "context" + "errors" + "reflect" + "sync/atomic" + "testing" + "time" + + "github.com/enboxorg/meshnet/control/controlclient" + "github.com/enboxorg/meshnet/tailcfg" + "github.com/enboxorg/meshnet/types/netmap" +) + +func TestDWNControlRejectsInvalidCoordinatorConfig(t *testing.T) { + valid := func() *DWNControlConfig { + return &DWNControlConfig{ + MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { return nil, nil }, + } + } + tests := []struct { + name string + cfg func() *DWNControlConfig + }{ + {name: "nil config", cfg: func() *DWNControlConfig { return nil }}, + {name: "missing loader", cfg: func() *DWNControlConfig { return &DWNControlConfig{} }}, + {name: "negative refresh timeout", cfg: func() *DWNControlConfig { + cfg := valid() + cfg.RefreshTimeout = -time.Second + return cfg + }}, + {name: "negative startup wait", cfg: func() *DWNControlConfig { + cfg := valid() + cfg.StartupSubscriptionWait = -time.Second + return cfg + }}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if cc, err := NewDWNControl(tc.cfg(), controlclient.Options{SkipStartForTests: true}); err == nil { + cc.Shutdown() + t.Fatal("NewDWNControl unexpectedly succeeded") + } + }) + } +} + +func TestDWNControlCoordinatorFirstSuccessReplaysLoadedMap(t *testing.T) { + var loads atomic.Int32 + var mapResults atomic.Int32 + var applies atomic.Int32 + peerValid := make(chan bool, 2) + + observer := dwnControlObserverFunc(func(client controlclient.Client, status controlclient.Status) { + if status.NetMap == nil { + return + } + call := applies.Add(1) + valid := len(status.NetMap.Peers) == 1 && status.NetMap.Peers[0].Valid() + if call == 1 { + // Mutate the backing slice, not just its header. The startup replay + // must own an independent slice captured before this callback. + status.NetMap.Peers[0] = tailcfg.NodeView{} + } + peerValid <- valid + // LocalBackend feeds Login back to a control client while applying + // status. DWN login is intentionally a no-op: startup already queued + // the load, and feedback must not create a trailing remote rebuild. + client.Login(0) + }) + + cc, err := NewDWNControl(&DWNControlConfig{ + MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { + loads.Add(1) + return routingTestNetMap(), nil + }, + OnMapResult: func(context.Context, *netmap.NetworkMap, error) { + mapResults.Add(1) + }, + PollInterval: time.Hour, + Logf: func(string, ...any) {}, + }, controlclient.Options{Observer: observer, SkipStartForTests: true}) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + defer cc.Shutdown() + + if err := cc.coordinator.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + waitForDWNControlCondition(t, func() bool { + health := cc.RefreshHealth() + return !health.InFlight && len(health.PendingReasons) == 0 && !health.LastSuccessAt.IsZero() + }) + if got := loads.Load(); got != 1 { + t.Fatalf("remote loads = %d, want 1", got) + } + if got := applies.Load(); got != 2 { + t.Fatalf("observer applies = %d, want 2", got) + } + if got := mapResults.Load(); got != 1 { + t.Fatalf("map-result callbacks = %d, want 1", got) + } + for apply := 1; apply <= 2; apply++ { + select { + case valid := <-peerValid: + if !valid { + t.Fatalf("apply %d received mutated peer slice", apply) + } + case <-time.After(time.Second): + t.Fatalf("timed out waiting for observer apply %d", apply) + } + } +} + +func TestDWNControlCoordinatorCancellationSuppressesReplay(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + var applies atomic.Int32 + var mapResults atomic.Int32 + observer := dwnControlObserverFunc(func(_ controlclient.Client, status controlclient.Status) { + if status.NetMap != nil { + applies.Add(1) + cancel() + } + }) + + cc, err := NewDWNControl(&DWNControlConfig{ + MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { + return routingTestNetMap(), nil + }, + OnMapResult: func(context.Context, *netmap.NetworkMap, error) { + mapResults.Add(1) + }, + PollInterval: time.Hour, + Logf: func(string, ...any) {}, + }, controlclient.Options{Observer: observer, SkipStartForTests: true}) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + defer cc.Shutdown() + + err = cc.refreshControlState(ctx, RefreshBatch{}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("refreshControlState error = %v, want context canceled", err) + } + if got := applies.Load(); got != 1 { + t.Fatalf("observer applies = %d, want only the initial apply", got) + } + if got := mapResults.Load(); got != 1 { + t.Fatalf("map-result callbacks = %d, want 1", got) + } +} + +func TestDWNControlCoordinatorRetriesDirtyFirstLoadThenReplaysSuccess(t *testing.T) { + loadErr := errors.New("initial DWN load failed") + var loads atomic.Int32 + var mapResults atomic.Int32 + statuses := make(chan controlclient.Status, 3) + observer := dwnControlObserverFunc(func(_ controlclient.Client, status controlclient.Status) { + statuses <- status + }) + + cc, err := NewDWNControl(&DWNControlConfig{ + MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { + if loads.Add(1) == 1 { + return nil, loadErr + } + return routingTestNetMap(), nil + }, + OnMapResult: func(context.Context, *netmap.NetworkMap, error) { + mapResults.Add(1) + }, + PollInterval: time.Hour, + Logf: func(string, ...any) {}, + }, controlclient.Options{Observer: observer, SkipStartForTests: true}) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + defer cc.Shutdown() + + clock := newCoordinatorFakeClock() + cc.coordinator, err = NewRefreshCoordinator(RefreshCoordinatorConfig{ + Refresh: cc.refreshControlState, + FallbackInterval: time.Hour, + HealthyInterval: 2 * time.Hour, + RetryBackoff: time.Second, + MaxRetryBackoff: time.Second, + Jitter: identityRefreshJitter, + RetryJitter: identityRefreshJitter, + Clock: clock, + }) + if err != nil { + t.Fatalf("NewRefreshCoordinator: %v", err) + } + if err := cc.coordinator.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + + first := receiveDWNControlStatus(t, statuses) + if !errors.Is(first.Err, loadErr) { + t.Fatalf("first observer error = %v, want %v", first.Err, loadErr) + } + waitForDWNControlCondition(t, func() bool { + return cc.RefreshHealth().ConsecutiveFailures == 1 + }) + health := cc.RefreshHealth() + if !reflect.DeepEqual(health.PendingReasons, []RefreshReason{RefreshReasonStartup}) { + t.Fatalf("pending reasons after failure = %v, want startup", health.PendingReasons) + } + if !health.RetryNotBefore.Equal(clock.Now().Add(time.Second)) { + t.Fatalf("retry deadline = %v, want %v", health.RetryNotBefore, clock.Now().Add(time.Second)) + } + + clock.Advance(time.Second) + for apply := 1; apply <= 2; apply++ { + status := receiveDWNControlStatus(t, statuses) + if status.Err != nil || status.NetMap == nil { + t.Fatalf("successful apply %d = %+v", apply, status) + } + } + waitForDWNControlCondition(t, func() bool { + health := cc.RefreshHealth() + return !health.InFlight && health.ConsecutiveFailures == 0 && health.LastError == "" + }) + if got := loads.Load(); got != 2 { + t.Fatalf("remote loads = %d, want 2", got) + } + if got := mapResults.Load(); got != 2 { + t.Fatalf("map-result callbacks = %d, want 2", got) + } + if got := cc.RefreshHealth().PendingReasons; len(got) != 0 { + t.Fatalf("pending reasons after success = %v", got) + } +} + +func TestDWNControlCoordinatorBoundsRefreshWithTimeout(t *testing.T) { + observerErrors := make(chan error, 1) + resultErrors := make(chan error, 1) + observer := dwnControlObserverFunc(func(_ controlclient.Client, status controlclient.Status) { + if status.Err != nil { + observerErrors <- status.Err + } + }) + + cc, err := NewDWNControl(&DWNControlConfig{ + MapResponseFunc: func(ctx context.Context) (*netmap.NetworkMap, error) { + <-ctx.Done() + return nil, ctx.Err() + }, + OnMapResult: func(_ context.Context, _ *netmap.NetworkMap, err error) { + resultErrors <- err + }, + PollInterval: time.Hour, + RefreshTimeout: 20 * time.Millisecond, + Logf: func(string, ...any) {}, + }, controlclient.Options{Observer: observer, SkipStartForTests: true}) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + defer cc.Shutdown() + + err = cc.refreshControlState(context.Background(), RefreshBatch{}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("refreshControlState error = %v, want deadline exceeded", err) + } + if got := receiveDWNControlError(t, observerErrors); !errors.Is(got, context.DeadlineExceeded) { + t.Fatalf("observer error = %v, want deadline exceeded", got) + } + if got := receiveDWNControlError(t, resultErrors); !errors.Is(got, context.DeadlineExceeded) { + t.Fatalf("map-result error = %v, want deadline exceeded", got) + } +} + +func TestDWNControlCoordinatorBoundsStartupReplayWithTimeout(t *testing.T) { + var applies atomic.Int32 + observer := dwnControlObserverFunc(func(_ controlclient.Client, status controlclient.Status) { + if status.NetMap != nil { + applies.Add(1) + } + }) + + cc, err := NewDWNControl(&DWNControlConfig{ + MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { + return &netmap.NetworkMap{}, nil + }, + PollInterval: time.Hour, + RefreshTimeout: 20 * time.Millisecond, + Logf: func(string, ...any) {}, + }, controlclient.Options{Observer: observer, SkipStartForTests: true}) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + defer cc.Shutdown() + + err = cc.refreshControlState(context.Background(), RefreshBatch{}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("refreshControlState error = %v, want deadline exceeded", err) + } + if got := applies.Load(); got != 1 { + t.Fatalf("observer applications = %d, want one initial application and no replay", got) + } + if !cc.initialMapApplied.Load() { + t.Fatal("timed-out replay did not reserve the one-time startup replay") + } + + if err := cc.refreshControlState(context.Background(), RefreshBatch{}); err != nil { + t.Fatalf("retry refreshControlState: %v", err) + } + if got := applies.Load(); got != 2 { + t.Fatalf("observer applications after retry = %d, want retry application to complete the deferred replay", got) + } +} + +func TestDWNControlCoordinatorCoalescesPreStartAndMidFlightNotify(t *testing.T) { + var loads atomic.Int32 + var applies atomic.Int32 + var mapResults atomic.Int32 + started := make(chan int32, 2) + releaseFirst := make(chan struct{}) + observer := dwnControlObserverFunc(func(_ controlclient.Client, status controlclient.Status) { + if status.NetMap != nil { + applies.Add(1) + } + }) + + cc, err := NewDWNControl(&DWNControlConfig{ + MapResponseFunc: func(ctx context.Context) (*netmap.NetworkMap, error) { + call := loads.Add(1) + started <- call + if call == 1 { + select { + case <-releaseFirst: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return routingTestNetMap(), nil + }, + OnMapResult: func(context.Context, *netmap.NetworkMap, error) { + mapResults.Add(1) + }, + PollInterval: time.Hour, + Logf: func(string, ...any) {}, + }, controlclient.Options{Observer: observer, SkipStartForTests: true}) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + defer cc.Shutdown() + + cc.Notify() + cc.Notify() + if err := cc.coordinator.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + if call := receiveDWNControlLoad(t, started); call != 1 { + t.Fatalf("first load number = %d, want 1", call) + } + cc.Notify() + cc.Notify() + close(releaseFirst) + if call := receiveDWNControlLoad(t, started); call != 2 { + t.Fatalf("trailing load number = %d, want 2", call) + } + + waitForDWNControlCondition(t, func() bool { + health := cc.RefreshHealth() + return !health.InFlight && len(health.PendingReasons) == 0 && !health.LastSuccessAt.IsZero() + }) + if got := loads.Load(); got != 2 { + t.Fatalf("remote loads = %d, want one initial and one trailing load", got) + } + if got := applies.Load(); got != 3 { + t.Fatalf("observer applies = %d, want initial, replay, and trailing", got) + } + if got := mapResults.Load(); got != 2 { + t.Fatalf("map-result callbacks = %d, want 2", got) + } +} + +func TestDWNControlCoordinatorPauseRetainsPendingUntilOneUnpausedLoad(t *testing.T) { + var loads atomic.Int32 + started := make(chan struct{}, 1) + cc, err := NewDWNControl(&DWNControlConfig{ + MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { + loads.Add(1) + started <- struct{}{} + return nil, nil + }, + PollInterval: time.Hour, + Logf: func(string, ...any) {}, + }, controlclient.Options{SkipStartForTests: true}) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + defer cc.Shutdown() + + cc.SetPaused(true) + cc.Notify() + cc.Notify() + if err := cc.coordinator.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + waitForDWNControlCondition(t, func() bool { + health := cc.RefreshHealth() + return health.Running && health.Paused + }) + if got := loads.Load(); got != 0 { + t.Fatalf("loads while paused = %d, want 0", got) + } + if got := cc.RefreshHealth().PendingReasons; !reflect.DeepEqual(got, []RefreshReason{RefreshReasonManual, RefreshReasonStartup}) { + t.Fatalf("pending reasons while paused = %v, want manual and startup", got) + } + + cc.SetPaused(false) + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("timed out waiting for unpaused refresh") + } + waitForDWNControlCondition(t, func() bool { + health := cc.RefreshHealth() + return !health.InFlight && len(health.PendingReasons) == 0 && !health.LastSuccessAt.IsZero() + }) + if got := loads.Load(); got != 1 { + t.Fatalf("loads after unpause = %d, want 1", got) + } +} + +func TestDWNControlPollLoopWaitsForLiveStreamsBeforeStartupLoad(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + loads := make(chan struct{}, 2) + cc, err := NewDWNControl(&DWNControlConfig{ + MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { + loads <- struct{}{} + return nil, nil + }, + PollInterval: time.Hour, + StartupSubscriptionWait: time.Second, + Logf: func(string, ...any) {}, + }, controlclient.Options{SkipStartForTests: true}) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + defer cc.Shutdown() + + cc.coordinator.SetStreamCovered(RefreshStreamTopology, true) + cc.coordinator.SetStreamCovered(RefreshStreamDelivery, true) + go cc.pollLoop(ctx) + + select { + case <-loads: + t.Fatal("startup loaded before covered streams became live") + case <-time.After(100 * time.Millisecond): + } + + cc.coordinator.SetStreamLive(RefreshStreamTopology, true, true) + cc.coordinator.SetStreamLive(RefreshStreamDelivery, true, true) + select { + case <-loads: + case <-time.After(time.Second): + t.Fatal("timed out waiting for post-subscribe startup load") + } + waitForDWNControlCondition(t, func() bool { + health := cc.RefreshHealth() + return !health.InFlight && health.StreamsHealthy + }) + select { + case <-loads: + t.Fatal("post-subscribe startup performed a duplicate remote load") + case <-time.After(100 * time.Millisecond): + } +} + +func TestRefreshStreamsReadyForStartup(t *testing.T) { + tests := []struct { + name string + streams map[RefreshStream]RefreshStreamHealth + want bool + }{ + {name: "unconfigured", streams: map[RefreshStream]RefreshStreamHealth{}, want: false}, + {name: "role delivery live", streams: map[RefreshStream]RefreshStreamHealth{ + RefreshStreamDelivery: {Covered: true, Live: true}, + }, want: true}, + {name: "covered delivery connecting", streams: map[RefreshStream]RefreshStreamHealth{ + RefreshStreamDelivery: {Covered: true}, + }, want: false}, + {name: "both live", streams: map[RefreshStream]RefreshStreamHealth{ + RefreshStreamTopology: {Covered: true, Live: true}, + RefreshStreamDelivery: {Covered: true, Live: true}, + }, want: true}, + {name: "topology connecting", streams: map[RefreshStream]RefreshStreamHealth{ + RefreshStreamTopology: {Covered: true}, + RefreshStreamDelivery: {Covered: true, Live: true}, + }, want: false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := refreshStreamsReadyForStartup(RefreshCoordinatorHealth{Streams: tc.streams}); got != tc.want { + t.Fatalf("ready = %v, want %v", got, tc.want) + } + }) + } +} + +func TestDWNControlCoordinatorUsesConfiguredRefreshIntervals(t *testing.T) { + const ( + fallback = 17 * time.Second + healthy = 91 * time.Second + timeout = 3 * time.Second + startup = 4 * time.Second + ) + cc, err := NewDWNControl(&DWNControlConfig{ + MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { return nil, nil }, + PollInterval: fallback, + HealthyPollInterval: healthy, + RefreshTimeout: timeout, + StartupSubscriptionWait: startup, + Logf: func(string, ...any) {}, + }, controlclient.Options{SkipStartForTests: true}) + if err != nil { + t.Fatalf("NewDWNControl: %v", err) + } + defer cc.Shutdown() + + if got := cc.config.PollInterval; got != fallback { + t.Fatalf("control fallback interval = %v, want %v", got, fallback) + } + if got := cc.config.HealthyPollInterval; got != healthy { + t.Fatalf("control healthy interval = %v, want %v", got, healthy) + } + if got := cc.config.RefreshTimeout; got != timeout { + t.Fatalf("control refresh timeout = %v, want %v", got, timeout) + } + if got := cc.config.StartupSubscriptionWait; got != startup { + t.Fatalf("control startup subscription wait = %v, want %v", got, startup) + } + if got := cc.coordinator.fallbackInterval; got != fallback { + t.Fatalf("coordinator fallback interval = %v, want %v", got, fallback) + } + if got := cc.coordinator.healthyInterval; got != healthy { + t.Fatalf("coordinator healthy interval = %v, want %v", got, healthy) + } + if got := cc.coordinator.debounce; got != 250*time.Millisecond { + t.Fatalf("coordinator debounce = %v, want 250ms", got) + } + if got := cc.coordinator.maxDebounce; got != time.Second { + t.Fatalf("coordinator max debounce = %v, want 1s", got) + } +} + +type dwnControlObserverFunc func(controlclient.Client, controlclient.Status) + +func (f dwnControlObserverFunc) SetControlClientStatus(client controlclient.Client, status controlclient.Status) { + f(client, status) +} + +func receiveDWNControlStatus(t *testing.T, statuses <-chan controlclient.Status) controlclient.Status { + t.Helper() + select { + case status := <-statuses: + return status + case <-time.After(time.Second): + t.Fatal("timed out waiting for control status") + return controlclient.Status{} + } +} + +func receiveDWNControlError(t *testing.T, errors <-chan error) error { + t.Helper() + select { + case err := <-errors: + return err + case <-time.After(time.Second): + t.Fatal("timed out waiting for control error") + return nil + } +} + +func receiveDWNControlLoad(t *testing.T, loads <-chan int32) int32 { + t.Helper() + select { + case load := <-loads: + return load + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for control load") + return 0 + } +} + +func waitForDWNControlCondition(t *testing.T, condition func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for !condition() { + if time.Now().After(deadline) { + t.Fatal("timed out waiting for DWN control condition") + } + time.Sleep(time.Millisecond) + } +} diff --git a/internal/engine/engine.go b/internal/engine/engine.go index ae58443..7885381 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -12,6 +12,7 @@ import ( "net/netip" "runtime" "sync" + "sync/atomic" "syscall" "time" @@ -71,7 +72,8 @@ type Engine struct { routingMu sync.RWMutex routing routingState - snapshots *meshSnapshotStore + snapshots *meshSnapshotStore + refreshCoordinator atomic.Pointer[RefreshCoordinator] } // Config holds the configuration for creating an Engine. @@ -104,9 +106,18 @@ type Config struct { // ListenPort is the WireGuard UDP port. 0 = auto-select. ListenPort uint16 - // PollInterval is how often to re-read DWN state. Default: 30s. + // PollInterval is the fallback reconciliation cadence. Default: 30s. + // When explicitly set, it also remains the healthy anti-entropy cadence + // unless HealthyPollInterval is provided. PollInterval time.Duration + // HealthyPollInterval is the anti-entropy cadence while all required + // subscription streams are live and repaired. Default: 5m. + HealthyPollInterval time.Duration + + // RefreshTimeout bounds one complete DWN state rebuild. Default: 2m. + RefreshTimeout time.Duration + // EncryptionKeyManager manages derived encryption keys for decrypting // protocol records. If nil, encrypted records cannot be read. EncryptionKeyManager *dwncrypto.EncryptionKeyManager @@ -233,10 +244,22 @@ func New(cfg Config) (*Engine, error) { l = slog.Default() } + pollConfigured := cfg.PollInterval != 0 pollInterval := cfg.PollInterval if pollInterval == 0 { pollInterval = 30 * time.Second } + healthyPollInterval := cfg.HealthyPollInterval + if healthyPollInterval == 0 { + healthyPollInterval = 5 * time.Minute + if pollConfigured { + healthyPollInterval = pollInterval + } + } + refreshTimeout := cfg.RefreshTimeout + if refreshTimeout == 0 { + refreshTimeout = 2 * time.Minute + } magicDNS := cfg.MagicDNSSuffix if magicDNS == "" { @@ -457,17 +480,25 @@ func New(cfg Config) (*Engine, error) { mapFn := mapResponseFunc(dwnClient.LoadState, converter, snapshots.record) var engineRef *Engine dwnControlConfig := &DWNControlConfig{ - MapResponseFunc: mapFn, - PollInterval: pollInterval, - Logf: logf, + MapResponseFunc: mapFn, + PollInterval: pollInterval, + HealthyPollInterval: healthyPollInterval, + RefreshTimeout: refreshTimeout, + StartupSubscriptionWait: time.Second, + Logf: logf, OnMapResult: func(ctx context.Context, nm *netmap.NetworkMap, err error) { if engineRef != nil { engineRef.handleControlMapResult(ctx, nm, err) } }, - // Capture the DWNControl reference so the subscription watcher - // can call Notify() to trigger immediate re-polls. - OnCreated: subWatcher.SetDWNControl, + // Share the single coordinator with subscription invalidation and + // daemon health readers before its worker starts. + OnCreated: func(cc *DWNControl) { + subWatcher.SetRefreshCoordinator(cc.coordinator) + if engineRef != nil { + engineRef.refreshCoordinator.Store(cc.coordinator) + } + }, } // If the caller provided a WireGuard private key (already published to @@ -551,13 +582,9 @@ func (e *Engine) Start(ctx context.Context) error { cancel() return fmt.Errorf("starting backend: %w", err) } - e.waitForInitialTUNRoutes(runCtx, 5*time.Second) - go e.runTUNRouteReconciler(runCtx) - - // Start the subscription watcher for real-time updates. - // This is best-effort: if the DWN server doesn't support WebSocket - // subscriptions, we fall back to polling only. The watcher will - // auto-reconnect on transient failures. + // Establish subscriptions before the coordinator's delayed startup load + // whenever possible. Events received during that load remain pending and + // produce one trailing repair, closing the startup lost-update window. if e.subWatcher != nil { if err := e.subWatcher.Start(runCtx); err != nil { e.logger.Warn("failed to start subscription watcher, falling back to polling", @@ -566,6 +593,9 @@ func (e *Engine) Start(ctx context.Context) error { } } + e.waitForInitialTUNRoutes(runCtx, 5*time.Second) + go e.runTUNRouteReconciler(runCtx) + e.running = true e.logger.InfoContext(ctx, "meshd engine started") return nil @@ -582,15 +612,15 @@ func (e *Engine) Stop() error { e.logger.Info("stopping meshd engine") - if e.cancel != nil { - e.cancel() - } - // Stop subscription watcher first (stops WebSocket connections). if e.subWatcher != nil { e.subWatcher.Stop() } + if e.cancel != nil { + e.cancel() + } + // Shutdown order matters: backend first (stops control polling), // then network monitor, then TUN device. e.backend.Shutdown() @@ -619,6 +649,20 @@ func (e *Engine) Running() bool { return e.running } +// RefreshHealth returns the coordinator scheduler and subscription health. +// It returns nil before the LocalBackend creates its control client. +func (e *Engine) RefreshHealth() *RefreshCoordinatorHealth { + if e == nil { + return nil + } + coordinator := e.refreshCoordinator.Load() + if coordinator == nil { + return nil + } + health := coordinator.Health() + return &health +} + // Backend returns the underlying meshnet LocalBackend for advanced use. // Most callers should use the Engine methods instead. func (e *Engine) Backend() *ipnlocal.LocalBackend { diff --git a/internal/engine/refresh_coordinator.go b/internal/engine/refresh_coordinator.go new file mode 100644 index 0000000..daa7e49 --- /dev/null +++ b/internal/engine/refresh_coordinator.go @@ -0,0 +1,808 @@ +package engine + +import ( + "context" + "errors" + "fmt" + "math/rand/v2" + "sort" + "sync" + "time" + + "github.com/enboxorg/meshd/internal/dwn" +) + +// RefreshReason identifies a source of control-plane invalidation. Reasons are +// retained until a refresh succeeds, and duplicate reasons are coalesced. +type RefreshReason string + +const ( + RefreshReasonStartup RefreshReason = "startup" + RefreshReasonPeriodic RefreshReason = "periodic" + RefreshReasonTopology RefreshReason = "topology" + RefreshReasonDelivery RefreshReason = "delivery" + RefreshReasonEndpoint RefreshReason = "endpoint" + RefreshReasonManual RefreshReason = "manual" +) + +// RefreshStream identifies an authoritative subscription stream. +type RefreshStream string + +const ( + RefreshStreamTopology RefreshStream = "topology" + RefreshStreamDelivery RefreshStream = "delivery" +) + +// RefreshBatch is the immutable set of invalidations represented by one +// refresh attempt. Attempt starts at one and increases across consecutive +// failures. Sequence is monotonically increasing within a coordinator. +type RefreshBatch struct { + Reasons []RefreshReason + Sequence uint64 + Attempt int + StartedAt time.Time +} + +// RefreshFunc rebuilds and publishes the complete control-plane view. +type RefreshFunc func(context.Context, RefreshBatch) error + +// RefreshTimer is the small timer surface used by RefreshCoordinator. It is +// exported so deterministic clocks can be supplied by users and tests. +type RefreshTimer interface { + C() <-chan time.Time + Stop() bool +} + +// RefreshClock supplies time and timers to RefreshCoordinator. +type RefreshClock interface { + Now() time.Time + NewTimer(time.Duration) RefreshTimer +} + +// RefreshCoordinatorConfig configures a single refresh worker. +type RefreshCoordinatorConfig struct { + Refresh RefreshFunc + + // FallbackInterval is used whenever either required stream is uncovered, + // disconnected, or has not yet been repaired. Default: 30 seconds. + FallbackInterval time.Duration + + // HealthyInterval is the anti-entropy interval when both topology and + // delivery streams are covered, live, and repaired. Default: 5 minutes. + HealthyInterval time.Duration + + // Debounce waits for a quiet window after event invalidations. MaxDebounce + // caps the total coalescing delay so a hot stream cannot starve refreshes. + // Zero Debounce disables both. + Debounce time.Duration + MaxDebounce time.Duration + + // RetryBackoff and MaxRetryBackoff bound exponential retry delays. + // Defaults: 1 second and 30 seconds. + RetryBackoff time.Duration + MaxRetryBackoff time.Duration + + // Jitter transforms periodic intervals. A nil function applies random + // plus or minus 10 percent jitter. + Jitter func(time.Duration) time.Duration + + // RetryJitter transforms retry delays. The result is clamped so it can + // never schedule earlier than exponential backoff or Retry-After. A nil + // function adds random zero to 20 percent delay. + RetryJitter func(time.Duration) time.Duration + + // Clock defaults to the system clock. + Clock RefreshClock +} + +// RefreshMode describes the current anti-entropy cadence. +type RefreshMode string + +const ( + RefreshModeFallback RefreshMode = "fallback" + RefreshModeHealthy RefreshMode = "healthy" +) + +// RefreshStreamHealth is an immutable view of one subscription source. +type RefreshStreamHealth struct { + Covered bool + Live bool + Repaired bool +} + +// RefreshCoordinatorHealth is a point-in-time, deep-copied scheduler view. +type RefreshCoordinatorHealth struct { + Running bool + Paused bool + InFlight bool + Mode RefreshMode + StreamsHealthy bool + Streams map[RefreshStream]RefreshStreamHealth + PendingReasons []RefreshReason + ConsecutiveFailures int + LastAttemptAt time.Time + LastSuccessAt time.Time + LastReasons []RefreshReason + LastDuration time.Duration + LastError string + RetryNotBefore time.Time + NextPeriodicAt time.Time + NextAttemptAt time.Time +} + +type refreshPending struct { + sequence uint64 +} + +type refreshStreamState struct { + covered bool + live bool + repaired bool + repairSequence uint64 +} + +// RefreshCoordinator serializes full control-plane refreshes while preserving +// every invalidation until a successful rebuild. Its wake channel is only a +// hint; durable pending work is always owned by mu. +type RefreshCoordinator struct { + refresh RefreshFunc + fallbackInterval time.Duration + healthyInterval time.Duration + retryBackoff time.Duration + maxRetryBackoff time.Duration + debounce time.Duration + maxDebounce time.Duration + jitter func(time.Duration) time.Duration + retryJitter func(time.Duration) time.Duration + clock RefreshClock + + mu sync.Mutex + pending map[RefreshReason]refreshPending + streams map[RefreshStream]refreshStreamState + wake chan struct{} + + sequence uint64 + started bool + stopped bool + running bool + paused bool + inFlight bool + consecutiveFailures int + lastAttemptAt time.Time + lastSuccessAt time.Time + lastReasons []RefreshReason + lastDuration time.Duration + lastError string + retryNotBefore time.Time + debounceStartedAt time.Time + debounceNotBefore time.Time + nextPeriodicAt time.Time + cancel context.CancelFunc + done chan struct{} +} + +// NewRefreshCoordinator constructs a coordinator. Call Start exactly once to +// launch its worker. +func NewRefreshCoordinator(cfg RefreshCoordinatorConfig) (*RefreshCoordinator, error) { + if cfg.Refresh == nil { + return nil, fmt.Errorf("refresh coordinator requires a refresh function") + } + + fallbackInterval := cfg.FallbackInterval + if fallbackInterval == 0 { + fallbackInterval = 30 * time.Second + } + healthyInterval := cfg.HealthyInterval + if healthyInterval == 0 { + healthyInterval = 5 * time.Minute + } + retryBackoff := cfg.RetryBackoff + if retryBackoff == 0 { + retryBackoff = time.Second + } + maxRetryBackoff := cfg.MaxRetryBackoff + if maxRetryBackoff == 0 { + maxRetryBackoff = 30 * time.Second + } + maxDebounce := cfg.MaxDebounce + if cfg.Debounce > 0 && maxDebounce == 0 { + maxDebounce = cfg.Debounce + } + if fallbackInterval < 0 || healthyInterval < 0 || retryBackoff < 0 || maxRetryBackoff < 0 || cfg.Debounce < 0 || maxDebounce < 0 { + return nil, fmt.Errorf("refresh coordinator durations must be positive") + } + if maxRetryBackoff < retryBackoff { + return nil, fmt.Errorf("refresh coordinator max retry backoff must be at least retry backoff") + } + if cfg.Debounce > 0 && maxDebounce < cfg.Debounce { + return nil, fmt.Errorf("refresh coordinator max debounce must be at least debounce") + } + + clock := cfg.Clock + if clock == nil { + clock = systemRefreshClock{} + } + jitter := cfg.Jitter + if jitter == nil { + jitter = defaultRefreshJitter + } + retryJitter := cfg.RetryJitter + if retryJitter == nil { + retryJitter = defaultRefreshRetryJitter + } + + return &RefreshCoordinator{ + refresh: cfg.Refresh, + fallbackInterval: fallbackInterval, + healthyInterval: healthyInterval, + retryBackoff: retryBackoff, + maxRetryBackoff: maxRetryBackoff, + debounce: cfg.Debounce, + maxDebounce: maxDebounce, + jitter: jitter, + retryJitter: retryJitter, + clock: clock, + pending: make(map[RefreshReason]refreshPending), + streams: map[RefreshStream]refreshStreamState{ + RefreshStreamTopology: {}, + RefreshStreamDelivery: {}, + }, + wake: make(chan struct{}, 1), + }, nil +} + +// Start launches the sole refresh worker and queues the initial rebuild. A +// coordinator cannot be restarted after Stop or parent-context cancellation. +func (c *RefreshCoordinator) Start(ctx context.Context) error { + if ctx == nil { + return fmt.Errorf("refresh coordinator requires a context") + } + + c.mu.Lock() + if c.started { + c.mu.Unlock() + return fmt.Errorf("refresh coordinator already started") + } + if c.stopped { + c.mu.Unlock() + return fmt.Errorf("refresh coordinator cannot be restarted") + } + runCtx, cancel := context.WithCancel(ctx) + c.started = true + c.running = true + c.cancel = cancel + c.done = make(chan struct{}) + c.enqueueLocked(RefreshReasonStartup) + c.resetPeriodicLocked(c.clock.Now()) + done := c.done + c.mu.Unlock() + + go c.run(runCtx, done) + c.signal() + return nil +} + +// Stop cancels the worker and waits for the in-flight refresh to observe +// cancellation and return. Stop is idempotent. +func (c *RefreshCoordinator) Stop() { + c.mu.Lock() + if c.stopped { + done := c.done + c.mu.Unlock() + if done != nil { + <-done + } + return + } + c.stopped = true + cancel := c.cancel + done := c.done + c.mu.Unlock() + + if cancel != nil { + cancel() + } + c.signal() + if done != nil { + <-done + } +} + +// Notify records non-stream work and wakes the worker without blocking. +func (c *RefreshCoordinator) Notify(reason RefreshReason) { + if reason == "" { + reason = RefreshReasonManual + } + c.mu.Lock() + if c.stopped { + c.mu.Unlock() + return + } + c.enqueueLocked(reason) + c.deferPendingLocked() + c.mu.Unlock() + c.signal() +} + +// SetPaused suspends refresh attempts without discarding pending work. +// Unpausing wakes the worker immediately. +func (c *RefreshCoordinator) SetPaused(paused bool) { + c.mu.Lock() + if c.stopped { + c.mu.Unlock() + return + } + wasPaused := c.paused + changed := wasPaused != paused + c.paused = paused + if wasPaused && !paused { + c.enqueueLocked(RefreshReasonManual) + c.deferPendingLocked() + } + c.mu.Unlock() + if changed { + c.signal() + } +} + +// SetStreamCovered reports whether an authoritative subscription exists for a +// required source. Newly covered streams require a successful repair after +// they become live. Coverage changes reset the adaptive periodic timer. +func (c *RefreshCoordinator) SetStreamCovered(stream RefreshStream, covered bool) { + c.mu.Lock() + if c.stopped { + c.mu.Unlock() + return + } + state := c.streams[stream] + if state.covered == covered { + c.mu.Unlock() + return + } + state.covered = covered + if !covered { + state.live = false + state.repaired = false + state.repairSequence = 0 + delete(c.pending, reasonForStream(stream)) + if len(c.pending) == 0 { + c.debounceStartedAt = time.Time{} + c.debounceNotBefore = time.Time{} + } + } else { + c.sequence++ + state.repaired = false + state.repairSequence = c.sequence + if state.live { + c.enqueueAtLocked(reasonForStream(stream), state.repairSequence) + } + } + c.streams[stream] = state + c.resetPeriodicLocked(c.clock.Now()) + c.mu.Unlock() + c.signal() +} + +// SetStreamLive reports a lifecycle transition. A non-live stream is marked +// stale when needsFullRefresh is true, but repair is deliberately held until a +// fresh stream is live (subscribe-before-repair). A live unrepaired stream +// queues one source-specific rebuild. +func (c *RefreshCoordinator) SetStreamLive(stream RefreshStream, live, needsFullRefresh bool) { + c.mu.Lock() + if c.stopped { + c.mu.Unlock() + return + } + state := c.streams[stream] + wasLive := state.live + changed := wasLive != live + state.live = live + if needsFullRefresh { + c.sequence++ + state.repaired = false + state.repairSequence = c.sequence + changed = true + if !live { + delete(c.pending, reasonForStream(stream)) + if len(c.pending) == 0 { + c.debounceStartedAt = time.Time{} + c.debounceNotBefore = time.Time{} + } + } + } + if live && state.covered && !state.repaired { + // Crossing the replay barrier receives a new revision even if the + // underlying invalidation predates an in-flight refresh. That older + // refresh must not be allowed to mark this stream repaired. + if !wasLive || state.repairSequence == 0 { + c.sequence++ + state.repairSequence = c.sequence + } + c.enqueueAtLocked(reasonForStream(stream), state.repairSequence) + } + c.streams[stream] = state + if changed { + c.resetPeriodicLocked(c.clock.Now()) + } + c.mu.Unlock() + c.signal() +} + +// InvalidateStream marks a subscription source stale. Replay events received +// while the source is not live remain pending but do not trigger a rebuild +// until SetStreamLive reports the replay barrier. +func (c *RefreshCoordinator) InvalidateStream(stream RefreshStream, reason RefreshReason) { + if reason == "" { + reason = reasonForStream(stream) + } + c.mu.Lock() + if c.stopped { + c.mu.Unlock() + return + } + state := c.streams[stream] + c.sequence++ + state.repaired = false + state.repairSequence = c.sequence + c.streams[stream] = state + if state.covered && state.live { + c.enqueueAtLocked(reason, state.repairSequence) + c.deferPendingLocked() + } + c.resetPeriodicLocked(c.clock.Now()) + c.mu.Unlock() + c.signal() +} + +// Health returns a deep copy. Mutating its maps or slices cannot affect the +// coordinator. +func (c *RefreshCoordinator) Health() RefreshCoordinatorHealth { + c.mu.Lock() + defer c.mu.Unlock() + + healthy := c.streamsHealthyLocked() + h := RefreshCoordinatorHealth{ + Running: c.running, + Paused: c.paused, + InFlight: c.inFlight, + Mode: modeForHealth(healthy), + StreamsHealthy: healthy, + Streams: make(map[RefreshStream]RefreshStreamHealth, len(c.streams)), + PendingReasons: sortedPendingReasons(c.pending), + ConsecutiveFailures: c.consecutiveFailures, + LastAttemptAt: c.lastAttemptAt, + LastSuccessAt: c.lastSuccessAt, + LastReasons: append([]RefreshReason(nil), c.lastReasons...), + LastDuration: c.lastDuration, + LastError: c.lastError, + RetryNotBefore: c.retryNotBefore, + NextPeriodicAt: c.nextPeriodicAt, + } + for source, state := range c.streams { + h.Streams[source] = RefreshStreamHealth{ + Covered: state.covered, + Live: state.live, + Repaired: state.repaired, + } + } + if c.running && !c.paused && !c.inFlight { + now := c.clock.Now() + if len(c.pending) > 0 { + h.NextAttemptAt = now + if c.retryNotBefore.After(h.NextAttemptAt) { + h.NextAttemptAt = c.retryNotBefore + } + if c.debounceNotBefore.After(h.NextAttemptAt) { + h.NextAttemptAt = c.debounceNotBefore + } + } else { + h.NextAttemptAt = c.nextPeriodicAt + } + } + return h +} + +type refreshWork struct { + batch RefreshBatch + pending map[RefreshReason]refreshPending + reasons []RefreshReason +} + +func (c *RefreshCoordinator) run(ctx context.Context, done chan struct{}) { + defer func() { + c.mu.Lock() + c.running = false + c.inFlight = false + c.stopped = true + c.mu.Unlock() + close(done) + }() + + for { + if ctx.Err() != nil { + return + } + + now := c.clock.Now() + c.mu.Lock() + work, ready := c.takeWorkLocked(now) + deadline := c.nextDeadlineLocked(now) + c.mu.Unlock() + + if ready { + err := c.refresh(ctx, work.batch) + c.finishWork(work, err) + continue + } + + wait := deadline.Sub(c.clock.Now()) + if wait < 0 { + wait = 0 + } + timer := c.clock.NewTimer(wait) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-c.wake: + timer.Stop() + case <-timer.C(): + c.handleTimer(c.clock.Now()) + } + } +} + +func (c *RefreshCoordinator) takeWorkLocked(now time.Time) (refreshWork, bool) { + if c.paused || c.inFlight || len(c.pending) == 0 || c.retryNotBefore.After(now) || c.debounceNotBefore.After(now) { + return refreshWork{}, false + } + pending := c.pending + c.pending = make(map[RefreshReason]refreshPending) + c.debounceStartedAt = time.Time{} + c.debounceNotBefore = time.Time{} + c.inFlight = true + c.lastAttemptAt = now + + var sequence uint64 + for _, item := range pending { + if item.sequence > sequence { + sequence = item.sequence + } + } + reasons := sortedPendingReasons(pending) + return refreshWork{ + batch: RefreshBatch{ + Reasons: append([]RefreshReason(nil), reasons...), + Sequence: sequence, + Attempt: c.consecutiveFailures + 1, + StartedAt: now, + }, + pending: pending, + reasons: reasons, + }, true +} + +func (c *RefreshCoordinator) finishWork(work refreshWork, err error) { + now := c.clock.Now() + c.mu.Lock() + defer c.mu.Unlock() + c.inFlight = false + c.lastReasons = append(c.lastReasons[:0], work.reasons...) + c.lastDuration = now.Sub(work.batch.StartedAt) + if c.lastDuration < 0 { + c.lastDuration = 0 + } + if err != nil { + for reason, item := range work.pending { + c.enqueueAtLocked(reason, item.sequence) + } + c.consecutiveFailures++ + c.lastError = err.Error() + minimumDelay := exponentialBackoff(c.retryBackoff, c.maxRetryBackoff, c.consecutiveFailures) + var rateLimit *dwn.RateLimitError + if errors.As(err, &rateLimit) && rateLimit.RetryAfter > minimumDelay { + minimumDelay = rateLimit.RetryAfter + } + delay := c.retryJitter(minimumDelay) + if delay < minimumDelay { + delay = minimumDelay + } + c.retryNotBefore = now.Add(delay) + return + } + + c.consecutiveFailures = 0 + c.lastSuccessAt = now + c.lastError = "" + c.retryNotBefore = time.Time{} + for source, state := range c.streams { + if state.covered && state.live && !state.repaired && state.repairSequence != 0 && state.repairSequence <= work.batch.Sequence { + state.repaired = true + c.streams[source] = state + } + } + c.resetPeriodicLocked(now) +} + +func (c *RefreshCoordinator) nextDeadlineLocked(now time.Time) time.Time { + deadline := c.nextPeriodicAt + if deadline.IsZero() { + deadline = now.Add(c.currentIntervalLocked()) + } + if !c.paused && len(c.pending) > 0 { + attemptAt := now + if c.retryNotBefore.After(attemptAt) { + attemptAt = c.retryNotBefore + } + if c.debounceNotBefore.After(attemptAt) { + attemptAt = c.debounceNotBefore + } + if deadline.IsZero() || attemptAt.Before(deadline) { + deadline = attemptAt + } + } + return deadline +} + +func (c *RefreshCoordinator) handleTimer(now time.Time) { + c.mu.Lock() + if !c.nextPeriodicAt.IsZero() && !now.Before(c.nextPeriodicAt) { + c.enqueueLocked(RefreshReasonPeriodic) + c.resetPeriodicLocked(now) + } + c.mu.Unlock() +} + +func (c *RefreshCoordinator) deferPendingLocked() { + if c.debounce <= 0 || len(c.pending) == 0 { + return + } + now := c.clock.Now() + if c.debounceStartedAt.IsZero() { + c.debounceStartedAt = now + } + deadline := now.Add(c.debounce) + if c.maxDebounce > 0 { + maximum := c.debounceStartedAt.Add(c.maxDebounce) + if deadline.After(maximum) { + deadline = maximum + } + } + if deadline.Before(now) { + deadline = now + } + c.debounceNotBefore = deadline +} + +func (c *RefreshCoordinator) enqueueLocked(reason RefreshReason) { + c.sequence++ + c.enqueueAtLocked(reason, c.sequence) +} + +func (c *RefreshCoordinator) enqueueAtLocked(reason RefreshReason, sequence uint64) { + if reason == "" { + reason = RefreshReasonManual + } + current, ok := c.pending[reason] + if !ok || sequence > current.sequence { + c.pending[reason] = refreshPending{sequence: sequence} + } +} + +func (c *RefreshCoordinator) resetPeriodicLocked(now time.Time) { + interval := c.jitter(c.currentIntervalLocked()) + if interval < 0 { + interval = 0 + } + c.nextPeriodicAt = now.Add(interval) +} + +func (c *RefreshCoordinator) currentIntervalLocked() time.Duration { + if c.streamsHealthyLocked() { + return c.healthyInterval + } + return c.fallbackInterval +} + +func (c *RefreshCoordinator) streamsHealthyLocked() bool { + for _, source := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} { + state := c.streams[source] + if !state.covered || !state.live || !state.repaired { + return false + } + } + return true +} + +func (c *RefreshCoordinator) signal() { + select { + case c.wake <- struct{}{}: + default: + } +} + +func reasonForStream(stream RefreshStream) RefreshReason { + switch stream { + case RefreshStreamTopology: + return RefreshReasonTopology + case RefreshStreamDelivery: + return RefreshReasonDelivery + default: + return RefreshReason(stream) + } +} + +func sortedPendingReasons(pending map[RefreshReason]refreshPending) []RefreshReason { + reasons := make([]RefreshReason, 0, len(pending)) + for reason := range pending { + reasons = append(reasons, reason) + } + sort.Slice(reasons, func(i, j int) bool { return reasons[i] < reasons[j] }) + return reasons +} + +func modeForHealth(healthy bool) RefreshMode { + if healthy { + return RefreshModeHealthy + } + return RefreshModeFallback +} + +func exponentialBackoff(base, maximum time.Duration, failures int) time.Duration { + if failures <= 1 { + return base + } + delay := base + for i := 1; i < failures; i++ { + if delay >= maximum || delay > maximum/2 { + return maximum + } + delay *= 2 + } + if delay > maximum { + return maximum + } + return delay +} + +func defaultRefreshJitter(interval time.Duration) time.Duration { + if interval <= 0 { + return interval + } + span := interval / 5 // total width is 20 percent. + if span <= 0 { + return interval + } + return interval - span/2 + time.Duration(rand.Int64N(int64(span)+1)) +} + +func defaultRefreshRetryJitter(delay time.Duration) time.Duration { + if delay <= 0 { + return delay + } + span := delay / 5 + if span <= 0 { + return delay + } + extra := time.Duration(rand.Int64N(int64(span) + 1)) + const maxDuration = time.Duration(1<<63 - 1) + if delay > maxDuration-extra { + return maxDuration + } + return delay + extra +} + +type systemRefreshClock struct{} + +func (systemRefreshClock) Now() time.Time { return time.Now() } + +func (systemRefreshClock) NewTimer(delay time.Duration) RefreshTimer { + return systemRefreshTimer{timer: time.NewTimer(delay)} +} + +type systemRefreshTimer struct { + timer *time.Timer +} + +func (t systemRefreshTimer) C() <-chan time.Time { return t.timer.C } +func (t systemRefreshTimer) Stop() bool { return t.timer.Stop() } diff --git a/internal/engine/refresh_coordinator_test.go b/internal/engine/refresh_coordinator_test.go new file mode 100644 index 0000000..f3412e0 --- /dev/null +++ b/internal/engine/refresh_coordinator_test.go @@ -0,0 +1,670 @@ +package engine + +import ( + "context" + "errors" + "fmt" + "reflect" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/enboxorg/meshd/internal/dwn" +) + +func TestRefreshCoordinatorCoalescesAndRunsOneTrailingRefresh(t *testing.T) { + clock := newCoordinatorFakeClock() + started := make(chan RefreshBatch, 4) + release := make(chan error, 4) + var active atomic.Int32 + var maximum atomic.Int32 + + coordinator := newTestRefreshCoordinator(t, clock, func(ctx context.Context, batch RefreshBatch) error { + current := active.Add(1) + defer active.Add(-1) + for { + old := maximum.Load() + if current <= old || maximum.CompareAndSwap(old, current) { + break + } + } + started <- batch + select { + case err := <-release: + return err + case <-ctx.Done(): + return ctx.Err() + } + }) + + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + first := receiveBatch(t, started) + assertReasons(t, first, RefreshReasonStartup) + + for range 100 { + coordinator.Notify(RefreshReasonTopology) + } + assertNoBatch(t, started) + release <- nil + + second := receiveBatch(t, started) + assertReasons(t, second, RefreshReasonTopology) + for range 100 { + coordinator.Notify(RefreshReasonDelivery) + } + assertNoBatch(t, started) + release <- nil + + third := receiveBatch(t, started) + assertReasons(t, third, RefreshReasonDelivery) + release <- nil + waitCoordinator(t, func() bool { return !coordinator.Health().InFlight }) + assertNoBatch(t, started) + if got := maximum.Load(); got != 1 { + t.Fatalf("maximum concurrent refreshes = %d, want 1", got) + } + + health := coordinator.Health() + if !reflect.DeepEqual(health.LastReasons, []RefreshReason{RefreshReasonDelivery}) { + t.Fatalf("last reasons = %v", health.LastReasons) + } + health.LastReasons[0] = RefreshReasonManual + if got := coordinator.Health().LastReasons[0]; got != RefreshReasonDelivery { + t.Fatalf("mutating Health.LastReasons changed coordinator: %q", got) + } +} + +func TestRefreshCoordinatorDebouncesBurstWithHardCap(t *testing.T) { + clock := newCoordinatorFakeClock() + started := make(chan RefreshBatch, 4) + coordinator, err := NewRefreshCoordinator(RefreshCoordinatorConfig{ + Refresh: func(_ context.Context, batch RefreshBatch) error { + started <- batch + return nil + }, + FallbackInterval: 100 * time.Second, + HealthyInterval: 200 * time.Second, + RetryBackoff: time.Second, + MaxRetryBackoff: time.Second, + Debounce: 2 * time.Second, + MaxDebounce: 5 * time.Second, + Jitter: identityRefreshJitter, + RetryJitter: identityRefreshJitter, + Clock: clock, + }) + if err != nil { + t.Fatal(err) + } + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + receiveBatch(t, started) + waitCoordinator(t, func() bool { return !coordinator.Health().InFlight }) + + coordinator.Notify(RefreshReasonTopology) + assertNoBatch(t, started) + clock.Advance(time.Second) + coordinator.Notify(RefreshReasonDelivery) + clock.Advance(time.Second) + coordinator.Notify(RefreshReasonEndpoint) + clock.Advance(time.Second) + coordinator.Notify(RefreshReasonTopology) + clock.Advance(time.Second) + coordinator.Notify(RefreshReasonDelivery) + assertNoBatch(t, started) + + health := coordinator.Health() + wantDeadline := clock.Now().Add(time.Second) + if !health.NextAttemptAt.Equal(wantDeadline) { + t.Fatalf("next attempt = %v, want hard-cap deadline %v", health.NextAttemptAt, wantDeadline) + } + clock.Advance(time.Second - time.Nanosecond) + assertNoBatch(t, started) + clock.Advance(time.Nanosecond) + batch := receiveBatch(t, started) + assertReasons(t, batch, RefreshReasonDelivery, RefreshReasonEndpoint, RefreshReasonTopology) + waitCoordinator(t, func() bool { return !coordinator.Health().InFlight }) + assertNoBatch(t, started) +} + +func TestRefreshCoordinatorUnpauseQueuesRefreshWithoutPriorPendingWork(t *testing.T) { + clock := newCoordinatorFakeClock() + started := make(chan RefreshBatch, 3) + coordinator := newTestRefreshCoordinator(t, clock, func(_ context.Context, batch RefreshBatch) error { + started <- batch + return nil + }) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + receiveBatch(t, started) + waitCoordinator(t, func() bool { return !coordinator.Health().InFlight }) + + coordinator.SetPaused(true) + coordinator.SetPaused(false) + batch := receiveBatch(t, started) + assertReasons(t, batch, RefreshReasonManual) +} + +func TestRefreshCoordinatorRetriesFailedBatchWithRateLimitAndBackoff(t *testing.T) { + clock := newCoordinatorFakeClock() + started := make(chan RefreshBatch, 4) + results := make(chan error, 4) + coordinator := newTestRefreshCoordinator(t, clock, func(ctx context.Context, batch RefreshBatch) error { + started <- batch + select { + case err := <-results: + return err + case <-ctx.Done(): + return ctx.Err() + } + }) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + + first := receiveBatch(t, started) + if first.Attempt != 1 { + t.Fatalf("first attempt = %d", first.Attempt) + } + clock.Advance(2 * time.Second) + results <- fmt.Errorf("query failed: %w", &dwn.RateLimitError{RetryAfter: 10 * time.Second, Detail: "slow down"}) + waitCoordinator(t, func() bool { return coordinator.Health().ConsecutiveFailures == 1 }) + + health := coordinator.Health() + wantNotBefore := clock.Now().Add(10 * time.Second) + if !health.RetryNotBefore.Equal(wantNotBefore) { + t.Fatalf("retry not before = %v, want %v", health.RetryNotBefore, wantNotBefore) + } + if health.LastDuration != 2*time.Second { + t.Fatalf("last duration = %v, want 2s", health.LastDuration) + } + assertReasonsSlice(t, health.PendingReasons, RefreshReasonStartup) + coordinator.Notify(RefreshReasonEndpoint) + clock.Advance(9 * time.Second) + assertNoBatch(t, started) + clock.Advance(time.Second) + + second := receiveBatch(t, started) + if second.Attempt != 2 { + t.Fatalf("second attempt = %d", second.Attempt) + } + assertReasons(t, second, RefreshReasonEndpoint, RefreshReasonPeriodic, RefreshReasonStartup) + results <- errors.New("still unavailable") + waitCoordinator(t, func() bool { return coordinator.Health().ConsecutiveFailures == 2 }) + wantNotBefore = clock.Now().Add(2 * time.Second) + if got := coordinator.Health().RetryNotBefore; !got.Equal(wantNotBefore) { + t.Fatalf("second retry not before = %v, want %v", got, wantNotBefore) + } + clock.Advance(time.Second) + assertNoBatch(t, started) + clock.Advance(time.Second) + + third := receiveBatch(t, started) + if third.Attempt != 3 { + t.Fatalf("third attempt = %d", third.Attempt) + } + results <- nil + waitCoordinator(t, func() bool { + health := coordinator.Health() + return !health.InFlight && health.ConsecutiveFailures == 0 && health.LastError == "" + }) + if got := coordinator.Health().PendingReasons; len(got) != 0 { + t.Fatalf("pending reasons after success = %v", got) + } +} + +func TestRefreshCoordinatorPauseRetainsPendingWork(t *testing.T) { + clock := newCoordinatorFakeClock() + started := make(chan RefreshBatch, 2) + coordinator := newTestRefreshCoordinator(t, clock, func(_ context.Context, batch RefreshBatch) error { + started <- batch + return nil + }) + coordinator.SetPaused(true) + coordinator.Notify(RefreshReasonEndpoint) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + assertNoBatch(t, started) + + health := coordinator.Health() + if !health.Paused || !health.NextAttemptAt.IsZero() { + t.Fatalf("paused health = %+v", health) + } + assertReasonsSlice(t, health.PendingReasons, RefreshReasonEndpoint, RefreshReasonStartup) + coordinator.SetPaused(false) + batch := receiveBatch(t, started) + assertReasons(t, batch, RefreshReasonEndpoint, RefreshReasonManual, RefreshReasonStartup) + waitCoordinator(t, func() bool { return !coordinator.Health().InFlight }) +} + +func TestRefreshCoordinatorStreamRepairGatingAndAdaptiveCadence(t *testing.T) { + clock := newCoordinatorFakeClock() + started := make(chan RefreshBatch, 8) + coordinator := newTestRefreshCoordinator(t, clock, func(_ context.Context, batch RefreshBatch) error { + started <- batch + return nil + }) + coordinator.SetStreamCovered(RefreshStreamTopology, true) + coordinator.SetStreamCovered(RefreshStreamDelivery, true) + coordinator.SetStreamLive(RefreshStreamTopology, true, true) + coordinator.SetStreamLive(RefreshStreamDelivery, true, true) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + + initial := receiveBatch(t, started) + assertReasons(t, initial, RefreshReasonDelivery, RefreshReasonStartup, RefreshReasonTopology) + waitCoordinator(t, func() bool { return coordinator.Health().StreamsHealthy }) + health := coordinator.Health() + if health.Mode != RefreshModeHealthy { + t.Fatalf("mode = %q, want healthy", health.Mode) + } + if want := clock.Now().Add(100 * time.Second); !health.NextPeriodicAt.Equal(want) { + t.Fatalf("healthy periodic = %v, want %v", health.NextPeriodicAt, want) + } + + coordinator.SetPaused(true) + coordinator.InvalidateStream(RefreshStreamTopology, RefreshReasonTopology) + coordinator.SetStreamLive(RefreshStreamTopology, false, true) + coordinator.SetPaused(false) + resumed := receiveBatch(t, started) + assertReasons(t, resumed, RefreshReasonManual) + waitCoordinator(t, func() bool { return !coordinator.Health().InFlight }) + health = coordinator.Health() + if health.Mode != RefreshModeFallback || health.Streams[RefreshStreamTopology].Repaired { + t.Fatalf("gap health = %+v", health) + } + if want := clock.Now().Add(10 * time.Second); !health.NextPeriodicAt.Equal(want) { + t.Fatalf("fallback periodic = %v, want %v", health.NextPeriodicAt, want) + } + clock.Advance(10 * time.Second) + fallback := receiveBatch(t, started) + assertReasons(t, fallback, RefreshReasonPeriodic) + waitCoordinator(t, func() bool { return !coordinator.Health().InFlight }) + if coordinator.Health().Streams[RefreshStreamTopology].Repaired { + t.Fatal("fallback anti-entropy incorrectly repaired a non-live stream") + } + + coordinator.SetStreamLive(RefreshStreamTopology, true, false) + repair := receiveBatch(t, started) + assertReasons(t, repair, RefreshReasonTopology) + waitCoordinator(t, func() bool { return coordinator.Health().StreamsHealthy }) + if want := clock.Now().Add(100 * time.Second); !coordinator.Health().NextPeriodicAt.Equal(want) { + t.Fatalf("repaired periodic = %v, want %v", coordinator.Health().NextPeriodicAt, want) + } + + clock.Advance(99 * time.Second) + assertNoBatch(t, started) + clock.Advance(time.Second) + periodic := receiveBatch(t, started) + assertReasons(t, periodic, RefreshReasonPeriodic) +} + +func TestRefreshCoordinatorDoesNotRepairMidFlightInvalidation(t *testing.T) { + clock := newCoordinatorFakeClock() + started := make(chan RefreshBatch, 4) + release := make(chan struct{}, 4) + coordinator := newTestRefreshCoordinator(t, clock, func(ctx context.Context, batch RefreshBatch) error { + started <- batch + select { + case <-release: + return nil + case <-ctx.Done(): + return ctx.Err() + } + }) + coordinator.SetStreamCovered(RefreshStreamTopology, true) + coordinator.SetStreamCovered(RefreshStreamDelivery, true) + coordinator.SetStreamLive(RefreshStreamTopology, true, true) + coordinator.SetStreamLive(RefreshStreamDelivery, true, true) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + receiveBatch(t, started) + + coordinator.InvalidateStream(RefreshStreamTopology, RefreshReasonTopology) + release <- struct{}{} + trailing := receiveBatch(t, started) + assertReasons(t, trailing, RefreshReasonTopology) + if coordinator.Health().Streams[RefreshStreamTopology].Repaired { + t.Fatal("mid-flight invalidation was incorrectly marked repaired") + } + release <- struct{}{} + waitCoordinator(t, func() bool { return coordinator.Health().StreamsHealthy }) +} + +func TestRefreshCoordinatorHealthIsDeepCopied(t *testing.T) { + clock := newCoordinatorFakeClock() + coordinator := newTestRefreshCoordinator(t, clock, func(context.Context, RefreshBatch) error { return nil }) + coordinator.SetPaused(true) + coordinator.Notify(RefreshReasonEndpoint) + first := coordinator.Health() + first.PendingReasons[0] = RefreshReasonManual + first.Streams[RefreshStreamTopology] = RefreshStreamHealth{Covered: true, Live: true, Repaired: true} + second := coordinator.Health() + assertReasonsSlice(t, second.PendingReasons, RefreshReasonEndpoint) + if second.Streams[RefreshStreamTopology].Covered { + t.Fatal("mutating returned stream map changed coordinator") + } +} + +func TestRefreshCoordinatorCallbackCannotMutateLastReasons(t *testing.T) { + clock := newCoordinatorFakeClock() + coordinator := newTestRefreshCoordinator(t, clock, func(_ context.Context, batch RefreshBatch) error { + batch.Reasons[0] = RefreshReasonManual + return nil + }) + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + waitCoordinator(t, func() bool { return !coordinator.Health().LastSuccessAt.IsZero() }) + assertReasonsSlice(t, coordinator.Health().LastReasons, RefreshReasonStartup) +} + +func TestRefreshCoordinatorContextShutdownStopsAcceptingWork(t *testing.T) { + clock := newCoordinatorFakeClock() + started := make(chan struct{}, 1) + ctx, cancel := context.WithCancel(context.Background()) + coordinator := newTestRefreshCoordinator(t, clock, func(ctx context.Context, _ RefreshBatch) error { + started <- struct{}{} + <-ctx.Done() + return ctx.Err() + }) + if err := coordinator.Start(ctx); err != nil { + t.Fatal(err) + } + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("refresh did not start") + } + cancel() + waitCoordinator(t, func() bool { return !coordinator.Health().Running }) + before := append([]RefreshReason(nil), coordinator.Health().PendingReasons...) + coordinator.Notify(RefreshReasonManual) + if after := coordinator.Health().PendingReasons; !reflect.DeepEqual(after, before) { + t.Fatalf("work accepted after context shutdown: before=%v after=%v", before, after) + } + coordinator.Stop() +} + +func TestRefreshCoordinatorConcurrentOperations(t *testing.T) { + var active atomic.Int32 + var maximum atomic.Int32 + coordinator, err := NewRefreshCoordinator(RefreshCoordinatorConfig{ + Refresh: func(context.Context, RefreshBatch) error { + current := active.Add(1) + defer active.Add(-1) + for { + old := maximum.Load() + if current <= old || maximum.CompareAndSwap(old, current) { + break + } + } + time.Sleep(time.Microsecond) + return nil + }, + FallbackInterval: time.Hour, + HealthyInterval: time.Hour, + Jitter: identityRefreshJitter, + RetryJitter: identityRefreshJitter, + }) + if err != nil { + t.Fatal(err) + } + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + + var wg sync.WaitGroup + for worker := range 8 { + wg.Add(1) + go func(worker int) { + defer wg.Done() + for i := range 200 { + coordinator.Notify(RefreshReasonManual) + coordinator.InvalidateStream(RefreshStreamTopology, RefreshReasonTopology) + coordinator.SetStreamCovered(RefreshStreamTopology, (worker+i)%2 == 0) + coordinator.SetStreamLive(RefreshStreamTopology, i%2 == 0, i%7 == 0) + coordinator.SetPaused(i%11 == 0) + _ = coordinator.Health() + } + }(worker) + } + wg.Wait() + coordinator.SetPaused(false) + coordinator.Stop() + if got := maximum.Load(); got > 1 { + t.Fatalf("maximum concurrent refreshes = %d", got) + } +} + +func TestNewRefreshCoordinatorValidation(t *testing.T) { + if _, err := NewRefreshCoordinator(RefreshCoordinatorConfig{}); err == nil { + t.Fatal("nil refresh function was accepted") + } + refresh := func(context.Context, RefreshBatch) error { return nil } + if _, err := NewRefreshCoordinator(RefreshCoordinatorConfig{Refresh: refresh, RetryBackoff: 2 * time.Second, MaxRetryBackoff: time.Second}); err == nil { + t.Fatal("inverted retry bounds were accepted") + } + coordinator := newTestRefreshCoordinator(t, newCoordinatorFakeClock(), refresh) + if err := coordinator.Start(nil); err == nil { + t.Fatal("nil start context was accepted") + } + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + if err := coordinator.Start(context.Background()); err == nil { + t.Fatal("second Start was accepted") + } + coordinator.Stop() +} + +func TestRefreshCoordinatorRetryJitterRespectsFloor(t *testing.T) { + clock := newCoordinatorFakeClock() + coordinator, err := NewRefreshCoordinator(RefreshCoordinatorConfig{ + Refresh: func(context.Context, RefreshBatch) error { + return errors.New("unavailable") + }, + FallbackInterval: time.Hour, + HealthyInterval: time.Hour, + RetryBackoff: time.Second, + MaxRetryBackoff: time.Second, + Jitter: identityRefreshJitter, + RetryJitter: func(delay time.Duration) time.Duration { + return delay + 500*time.Millisecond + }, + Clock: clock, + }) + if err != nil { + t.Fatal(err) + } + if err := coordinator.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(coordinator.Stop) + waitCoordinator(t, func() bool { return coordinator.Health().ConsecutiveFailures == 1 }) + if got, want := coordinator.Health().RetryNotBefore, clock.Now().Add(1500*time.Millisecond); !got.Equal(want) { + t.Fatalf("retry deadline = %v, want jittered %v", got, want) + } +} + +func TestDefaultRefreshRetryJitterNeverShortens(t *testing.T) { + const delay = 10 * time.Second + for range 100 { + got := defaultRefreshRetryJitter(delay) + if got < delay || got > 12*time.Second { + t.Fatalf("retry jitter = %v, want in [%v,%v]", got, delay, 12*time.Second) + } + } +} + +func TestRefreshCoordinatorExponentialBackoffCaps(t *testing.T) { + for _, test := range []struct { + failures int + want time.Duration + }{ + {failures: 0, want: time.Second}, + {failures: 1, want: time.Second}, + {failures: 2, want: 2 * time.Second}, + {failures: 3, want: 4 * time.Second}, + {failures: 4, want: 8 * time.Second}, + {failures: 40, want: 8 * time.Second}, + } { + if got := exponentialBackoff(time.Second, 8*time.Second, test.failures); got != test.want { + t.Fatalf("failures %d: backoff = %v, want %v", test.failures, got, test.want) + } + } +} + +func newTestRefreshCoordinator(t *testing.T, clock RefreshClock, refresh RefreshFunc) *RefreshCoordinator { + t.Helper() + coordinator, err := NewRefreshCoordinator(RefreshCoordinatorConfig{ + Refresh: refresh, + FallbackInterval: 10 * time.Second, + HealthyInterval: 100 * time.Second, + RetryBackoff: time.Second, + MaxRetryBackoff: 8 * time.Second, + Jitter: identityRefreshJitter, + RetryJitter: identityRefreshJitter, + Clock: clock, + }) + if err != nil { + t.Fatal(err) + } + return coordinator +} + +func identityRefreshJitter(interval time.Duration) time.Duration { return interval } + +func receiveBatch(t *testing.T, batches <-chan RefreshBatch) RefreshBatch { + t.Helper() + select { + case batch := <-batches: + return batch + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for refresh") + return RefreshBatch{} + } +} + +func assertNoBatch(t *testing.T, batches <-chan RefreshBatch) { + t.Helper() + select { + case batch := <-batches: + t.Fatalf("unexpected refresh: %+v", batch) + case <-time.After(20 * time.Millisecond): + } +} + +func assertReasons(t *testing.T, batch RefreshBatch, reasons ...RefreshReason) { + t.Helper() + assertReasonsSlice(t, batch.Reasons, reasons...) +} + +func assertReasonsSlice(t *testing.T, got []RefreshReason, reasons ...RefreshReason) { + t.Helper() + want := append([]RefreshReason(nil), reasons...) + if !reflect.DeepEqual(got, want) { + t.Fatalf("reasons = %v, want %v", got, want) + } +} + +func waitCoordinator(t *testing.T, condition func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for !condition() { + if time.Now().After(deadline) { + t.Fatal("timed out waiting for coordinator state") + } + time.Sleep(time.Millisecond) + } +} + +type coordinatorFakeClock struct { + mu sync.Mutex + now time.Time + timers map[*coordinatorFakeTimer]struct{} +} + +func newCoordinatorFakeClock() *coordinatorFakeClock { + return &coordinatorFakeClock{ + now: time.Date(2026, 7, 11, 12, 0, 0, 0, time.UTC), + timers: make(map[*coordinatorFakeTimer]struct{}), + } +} + +func (c *coordinatorFakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *coordinatorFakeClock) NewTimer(delay time.Duration) RefreshTimer { + c.mu.Lock() + defer c.mu.Unlock() + timer := &coordinatorFakeTimer{ + clock: c, + deadline: c.now.Add(delay), + ch: make(chan time.Time, 1), + } + c.timers[timer] = struct{}{} + if delay <= 0 { + timer.fired = true + delete(c.timers, timer) + timer.ch <- c.now + } + return timer +} + +func (c *coordinatorFakeClock) Advance(delta time.Duration) { + c.mu.Lock() + c.now = c.now.Add(delta) + now := c.now + var due []*coordinatorFakeTimer + for timer := range c.timers { + if !timer.deadline.After(now) { + timer.fired = true + delete(c.timers, timer) + due = append(due, timer) + } + } + c.mu.Unlock() + for _, timer := range due { + timer.ch <- now + } +} + +type coordinatorFakeTimer struct { + clock *coordinatorFakeClock + deadline time.Time + ch chan time.Time + fired bool + stopped bool +} + +func (t *coordinatorFakeTimer) C() <-chan time.Time { return t.ch } + +func (t *coordinatorFakeTimer) Stop() bool { + t.clock.mu.Lock() + defer t.clock.mu.Unlock() + if t.fired || t.stopped { + return false + } + t.stopped = true + delete(t.clock.timers, t) + return true +} diff --git a/internal/engine/snapshot.go b/internal/engine/snapshot.go index a4c5191..3279acf 100644 --- a/internal/engine/snapshot.go +++ b/internal/engine/snapshot.go @@ -40,7 +40,42 @@ func (s *meshSnapshotStore) load() *MeshSnapshot { if s == nil { return nil } - return cloneMeshSnapshot(s.current.Load()) + snapshot := cloneMeshSnapshot(s.current.Load()) + refreshSnapshotLiveness(snapshot, time.Now()) + return snapshot +} + +func refreshSnapshotLiveness(snapshot *MeshSnapshot, now time.Time) { + if snapshot == nil { + return + } + if snapshot.Self != nil { + snapshot.Self.Online = peerSnapshotOnline(*snapshot.Self, now) + } + peers := snapshot.Peers[:0] + for _, peer := range snapshot.Peers { + if peerSnapshotExpired(peer, now) { + continue + } + peer.Online = peerSnapshotOnline(peer, now) + peers = append(peers, peer) + } + snapshot.Peers = peers +} + +func peerSnapshotOnline(snapshot PeerSnapshot, now time.Time) bool { + return !peerSnapshotExpired(snapshot, now) && + snapshot.LastSeen != nil && + !snapshot.LastSeen.IsZero() && + now.Sub(*snapshot.LastSeen) <= control.DefaultPeerStaleThreshold +} + +func peerSnapshotExpired(snapshot PeerSnapshot, now time.Time) bool { + if snapshot.ExpiresAt == "" { + return false + } + expiresAt, err := time.Parse(time.RFC3339Nano, snapshot.ExpiresAt) + return err == nil && !now.Before(expiresAt) } func (s *meshSnapshotStore) record(resp *control.MapResponse, err error) { diff --git a/internal/engine/snapshot_test.go b/internal/engine/snapshot_test.go index f73377f..469705e 100644 --- a/internal/engine/snapshot_test.go +++ b/internal/engine/snapshot_test.go @@ -14,7 +14,7 @@ import ( ) func TestMeshSnapshotStorePublishesRichSnapshotAndDeepCopies(t *testing.T) { - lastSeen := time.Date(2026, 7, 11, 14, 30, 0, 0, time.UTC) + lastSeen := time.Now().UTC().Add(-time.Minute) resp := &control.MapResponse{ Node: &control.Node{ DID: "did:example:self", @@ -114,6 +114,96 @@ func TestMeshSnapshotStorePublishesRichSnapshotAndDeepCopies(t *testing.T) { } } +func TestMeshSnapshotLivenessIsDerivedFromLastSeenAtReadTime(t *testing.T) { + now := time.Now().UTC() + fresh := now.Add(-control.DefaultPeerStaleThreshold + time.Minute) + stale := now.Add(-control.DefaultPeerStaleThreshold - time.Minute) + store := &meshSnapshotStore{} + store.record(&control.MapResponse{ + Node: &control.Node{ + DID: "did:example:self", + Online: false, + LastSeen: fresh, + }, + Peers: []*control.Node{ + { + DID: "did:example:fresh", + Online: false, + LastSeen: fresh, + }, + { + DID: "did:example:stale", + Online: true, + LastSeen: stale, + }, + { + DID: "did:example:never-seen", + Online: true, + }, + }, + }, nil) + + engine := &Engine{snapshots: store} + snapshot := engine.MeshSnapshot() + if snapshot == nil || snapshot.Self == nil || !snapshot.Self.Online { + t.Fatalf("fresh self liveness = %#v, want online", snapshot) + } + if len(snapshot.Peers) != 3 { + t.Fatalf("Peers length = %d, want 3", len(snapshot.Peers)) + } + if !snapshot.Peers[0].Online || snapshot.Peers[1].Online || snapshot.Peers[2].Online { + t.Fatalf("peer liveness = %#v, want fresh only online", snapshot.Peers) + } + + peers := engine.PeerSnapshots() + if len(peers) != 3 || !peers[0].Online || peers[1].Online || peers[2].Online { + t.Fatalf("PeerSnapshots liveness = %#v, want fresh only online", peers) + } + + // The same materialized snapshot ages offline without another remote + // refresh; only the daemon-facing copy is changed. + refreshSnapshotLiveness(snapshot, now.Add(2*control.DefaultPeerStaleThreshold)) + if snapshot.Self.Online || snapshot.Peers[0].Online { + t.Fatalf("aged snapshot remained online: Self=%#v Peers=%#v", snapshot.Self, snapshot.Peers) + } +} + +func TestMeshSnapshotProjectionAppliesMembershipExpiry(t *testing.T) { + now := time.Now().UTC() + freshSeen := now.Add(-time.Minute) + store := &meshSnapshotStore{} + store.record(&control.MapResponse{ + Node: &control.Node{ + DID: "did:example:self", + ExpiresAt: now.Add(-time.Minute).Format(time.RFC3339Nano), + LastSeen: freshSeen, + Online: true, + }, + Peers: []*control.Node{ + { + DID: "did:example:expired", + ExpiresAt: now.Add(-time.Second).Format(time.RFC3339Nano), + LastSeen: freshSeen, + Online: true, + }, + { + DID: "did:example:active", + ExpiresAt: now.Add(time.Hour).Format(time.RFC3339Nano), + LastSeen: freshSeen, + Online: true, + }, + }, + }, nil) + + snapshot := store.load() + if snapshot == nil || snapshot.Self == nil || snapshot.Self.Online { + t.Fatalf("expired self projection = %#v, want retained but offline", snapshot) + } + if len(snapshot.Peers) != 1 || snapshot.Peers[0].NodeDID != "did:example:active" || !snapshot.Peers[0].Online { + t.Fatalf("expiry-filtered peers = %#v, want only active peer", snapshot.Peers) + } +} + func TestMeshSnapshotStoreFailurePreservesLastGood(t *testing.T) { store := &meshSnapshotStore{} first := &control.MapResponse{ diff --git a/internal/engine/subscribe.go b/internal/engine/subscribe.go index bf3a58b..ea46005 100644 --- a/internal/engine/subscribe.go +++ b/internal/engine/subscribe.go @@ -1,13 +1,12 @@ // Package engine — subscription watcher for real-time DWN updates. // // Instead of polling every 30 seconds, the SubscriptionWatcher opens a -// WebSocket subscription to the anchor DWN and triggers an immediate -// re-poll whenever mesh records change (nodes, endpoints). +// WebSocket subscription to the anchor DWN and invalidates the local +// control-plane snapshot whenever mesh records change (nodes, endpoints). // // The watcher subscribes to all records under the wireguard-mesh protocol -// within the network's contextId. When an event arrives, it calls -// DWNControl.Notify() which wakes the poll loop for an immediate -// loadAndPush cycle. +// within the network's contextId. The refresh coordinator coalesces events and +// serializes the resulting full rebuilds. package engine import ( @@ -36,13 +35,20 @@ type subscriptionManager interface { type subscriptionManagerFactory func(string, *slog.Logger) subscriptionManager -type subscriptionSourceState struct { - live bool - dirty bool +type subscriptionWatcherStreamState struct { + covered bool + live bool } -// SubscriptionWatcher subscribes to DWN record changes and triggers -// engine re-polls via DWNControl.Notify(). +type subscriptionRefreshCoordinator interface { + SetStreamCovered(RefreshStream, bool) + SetStreamLive(RefreshStream, bool, bool) + InvalidateStream(RefreshStream, RefreshReason) + Notify(RefreshReason) +} + +// SubscriptionWatcher subscribes to DWN record changes and reports stream +// health and invalidations to the engine's refresh coordinator. // // This replaces pure 30-second polling with event-driven updates, // reducing peer discovery latency from up to 30s to near-instant. @@ -62,13 +68,11 @@ type SubscriptionWatcher struct { cancel context.CancelFunc done chan struct{} - // dwnControl is set via the OnCreated callback from DWNControlConfig. - // It's nil until the control client is created by the LocalBackend. - controlMu sync.Mutex - dwnControl *DWNControl + coordinatorMu sync.RWMutex + coordinator subscriptionRefreshCoordinator - sourceMu sync.Mutex - sources map[string]subscriptionSourceState + streams map[RefreshStream]subscriptionWatcherStreamState + streamsConfigured bool } // SubscriptionWatcherConfig holds the configuration for creating a @@ -119,7 +123,10 @@ func NewSubscriptionWatcher(cfg SubscriptionWatcherConfig) *SubscriptionWatcher newManager: func(endpoint string, logger *slog.Logger) subscriptionManager { return dwn.NewSubscriptionManager(endpoint, logger) }, - sources: make(map[string]subscriptionSourceState), + streams: map[RefreshStream]subscriptionWatcherStreamState{ + RefreshStreamTopology: {}, + RefreshStreamDelivery: {}, + }, } } @@ -139,136 +146,156 @@ func broadSubscriptionAuth(readAuth dwn.MessageAuth) (dwn.MessageAuth, bool) { return auth, auth.ProtocolRole == "" } -// SetDWNControl is called by the DWNControlConfig.OnCreated callback -// to provide the DWNControl reference for triggering Notify(). -func (w *SubscriptionWatcher) SetDWNControl(cc *DWNControl) { - w.controlMu.Lock() - defer w.controlMu.Unlock() - w.dwnControl = cc +// SetRefreshCoordinator supplies the refresh coordinator that owns stream +// health, invalidation coalescing, and serialized control-plane rebuilds. +// If LocalBackend replaces its control client while subscriptions remain live, +// their current coverage is replayed and forced through one repair. +func (w *SubscriptionWatcher) SetRefreshCoordinator(coordinator subscriptionRefreshCoordinator) { + w.coordinatorMu.Lock() + defer w.coordinatorMu.Unlock() + if coordinator != nil && w.streamsConfigured { + for _, stream := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} { + coordinator.SetStreamCovered(stream, w.streams[stream].covered) + } + for _, stream := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} { + state := w.streams[stream] + coordinator.SetStreamLive(stream, state.live, state.live) + } + } + w.coordinator = coordinator +} + +func (w *SubscriptionWatcher) invalidateStream(stream RefreshStream, reason RefreshReason) { + w.coordinatorMu.RLock() + defer w.coordinatorMu.RUnlock() + if w.coordinator != nil { + w.coordinator.InvalidateStream(stream, reason) + } } -// notify triggers an immediate re-poll on the DWNControl if available. -func (w *SubscriptionWatcher) notify() { - w.controlMu.Lock() - cc := w.dwnControl - w.controlMu.Unlock() +func (w *SubscriptionWatcher) notify(reason RefreshReason) { + w.coordinatorMu.RLock() + defer w.coordinatorMu.RUnlock() + if w.coordinator != nil { + w.coordinator.Notify(reason) + } +} - if cc != nil { - cc.Notify() +func (w *SubscriptionWatcher) setStreamLive(stream RefreshStream, live, needsFullRefresh bool) { + w.coordinatorMu.Lock() + state := w.streams[stream] + state.live = live + w.streams[stream] = state + if w.coordinator != nil { + w.coordinator.SetStreamLive(stream, live, needsFullRefresh) } + w.coordinatorMu.Unlock() } -// handleSubscriptionMessage translates DWN events and lifecycle signals into -// cache invalidations. DWNControl.Notify coalesces repeated signals, so an EOSE -// following a burst of events schedules at most one additional refresh. -func (w *SubscriptionWatcher) handleSubscriptionMessage(source string, message *dwn.SubscriptionMessage) error { +// handleSubscriptionMessage translates DWN events into cache invalidations. +// The coordinator retains and coalesces invalidations; EOSE remains a barrier +// only and is reflected by the established lifecycle callback. +func (w *SubscriptionWatcher) handleSubscriptionMessage(stream RefreshStream, message *dwn.SubscriptionMessage) error { if message == nil { return nil } - invalidates := false switch message.Type { case "event": w.logger.Debug("DWN record changed", - slog.String("source", source), + slog.String("source", string(stream)), slog.Any("cursor", message.Cursor), ) - invalidates = true + w.invalidateStream(stream, reasonForStream(stream)) case "eose": - // EOSE is only a replay barrier. A replayed event already marked the - // source dirty; an empty replay must not force an expensive rebuild. + // EOSE is only a replay barrier. The lifecycle callback establishes the + // stream after replay and lets the coordinator release pending repairs. w.logger.Debug("subscription caught up to live events", - slog.String("source", source), + slog.String("source", string(stream)), slog.Any("cursor", message.Cursor), ) case "error": w.logger.Warn("subscription reported an error", - slog.String("source", source), + slog.String("source", string(stream)), slog.Any("cursor", message.Cursor), slog.Any("error", message.Error), ) - invalidates = true default: w.logger.Debug("ignoring unknown subscription message", - slog.String("source", source), + slog.String("source", string(stream)), slog.String("type", message.Type), ) } - if invalidates && w.recordSourceInvalidation(source) { - w.notify() - } return nil } -// recordSourceInvalidation marks replay frames dirty until their subscription -// is established. Once live, each frame invalidates immediately; DWNControl's -// size-one notify channel coalesces bursts. -func (w *SubscriptionWatcher) recordSourceInvalidation(source string) bool { - w.sourceMu.Lock() - defer w.sourceMu.Unlock() - state := w.sources[source] - if !state.live { - state.dirty = true - w.sources[source] = state - return false - } - return true -} - -func (w *SubscriptionWatcher) handleSubscriptionLifecycle(source string, event dwn.SubscriptionLifecycleEvent) { +func (w *SubscriptionWatcher) handleSubscriptionLifecycle(stream RefreshStream, event dwn.SubscriptionLifecycleEvent) { switch event.Kind { case dwn.SubscriptionLifecycleEstablished: w.logger.Debug("subscription established", - slog.String("source", source), + slog.String("source", string(stream)), slog.Bool("needsFullRefresh", event.NeedsFullRefresh), ) - w.sourceMu.Lock() - state := w.sources[source] - shouldRefresh := event.NeedsFullRefresh || state.dirty - state.live = true - state.dirty = false - w.sources[source] = state - w.sourceMu.Unlock() - if shouldRefresh { - w.notify() - } + w.setStreamLive(stream, true, event.NeedsFullRefresh) case dwn.SubscriptionLifecycleProgressGap: // Wait for the fresh replacement stream before rebuilding. This keeps // the subscribe-before-repair invariant and avoids racing cursor reset. w.logger.Warn("subscription progress gap; waiting for fresh stream", - slog.String("source", source), + slog.String("source", string(stream)), slog.Any("gap", event.Gap), ) - w.markSourceNotLive(source) + w.setStreamLive(stream, false, true) case dwn.SubscriptionLifecycleRetrying: - // The periodic poll remains the fallback while the stream reconnects. + // The coordinator switches to the fallback cadence while reconnecting. w.logger.Debug("subscription retrying", - slog.String("source", source), + slog.String("source", string(stream)), slog.Int("attempt", event.Attempt), slog.Any("error", event.Err), ) - w.markSourceNotLive(source) + w.setStreamLive(stream, false, false) case dwn.SubscriptionLifecycleTerminal: w.logger.Warn("subscription terminated; triggering reconciliation before polling fallback", - slog.String("source", source), + slog.String("source", string(stream)), slog.Any("error", event.Err), ) - w.markSourceNotLive(source) - w.notify() + w.setStreamLive(stream, false, false) + w.notify(reasonForStream(stream)) default: w.logger.Debug("ignoring unknown subscription lifecycle event", - slog.String("source", source), + slog.String("source", string(stream)), slog.String("kind", string(event.Kind)), ) } } -func (w *SubscriptionWatcher) markSourceNotLive(source string) { - w.sourceMu.Lock() - defer w.sourceMu.Unlock() - state := w.sources[source] - state.live = false - w.sources[source] = state +func (w *SubscriptionWatcher) initializeStreamCoverage(topologyCovered bool) { + w.coordinatorMu.Lock() + defer w.coordinatorMu.Unlock() + w.streamsConfigured = true + w.streams[RefreshStreamTopology] = subscriptionWatcherStreamState{covered: topologyCovered} + w.streams[RefreshStreamDelivery] = subscriptionWatcherStreamState{covered: true} + if w.coordinator == nil { + return + } + w.coordinator.SetStreamLive(RefreshStreamTopology, false, false) + w.coordinator.SetStreamLive(RefreshStreamDelivery, false, false) + w.coordinator.SetStreamCovered(RefreshStreamTopology, topologyCovered) + w.coordinator.SetStreamCovered(RefreshStreamDelivery, true) +} + +func (w *SubscriptionWatcher) clearStreamCoverage() { + w.coordinatorMu.Lock() + defer w.coordinatorMu.Unlock() + w.streamsConfigured = false + w.streams[RefreshStreamTopology] = subscriptionWatcherStreamState{} + w.streams[RefreshStreamDelivery] = subscriptionWatcherStreamState{} + if w.coordinator == nil { + return + } + w.coordinator.SetStreamLive(RefreshStreamTopology, false, false) + w.coordinator.SetStreamLive(RefreshStreamDelivery, false, false) + w.coordinator.SetStreamCovered(RefreshStreamTopology, false) + w.coordinator.SetStreamCovered(RefreshStreamDelivery, false) } // Start begins subscribing to DWN record changes. It subscribes to all @@ -291,9 +318,6 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { subCtx, cancel := context.WithCancel(ctx) done := make(chan struct{}) manager := w.newManager(w.endpoint, w.logger) - w.sourceMu.Lock() - clear(w.sources) - w.sourceMu.Unlock() // Subscribe to all records under the wireguard-mesh protocol for // this network. This catches node, endpoint, and relay @@ -301,6 +325,7 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { // RecordsSubscribe broadly. A role-only invocation cannot use this filter: // production DWN requires protocolPath and direct-parent scoping. meshAuth, topologySupported := broadSubscriptionAuth(w.readAuth) + w.initializeStreamCoverage(topologySupported) if topologySupported { filter := dwn.RecordsFilter{ Protocol: protocols.MeshProtocolURI, @@ -313,14 +338,16 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { filter, meshAuth, func(message *dwn.SubscriptionMessage) error { - return w.handleSubscriptionMessage("mesh", message) + return w.handleSubscriptionMessage(RefreshStreamTopology, message) }, func(event dwn.SubscriptionLifecycleEvent) { - w.handleSubscriptionLifecycle("mesh", event) + w.handleSubscriptionLifecycle(RefreshStreamTopology, event) }, ); err != nil { + w.clearStreamCoverage() cancel() manager.CloseAll() + w.clearStreamCoverage() return fmt.Errorf("subscribing to mesh records: %w", err) } } else { @@ -334,13 +361,14 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { // recipient and the control-record tags used by delivery lookups. These // records are recipient-readable and must not invoke the mesh read role or // grant. + // Context IDs differ for owner nodes and member-associated nodes, so the + // recipient-scoped stream deliberately omits an exact contextId tag. deliveryFilter := dwn.RecordsFilter{ Protocol: protocols.MeshProtocolURI, ProtocolPath: dwncrypto.EncryptionControlDeliveryPath, Recipient: w.selfDID, Tags: map[string]any{ - "protocol": protocols.MeshProtocolURI, - "contextId": w.networkRecordID, + "protocol": protocols.MeshProtocolURI, }, } if _, err := manager.SubscribeWithAuthAndLifecycle( @@ -350,14 +378,16 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { deliveryFilter, dwn.MessageAuth{}, func(message *dwn.SubscriptionMessage) error { - return w.handleSubscriptionMessage("delivery", message) + return w.handleSubscriptionMessage(RefreshStreamDelivery, message) }, func(event dwn.SubscriptionLifecycleEvent) { - w.handleSubscriptionLifecycle("delivery", event) + w.handleSubscriptionLifecycle(RefreshStreamDelivery, event) }, ); err != nil { + w.clearStreamCoverage() cancel() manager.CloseAll() + w.clearStreamCoverage() return fmt.Errorf("subscribing to delivery records: %w", err) } @@ -374,6 +404,7 @@ func (w *SubscriptionWatcher) Start(ctx context.Context) error { go func() { <-subCtx.Done() manager.CloseAll() + w.clearStreamCoverage() close(done) }() @@ -386,6 +417,7 @@ func (w *SubscriptionWatcher) Stop() { cancel := w.cancel done := w.done if cancel != nil { + w.clearStreamCoverage() cancel() if done != nil { <-done diff --git a/internal/engine/subscribe_test.go b/internal/engine/subscribe_test.go index a66fc7a..5d7f6b1 100644 --- a/internal/engine/subscribe_test.go +++ b/internal/engine/subscribe_test.go @@ -33,6 +33,7 @@ type recordingSubscriptionManager struct { mu sync.Mutex calls []recordedSubscription failCall int + closeHook func() closeCall atomic.Int32 } @@ -64,9 +65,136 @@ func (m *recordingSubscriptionManager) SubscribeWithAuthAndLifecycle( } func (m *recordingSubscriptionManager) CloseAll() { + if m.closeHook != nil { + m.closeHook() + } m.closeCall.Add(1) } +type refreshCoordinatorCall struct { + method string + stream RefreshStream + covered bool + live bool + needsFullRefresh bool + reason RefreshReason +} + +type recordingRefreshCoordinator struct { + mu sync.Mutex + calls []refreshCoordinatorCall + streams map[RefreshStream]RefreshStreamHealth +} + +func newRecordingRefreshCoordinator() *recordingRefreshCoordinator { + return &recordingRefreshCoordinator{streams: make(map[RefreshStream]RefreshStreamHealth)} +} + +func (c *recordingRefreshCoordinator) SetStreamCovered(stream RefreshStream, covered bool) { + c.mu.Lock() + defer c.mu.Unlock() + health := c.streams[stream] + health.Covered = covered + if !covered { + health.Live = false + } + c.streams[stream] = health + c.calls = append(c.calls, refreshCoordinatorCall{method: "covered", stream: stream, covered: covered}) +} + +func (c *recordingRefreshCoordinator) SetStreamLive(stream RefreshStream, live, needsFullRefresh bool) { + c.mu.Lock() + defer c.mu.Unlock() + health := c.streams[stream] + health.Live = live + c.streams[stream] = health + c.calls = append(c.calls, refreshCoordinatorCall{ + method: "live", + stream: stream, + live: live, + needsFullRefresh: needsFullRefresh, + }) +} + +func (c *recordingRefreshCoordinator) InvalidateStream(stream RefreshStream, reason RefreshReason) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls = append(c.calls, refreshCoordinatorCall{method: "invalidate", stream: stream, reason: reason}) +} + +func (c *recordingRefreshCoordinator) Notify(reason RefreshReason) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls = append(c.calls, refreshCoordinatorCall{method: "notify", reason: reason}) +} + +func (c *recordingRefreshCoordinator) takeCalls() []refreshCoordinatorCall { + c.mu.Lock() + defer c.mu.Unlock() + calls := append([]refreshCoordinatorCall(nil), c.calls...) + c.calls = nil + return calls +} + +func (c *recordingRefreshCoordinator) streamHealth(stream RefreshStream) RefreshStreamHealth { + c.mu.Lock() + defer c.mu.Unlock() + return c.streams[stream] +} + +type blockingReplacementCoordinator struct { + *recordingRefreshCoordinator + entered chan struct{} + release chan struct{} + once sync.Once +} + +func newBlockingReplacementCoordinator() *blockingReplacementCoordinator { + return &blockingReplacementCoordinator{ + recordingRefreshCoordinator: newRecordingRefreshCoordinator(), + entered: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (c *blockingReplacementCoordinator) SetStreamCovered(stream RefreshStream, covered bool) { + c.once.Do(func() { + close(c.entered) + <-c.release + }) + c.recordingRefreshCoordinator.SetStreamCovered(stream, covered) +} + +type blockingInvalidationCoordinator struct { + *recordingRefreshCoordinator + entered chan struct{} + release chan struct{} + once sync.Once +} + +func newBlockingInvalidationCoordinator() *blockingInvalidationCoordinator { + return &blockingInvalidationCoordinator{ + recordingRefreshCoordinator: newRecordingRefreshCoordinator(), + entered: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (c *blockingInvalidationCoordinator) InvalidateStream(stream RefreshStream, reason RefreshReason) { + c.once.Do(func() { + close(c.entered) + <-c.release + }) + c.recordingRefreshCoordinator.InvalidateStream(stream, reason) +} + +func requireRefreshCoordinatorCalls(t *testing.T, coordinator *recordingRefreshCoordinator, want []refreshCoordinatorCall) { + t.Helper() + if got := coordinator.takeCalls(); !reflect.DeepEqual(got, want) { + t.Fatalf("coordinator calls = %#v, want %#v", got, want) + } +} + type blockingCloseSubscriptionManager struct { recordingSubscriptionManager closeStarted chan struct{} @@ -148,84 +276,148 @@ func TestSubscriptionWatcherDoubleStop(t *testing.T) { w.Stop() } -func TestSubscriptionWatcherSetDWNControl(t *testing.T) { - w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ - AnchorEndpoint: "https://example.com", - AnchorTenant: "did:dht:anchor", - NetworkRecordID: "record123", - SelfDID: "did:dht:self", - Signer: &dwn.Signer{DID: "did:dht:self"}, - }) - - // Before setting, notify should not panic (no-op). - w.notify() +func TestSubscriptionWatcherSetRefreshCoordinator(t *testing.T) { + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) - // Create a DWNControl with SkipStartForTests to avoid the poll loop. - cc, err := NewDWNControl( - &DWNControlConfig{ - MapResponseFunc: func(ctx context.Context) (*netmap.NetworkMap, error) { - return nil, nil - }, - PollInterval: time.Hour, - }, - controlclient.Options{SkipStartForTests: true}, - ) - if err != nil { - t.Fatalf("NewDWNControl: %v", err) + // Callbacks are harmless until engine startup supplies a coordinator. + if err := w.handleSubscriptionMessage(RefreshStreamTopology, &dwn.SubscriptionMessage{Type: "event"}); err != nil { + t.Fatalf("event without coordinator: %v", err) } - defer cc.Shutdown() - // Set the DWNControl on the watcher. - w.SetDWNControl(cc) + coordinator := newRecordingRefreshCoordinator() + w.SetRefreshCoordinator(coordinator) + if err := w.handleSubscriptionMessage(RefreshStreamTopology, &dwn.SubscriptionMessage{Type: "event"}); err != nil { + t.Fatalf("event with coordinator: %v", err) + } + requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{{ + method: "invalidate", stream: RefreshStreamTopology, reason: RefreshReasonTopology, + }}) +} - // Now notify should trigger the DWNControl's notify channel. - // This is a smoke test — we just verify it doesn't panic. - w.notify() +func TestSubscriptionWatcherReplacementCoordinatorReplaysLiveCoverage(t *testing.T) { + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) + first := newRecordingRefreshCoordinator() + w.SetRefreshCoordinator(first) + w.initializeStreamCoverage(true) + w.handleSubscriptionLifecycle(RefreshStreamTopology, dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + }) + w.handleSubscriptionLifecycle(RefreshStreamDelivery, dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + }) + first.takeCalls() + + replacement := newRecordingRefreshCoordinator() + w.SetRefreshCoordinator(replacement) + requireRefreshCoordinatorCalls(t, replacement, []refreshCoordinatorCall{ + {method: "covered", stream: RefreshStreamTopology, covered: true}, + {method: "covered", stream: RefreshStreamDelivery, covered: true}, + {method: "live", stream: RefreshStreamTopology, live: true, needsFullRefresh: true}, + {method: "live", stream: RefreshStreamDelivery, live: true, needsFullRefresh: true}, + }) } -func TestSubscriptionWatcherOnCreatedCallback(t *testing.T) { - w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ - AnchorEndpoint: "https://example.com", - AnchorTenant: "did:dht:anchor", - NetworkRecordID: "record123", - SelfDID: "did:dht:self", - Signer: &dwn.Signer{DID: "did:dht:self"}, +func TestSubscriptionWatcherReplacementHandoffDoesNotOverwriteConcurrentDisconnect(t *testing.T) { + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) + first := newRecordingRefreshCoordinator() + w.SetRefreshCoordinator(first) + w.initializeStreamCoverage(true) + w.handleSubscriptionLifecycle(RefreshStreamTopology, dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + }) + w.handleSubscriptionLifecycle(RefreshStreamDelivery, dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, }) - // Simulate the factory calling OnCreated. - var called atomic.Bool - config := &DWNControlConfig{ - MapResponseFunc: func(ctx context.Context) (*netmap.NetworkMap, error) { - return nil, nil - }, - PollInterval: time.Hour, - OnCreated: func(cc *DWNControl) { - called.Store(true) - w.SetDWNControl(cc) - }, - } + replacement := newBlockingReplacementCoordinator() + handoffDone := make(chan struct{}) + go func() { + w.SetRefreshCoordinator(replacement) + close(handoffDone) + }() + <-replacement.entered - cc, err := NewDWNControl( - config, - controlclient.Options{SkipStartForTests: true}, - ) - if err != nil { - t.Fatalf("NewDWNControl: %v", err) + disconnectDone := make(chan struct{}) + go func() { + w.handleSubscriptionLifecycle(RefreshStreamTopology, dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleRetrying, + }) + close(disconnectDone) + }() + select { + case <-disconnectDone: + t.Fatal("disconnect interleaved with an incomplete coordinator handoff") + case <-time.After(50 * time.Millisecond): } - defer cc.Shutdown() - if !called.Load() { - t.Error("OnCreated callback was not called") + close(replacement.release) + select { + case <-handoffDone: + case <-time.After(time.Second): + t.Fatal("coordinator handoff did not finish") } + select { + case <-disconnectDone: + case <-time.After(time.Second): + t.Fatal("disconnect did not reach replacement coordinator") + } + if got := replacement.streamHealth(RefreshStreamTopology); got.Live { + t.Fatalf("replacement topology health = %+v, want disconnected", got) + } +} + +func TestSubscriptionWatcherEventIsFencedByCoordinatorHandoff(t *testing.T) { + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) + first := newBlockingInvalidationCoordinator() + w.SetRefreshCoordinator(first) + w.initializeStreamCoverage(true) + w.handleSubscriptionLifecycle(RefreshStreamTopology, dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + }) + w.handleSubscriptionLifecycle(RefreshStreamDelivery, dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + }) + first.takeCalls() + + eventDone := make(chan struct{}) + go func() { + _ = w.handleSubscriptionMessage(RefreshStreamTopology, &dwn.SubscriptionMessage{Type: "event"}) + close(eventDone) + }() + <-first.entered - // The watcher should now have the DWNControl reference. - w.controlMu.Lock() - hasControl := w.dwnControl != nil - w.controlMu.Unlock() + replacement := newRecordingRefreshCoordinator() + handoffDone := make(chan struct{}) + go func() { + w.SetRefreshCoordinator(replacement) + close(handoffDone) + }() + select { + case <-handoffDone: + t.Fatal("coordinator handoff interleaved with event invalidation") + case <-time.After(50 * time.Millisecond): + } - if !hasControl { - t.Error("watcher does not have DWNControl reference after OnCreated") + close(first.release) + select { + case <-eventDone: + case <-time.After(time.Second): + t.Fatal("event invalidation did not finish") } + select { + case <-handoffDone: + case <-time.After(time.Second): + t.Fatal("coordinator handoff did not finish") + } + requireRefreshCoordinatorCalls(t, first.recordingRefreshCoordinator, []refreshCoordinatorCall{{ + method: "invalidate", stream: RefreshStreamTopology, reason: RefreshReasonTopology, + }}) + requireRefreshCoordinatorCalls(t, replacement, []refreshCoordinatorCall{ + {method: "covered", stream: RefreshStreamTopology, covered: true}, + {method: "covered", stream: RefreshStreamDelivery, covered: true}, + {method: "live", stream: RefreshStreamTopology, live: true, needsFullRefresh: true}, + {method: "live", stream: RefreshStreamDelivery, live: true, needsFullRefresh: true}, + }) } func TestDWNControlPublishesInitialDiscoKey(t *testing.T) { @@ -356,6 +548,7 @@ func TestBroadSubscriptionAuth(t *testing.T) { func TestSubscriptionWatcherStartUsesReadGrantAndRecipientDeliveryFilter(t *testing.T) { signer := &dwn.Signer{DID: "did:dht:self"} manager := &recordingSubscriptionManager{} + coordinator := newRecordingRefreshCoordinator() w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ AnchorEndpoint: "https://example.com", AnchorTenant: "did:dht:anchor", @@ -368,11 +561,18 @@ func TestSubscriptionWatcherStartUsesReadGrantAndRecipientDeliveryFilter(t *test }, }) w.newManager = func(_ string, _ *slog.Logger) subscriptionManager { return manager } + w.SetRefreshCoordinator(coordinator) if err := w.Start(context.Background()); err != nil { t.Fatalf("Start: %v", err) } defer w.Stop() + requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{ + {method: "live", stream: RefreshStreamTopology}, + {method: "live", stream: RefreshStreamDelivery}, + {method: "covered", stream: RefreshStreamTopology, covered: true}, + {method: "covered", stream: RefreshStreamDelivery, covered: true}, + }) if len(manager.calls) != 2 { t.Fatalf("subscription calls = %d, want 2", len(manager.calls)) @@ -401,8 +601,7 @@ func TestSubscriptionWatcherStartUsesReadGrantAndRecipientDeliveryFilter(t *test ProtocolPath: dwncrypto.EncryptionControlDeliveryPath, Recipient: "did:dht:self", Tags: map[string]any{ - "protocol": protocols.MeshProtocolURI, - "contextId": "network-record", + "protocol": protocols.MeshProtocolURI, }, } if !reflect.DeepEqual(deliveryCall.filter, wantDeliveryFilter) { @@ -418,6 +617,7 @@ func TestSubscriptionWatcherStartUsesReadGrantAndRecipientDeliveryFilter(t *test func TestSubscriptionWatcherRoleOnlyUsesDeliveryAndPolling(t *testing.T) { manager := &recordingSubscriptionManager{} + coordinator := newRecordingRefreshCoordinator() w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ AnchorEndpoint: "https://example.com", AnchorTenant: "did:dht:anchor", @@ -427,11 +627,18 @@ func TestSubscriptionWatcherRoleOnlyUsesDeliveryAndPolling(t *testing.T) { ReadAuth: dwn.MessageAuth{ProtocolRole: "network/node"}, }) w.newManager = func(_ string, _ *slog.Logger) subscriptionManager { return manager } + w.SetRefreshCoordinator(coordinator) if err := w.Start(context.Background()); err != nil { t.Fatalf("Start: %v", err) } defer w.Stop() + requireRefreshCoordinatorCalls(t, coordinator, []refreshCoordinatorCall{ + {method: "live", stream: RefreshStreamTopology}, + {method: "live", stream: RefreshStreamDelivery}, + {method: "covered", stream: RefreshStreamTopology, covered: false}, + {method: "covered", stream: RefreshStreamDelivery, covered: true}, + }) if len(manager.calls) != 1 { t.Fatalf("subscription calls = %d, want delivery only", len(manager.calls)) @@ -443,6 +650,7 @@ func TestSubscriptionWatcherRoleOnlyUsesDeliveryAndPolling(t *testing.T) { func TestSubscriptionWatcherSetupFailureClosesPartialSubscriptions(t *testing.T) { manager := &recordingSubscriptionManager{failCall: 2} + coordinator := newRecordingRefreshCoordinator() w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ AnchorEndpoint: "https://example.com", AnchorTenant: "did:dht:anchor", @@ -451,6 +659,13 @@ func TestSubscriptionWatcherSetupFailureClosesPartialSubscriptions(t *testing.T) Signer: &dwn.Signer{DID: "did:dht:self"}, }) w.newManager = func(_ string, _ *slog.Logger) subscriptionManager { return manager } + w.SetRefreshCoordinator(coordinator) + manager.closeHook = func() { + manager.mu.Lock() + lifecycle := manager.calls[0].lifecycle + manager.mu.Unlock() + lifecycle(dwn.SubscriptionLifecycleEvent{Kind: dwn.SubscriptionLifecycleEstablished}) + } if err := w.Start(context.Background()); err == nil { t.Fatal("Start succeeded despite delivery subscription failure") @@ -461,160 +676,171 @@ func TestSubscriptionWatcherSetupFailureClosesPartialSubscriptions(t *testing.T) if w.cancel != nil || w.manager != nil || w.done != nil { t.Fatal("failed Start retained running watcher state") } -} - -func newWatcherTestControl(t *testing.T) *DWNControl { - t.Helper() - cc, err := NewDWNControl( - &DWNControlConfig{ - MapResponseFunc: func(context.Context) (*netmap.NetworkMap, error) { - return nil, nil - }, - PollInterval: time.Hour, - }, - controlclient.Options{SkipStartForTests: true}, - ) - if err != nil { - t.Fatalf("NewDWNControl: %v", err) - } - t.Cleanup(cc.Shutdown) - return cc -} - -func expectWatcherNotification(t *testing.T, cc *DWNControl, want bool) { - t.Helper() - select { - case <-cc.notify: - if !want { - t.Fatal("unexpected refresh notification") - } - default: - if want { - t.Fatal("refresh notification was not queued") + for _, stream := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} { + health := coordinator.streamHealth(stream) + if health.Covered || health.Live { + t.Fatalf("%s stream remained available after failed Start: %#v", stream, health) } } } -func TestSubscriptionWatcherDefersReplayUntilEstablished(t *testing.T) { - w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) - cc := newWatcherTestControl(t) - w.SetDWNControl(cc) - - if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: "event"}); err != nil { - t.Fatalf("pre-establishment event: %v", err) - } - if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: "eose"}); err != nil { - t.Fatalf("pre-establishment EOSE: %v", err) - } - expectWatcherNotification(t, cc, false) - - w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleEstablished, - NeedsFullRefresh: false, - }) - expectWatcherNotification(t, cc, true) - - if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: "event"}); err != nil { - t.Fatalf("live event: %v", err) +func TestSubscriptionWatcherMessageCallbacks(t *testing.T) { + tests := []struct { + name string + stream RefreshStream + message *dwn.SubscriptionMessage + want []refreshCoordinatorCall + }{ + {name: "nil message", stream: RefreshStreamTopology}, + { + name: "topology event", + stream: RefreshStreamTopology, + message: &dwn.SubscriptionMessage{Type: dwn.SubscriptionEventType}, + want: []refreshCoordinatorCall{{ + method: "invalidate", stream: RefreshStreamTopology, reason: RefreshReasonTopology, + }}, + }, + { + name: "delivery event", + stream: RefreshStreamDelivery, + message: &dwn.SubscriptionMessage{Type: dwn.SubscriptionEventType}, + want: []refreshCoordinatorCall{{ + method: "invalidate", stream: RefreshStreamDelivery, reason: RefreshReasonDelivery, + }}, + }, + { + name: "end of stored events", + stream: RefreshStreamTopology, + message: &dwn.SubscriptionMessage{Type: dwn.SubscriptionEOSEType}, + }, + { + name: "error frame is lifecycle owned", + stream: RefreshStreamTopology, + message: &dwn.SubscriptionMessage{ + Type: "error", + Error: &dwn.SubscriptionError{Code: "SubscriptionFailed", Detail: "closed"}, + }, + }, + {name: "unknown frame", stream: RefreshStreamDelivery, message: &dwn.SubscriptionMessage{Type: "future"}}, } - expectWatcherNotification(t, cc, true) - w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleRetrying, - Attempt: 1, - }) - if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: "event"}); err != nil { - t.Fatalf("replay event: %v", err) + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + coordinator := newRecordingRefreshCoordinator() + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) + w.SetRefreshCoordinator(coordinator) + if err := w.handleSubscriptionMessage(test.stream, test.message); err != nil { + t.Fatalf("handleSubscriptionMessage: %v", err) + } + requireRefreshCoordinatorCalls(t, coordinator, test.want) + }) } - expectWatcherNotification(t, cc, false) - w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleEstablished, - NeedsFullRefresh: false, - }) - expectWatcherNotification(t, cc, true) } -func TestSubscriptionWatcherEOSEBarrierRefreshesOnlyAfterReplay(t *testing.T) { +func TestSubscriptionWatcherLifecycleCallbacks(t *testing.T) { tests := []struct { - name string - replayEvent bool - wantRefresh bool + name string + stream RefreshStream + event dwn.SubscriptionLifecycleEvent + want []refreshCoordinatorCall }{ - {name: "empty replay", wantRefresh: false}, - {name: "replayed event", replayEvent: true, wantRefresh: true}, + { + name: "established", + stream: RefreshStreamTopology, + event: dwn.SubscriptionLifecycleEvent{Kind: dwn.SubscriptionLifecycleEstablished}, + want: []refreshCoordinatorCall{{method: "live", stream: RefreshStreamTopology, live: true}}, + }, + { + name: "established requires full refresh", + stream: RefreshStreamDelivery, + event: dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleEstablished, + NeedsFullRefresh: true, + }, + want: []refreshCoordinatorCall{{ + method: "live", stream: RefreshStreamDelivery, live: true, needsFullRefresh: true, + }}, + }, + { + name: "progress gap", + stream: RefreshStreamDelivery, + event: dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleProgressGap, + Gap: &dwn.ProgressGapInfo{Reason: "compacted"}, + }, + want: []refreshCoordinatorCall{{ + method: "live", stream: RefreshStreamDelivery, needsFullRefresh: true, + }}, + }, + { + name: "retrying preserves pending invalidation", + stream: RefreshStreamTopology, + event: dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleRetrying, Attempt: 2, Err: errors.New("offline"), + }, + want: []refreshCoordinatorCall{{method: "live", stream: RefreshStreamTopology}}, + }, + { + name: "terminal reconciles once", + stream: RefreshStreamTopology, + event: dwn.SubscriptionLifecycleEvent{ + Kind: dwn.SubscriptionLifecycleTerminal, Err: errors.New("denied"), + }, + want: []refreshCoordinatorCall{ + {method: "live", stream: RefreshStreamTopology}, + {method: "notify", reason: RefreshReasonTopology}, + }, + }, + {name: "unknown lifecycle", stream: RefreshStreamDelivery, event: dwn.SubscriptionLifecycleEvent{Kind: "future"}}, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { + coordinator := newRecordingRefreshCoordinator() w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) - cc := newWatcherTestControl(t) - w.SetDWNControl(cc) - - if test.replayEvent { - if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: dwn.SubscriptionEventType}); err != nil { - t.Fatalf("replay event: %v", err) - } - } - if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{Type: dwn.SubscriptionEOSEType}); err != nil { - t.Fatalf("EOSE: %v", err) - } - expectWatcherNotification(t, cc, false) - - w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleEstablished, - NeedsFullRefresh: false, - }) - expectWatcherNotification(t, cc, test.wantRefresh) - // Replay event + EOSE is one invalidation batch, never two. - expectWatcherNotification(t, cc, false) + w.SetRefreshCoordinator(coordinator) + w.handleSubscriptionLifecycle(test.stream, test.event) + requireRefreshCoordinatorCalls(t, coordinator, test.want) }) } } -func TestSubscriptionWatcherGapWaitsForFreshEstablishment(t *testing.T) { - w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) - cc := newWatcherTestControl(t) - w.SetDWNControl(cc) - - w.handleSubscriptionLifecycle("delivery", dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleEstablished, - NeedsFullRefresh: false, - }) - expectWatcherNotification(t, cc, false) - w.handleSubscriptionLifecycle("delivery", dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleProgressGap, - Gap: &dwn.ProgressGapInfo{Reason: "compacted"}, - }) - expectWatcherNotification(t, cc, false) - w.handleSubscriptionLifecycle("delivery", dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleEstablished, - NeedsFullRefresh: true, +func TestSubscriptionWatcherStopFencesLifecycleCallbacks(t *testing.T) { + manager := &recordingSubscriptionManager{} + coordinator := newRecordingRefreshCoordinator() + w := NewSubscriptionWatcher(SubscriptionWatcherConfig{ + AnchorEndpoint: "https://example.com", + AnchorTenant: "did:dht:anchor", + NetworkRecordID: "network-record", + SelfDID: "did:dht:self", + Signer: &dwn.Signer{DID: "did:dht:self"}, }) - expectWatcherNotification(t, cc, true) -} - -func TestSubscriptionWatcherLiveErrorAndTerminalReconcile(t *testing.T) { - w := NewSubscriptionWatcher(SubscriptionWatcherConfig{}) - cc := newWatcherTestControl(t) - w.SetDWNControl(cc) + w.newManager = func(_ string, _ *slog.Logger) subscriptionManager { return manager } + w.SetRefreshCoordinator(coordinator) + if err := w.Start(context.Background()); err != nil { + t.Fatalf("Start: %v", err) + } + coordinator.takeCalls() - w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleEstablished, - NeedsFullRefresh: false, - }) - if err := w.handleSubscriptionMessage("mesh", &dwn.SubscriptionMessage{ - Type: "error", - Error: &dwn.SubscriptionError{Code: "SubscriptionFailed", Detail: "closed"}, - }); err != nil { - t.Fatalf("live error: %v", err) + manager.closeHook = func() { + manager.mu.Lock() + calls := append([]recordedSubscription(nil), manager.calls...) + manager.mu.Unlock() + for _, call := range calls { + call.lifecycle(dwn.SubscriptionLifecycleEvent{Kind: dwn.SubscriptionLifecycleEstablished}) + } } - expectWatcherNotification(t, cc, true) + w.Stop() - w.handleSubscriptionLifecycle("mesh", dwn.SubscriptionLifecycleEvent{ - Kind: dwn.SubscriptionLifecycleTerminal, - Err: errors.New("stream terminated"), - }) - expectWatcherNotification(t, cc, true) + if manager.closeCall.Load() != 1 { + t.Fatalf("CloseAll calls = %d, want 1", manager.closeCall.Load()) + } + for _, stream := range []RefreshStream{RefreshStreamTopology, RefreshStreamDelivery} { + health := coordinator.streamHealth(stream) + if health.Covered || health.Live { + t.Fatalf("%s stream remained available after Stop: %#v", stream, health) + } + } } func TestSubscriptionWatcherStartFailsGracefully(t *testing.T) {