diff --git a/pkg/daemon/platform/daemon.go b/pkg/daemon/platform/daemon.go index 0e3fdc2b1..ba95f85c2 100644 --- a/pkg/daemon/platform/daemon.go +++ b/pkg/daemon/platform/daemon.go @@ -17,8 +17,8 @@ import ( "github.com/devsy-org/devsy/pkg/platform/client" "github.com/devsy-org/devsy/pkg/ts" "tailscale.com/client/local" + "tailscale.com/ipn/ipnstate" "tailscale.com/tsnet" - "tailscale.com/types/netmap" ) type Daemon struct { @@ -166,10 +166,10 @@ func (d *Daemon) watchNetmap(ctx context.Context) error { return err } - return ts.WatchNetmap(ctx, lc, func(netMap *netmap.NetworkMap) { - nm, err := json.Marshal(netMap) + return ts.WatchNetmap(ctx, lc, func(status *ipnstate.Status) { + nm, err := json.Marshal(status) if err != nil { - log.Errorf("Failed to marshal netmap: %v", err) + log.Errorf("failed to marshal netmap: %v", err) } else { _ = os.WriteFile(filepath.Join(d.rootDir, "netmap.json"), nm, 0o644) } diff --git a/pkg/ts/util.go b/pkg/ts/util.go index bc148846c..b575c6166 100644 --- a/pkg/ts/util.go +++ b/pkg/ts/util.go @@ -12,7 +12,7 @@ import ( "github.com/devsy-org/devsy/pkg/log" "tailscale.com/client/local" "tailscale.com/ipn" - "tailscale.com/types/netmap" + "tailscale.com/ipn/ipnstate" ) // DevsyTSNetDomain is the MagicDNS suffix for the Devsy tailnet. @@ -80,34 +80,88 @@ func WaitHostReachable( return fmt.Errorf("host %s not reachable", addr.String()) } +// ipnWatcher is the subset of *local.IPNBusWatcher used by watchNetmap. +type ipnWatcher interface { + Next() (ipn.Notify, error) +} + +// WatchNetmap invokes netmapChangedFn whenever the tailnet state changes. func WatchNetmap( ctx context.Context, lc *local.Client, - netmapChangedFn func(nm *netmap.NetworkMap), + netmapChangedFn func(status *ipnstate.Status), ) error { watcher, err := lc.WatchIPNBus( ctx, - ipn.NotifyInitialNetMap|ipn.NotifyRateLimit|ipn.NotifyWatchEngineUpdates, + ipn.NotifyInitialStatus|ipn.NotifyWatchEngineUpdates|ipn.NotifyPeerChanges, ) if err != nil { return err } defer func() { _ = watcher.Close() }() - var netMap *netmap.NetworkMap + return watchNetmap(ctx, watcher, lc.Status, netmapChangedFn) +} + +// watchNetmap drains watcher on a dedicated goroutine so notifications keep +// flowing while fetchStatus is in flight, coalescing any changes that arrive +// during a fetch into a single follow-up call. Without this, a burst of +// notifications (e.g. NotifyPeerChanges deltas) can fill the IPN bus's +// 128-entry queue while netmapChangedFn's caller blocks in fetchStatus, +// causing tailscaled to close the watch ("IPN bus consumer fell behind"). +func watchNetmap( + ctx context.Context, + watcher ipnWatcher, + fetchStatus func(context.Context) (*ipnstate.Status, error), + netmapChangedFn func(status *ipnstate.Status), +) error { + trigger := make(chan struct{}, 1) + errc := make(chan error, 1) + + go drainNotifications(watcher, trigger, errc) + + for { + select { + case err := <-errc: + return err + case <-trigger: + status, err := fetchStatus(ctx) + if err != nil { + return fmt.Errorf("fetch status: %w", err) + } + netmapChangedFn(status) + } + } +} + +// drainNotifications continuously reads watcher.Next(), coalescing every +// relevant change into a non-blocking signal on trigger, until watcher +// reports a terminal error (or a bus-side ErrMessage) on errc. +func drainNotifications(watcher ipnWatcher, trigger chan<- struct{}, errc chan<- error) { for { n, err := watcher.Next() if err != nil { - return fmt.Errorf("watch ipn: %w", err) + errc <- fmt.Errorf("watch ipn: %w", err) + return } if n.ErrMessage != nil { - return fmt.Errorf("tailscale error: %w", errors.New(*n.ErrMessage)) + errc <- fmt.Errorf("tailscale error: %w", errors.New(*n.ErrMessage)) + return } - if n.NetMap != nil { - if n.NetMap != netMap { - netMap = n.NetMap - netmapChangedFn(netMap) - } + if !netmapChanged(n) { + continue + } + select { + case trigger <- struct{}{}: + default: } } } + +// netmapChanged reports whether a bus notification indicates a tailnet +// state change. +func netmapChanged(n ipn.Notify) bool { + return n.InitialStatus != nil || n.SelfChange != nil || + len(n.PeersChanged) > 0 || len(n.PeersRemoved) > 0 || + len(n.PeerChangedPatch) > 0 +} diff --git a/pkg/ts/util_test.go b/pkg/ts/util_test.go new file mode 100644 index 000000000..e24c5fdef --- /dev/null +++ b/pkg/ts/util_test.go @@ -0,0 +1,128 @@ +package ts + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "tailscale.com/ipn" + "tailscale.com/ipn/ipnstate" + "tailscale.com/tailcfg" +) + +// waitUntil polls cond until it reports true, failing the test if it does +// not become true within timeout. +func waitUntil(t *testing.T, timeout time.Duration, msg string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for !cond() { + if time.Now().After(deadline) { + t.Fatal(msg) + } + time.Sleep(time.Millisecond) + } +} + +func waitFor(t *testing.T, timeout time.Duration, msg string, ch <-chan struct{}) { + t.Helper() + select { + case <-ch: + case <-time.After(timeout): + t.Fatal(msg) + } +} + +func selfChangeNotifs(n int) []ipn.Notify { + notifs := make([]ipn.Notify, n) + for i := range notifs { + notifs[i] = ipn.Notify{SelfChange: &tailcfg.Node{}} + } + return notifs +} + +// fakeIPNWatcher replays a fixed sequence of notifications, then blocks until +// the test signals it to report the watch as closed. +type fakeIPNWatcher struct { + notifs []ipn.Notify + idx atomic.Int32 + afterFirst chan struct{} + closed chan struct{} +} + +func (w *fakeIPNWatcher) Next() (ipn.Notify, error) { + i := w.idx.Load() + if i == 1 { + <-w.afterFirst + } + if int(i) < len(w.notifs) { + w.idx.Add(1) + return w.notifs[i], nil + } + <-w.closed + return ipn.Notify{}, errors.New("watcher closed") +} + +// TestWatchNetmapDrainsBurstWhileStatusFetchBlocked verifies that a burst of +// notifications arriving while fetchStatus is in flight does not stall +// notification draining: watchNetmap must keep calling watcher.Next() so a +// burst larger than the IPN bus's 128-entry queue never causes tailscaled to +// close the watch, and must coalesce the burst into a single follow-up fetch. +func TestWatchNetmapDrainsBurstWhileStatusFetchBlocked(t *testing.T) { + const burst = 129 + + watcher := &fakeIPNWatcher{ + notifs: selfChangeNotifs(burst), + afterFirst: make(chan struct{}), + closed: make(chan struct{}), + } + + firstFetchStarted := make(chan struct{}) + unblockFirstFetch := make(chan struct{}) + var fetchCount atomic.Int32 + + fetchStatus := func(context.Context) (*ipnstate.Status, error) { //nolint:unparam // exercises the success path only + if fetchCount.Add(1) == 1 { + close(firstFetchStarted) + <-unblockFirstFetch + } + return &ipnstate.Status{}, nil + } + + var callbackCount atomic.Int32 + callback := func(*ipnstate.Status) { callbackCount.Add(1) } + + errc := make(chan error, 1) + go func() { + errc <- watchNetmap(context.Background(), watcher, fetchStatus, callback) + }() + + waitFor(t, 5*time.Second, "first status fetch never started", firstFetchStarted) + close(watcher.afterFirst) + + waitUntil(t, 5*time.Second, "watcher stalled while status fetch was blocked", func() bool { + return watcher.idx.Load() == burst + }) + + close(unblockFirstFetch) + + waitUntil(t, 5*time.Second, "expected a coalesced follow-up fetch", func() bool { + return fetchCount.Load() >= 2 + }) + + if got := fetchCount.Load(); got != 2 { + t.Errorf( + "fetchStatus called %d times, want exactly 2 (initial + one coalesced follow-up)", + got, + ) + } + if got := callbackCount.Load(); got != 2 { + t.Errorf("callback invoked %d times, want 2", got) + } + + close(watcher.closed) + if err := <-errc; err == nil { + t.Fatal("watchNetmap returned nil error after watcher closed, want an error") + } +} diff --git a/pkg/ts/workspace_server.go b/pkg/ts/workspace_server.go index 2836db2aa..01da5c28b 100644 --- a/pkg/ts/workspace_server.go +++ b/pkg/ts/workspace_server.go @@ -20,9 +20,9 @@ import ( sshServer "github.com/devsy-org/devsy/pkg/ssh/server" "tailscale.com/client/local" "tailscale.com/envknob" + "tailscale.com/ipn/ipnstate" "tailscale.com/ipn/store/mem" "tailscale.com/tsnet" - "tailscale.com/types/netmap" ) const ( @@ -87,20 +87,20 @@ func (s *WorkspaceServer) Start(ctx context.Context) error { go func() { lastUpdate := time.Now() - if err := WatchNetmap(ctx, lc, func(netMap *netmap.NetworkMap) { + if err := WatchNetmap(ctx, lc, func(status *ipnstate.Status) { if time.Since(lastUpdate) < netMapCooldown { return } lastUpdate = time.Now() - nm, err := json.Marshal(netMap) + nm, err := json.Marshal(status) if err != nil { - log.Errorf("Failed to marshal netmap: %v", err) + log.Errorf("failed to marshal netmap: %v", err) } else { _ = os.WriteFile(filepath.Join(s.config.RootDir, "netmap.json"), nm, 0o644) } }); err != nil { - log.Errorf("Failed to watch netmap: %v", err) + log.Errorf("failed to watch netmap: %v", err) } }()