From 8ec17425dcf1d2206635419c21aa88879749036c Mon Sep 17 00:00:00 2001 From: Samuel K Date: Wed, 26 Aug 2026 12:43:05 +0000 Subject: [PATCH 1/5] fix: replace deprecated tailscale n.NetMap usage WatchNetmap subscribes to NotifyInitialStatus and reacts to SelfChange and peer deltas, fetching a fresh ipnstate.Status on demand via LocalClient.Status instead of reading the deprecated ipn.Notify.NetMap bus field (staticcheck SA1019). Both consumers write the same netmap.json debug snapshot. --- pkg/daemon/platform/daemon.go | 8 ++++---- pkg/ts/util.go | 32 +++++++++++++++++++++++--------- pkg/ts/workspace_server.go | 10 +++++----- 3 files changed, 32 insertions(+), 18 deletions(-) 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..ec64ec5cb 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,21 +80,24 @@ func WaitHostReachable( return fmt.Errorf("host %s not reachable", addr.String()) } +// WatchNetmap invokes netmapChangedFn whenever the tailnet state changes: +// first with the initial status, then on self or peer updates. The full +// snapshot is fetched on demand via LocalClient.Status instead of reading +// the deprecated ipn.Notify.NetMap bus field. 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, ) if err != nil { return err } defer func() { _ = watcher.Close() }() - var netMap *netmap.NetworkMap for { n, err := watcher.Next() if err != nil { @@ -103,11 +106,22 @@ func WatchNetmap( if n.ErrMessage != nil { return fmt.Errorf("tailscale error: %w", errors.New(*n.ErrMessage)) } - if n.NetMap != nil { - if n.NetMap != netMap { - netMap = n.NetMap - netmapChangedFn(netMap) - } + if !netmapChanged(n) { + continue } + status, err := lc.Status(ctx) + if err != nil { + return fmt.Errorf("fetch status: %w", err) + } + netmapChangedFn(status) } } + +// netmapChanged reports whether a bus notification indicates a tailnet +// state change, mirroring how upstream consumers react to InitialStatus, +// SelfChange, and peer deltas. +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/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) } }() From 5955c29de15e6328ec6e7cfb6e22082031dfb245 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Wed, 26 Aug 2026 11:15:36 -0500 Subject: [PATCH 2/5] style: update comment --- pkg/ts/util.go | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/pkg/ts/util.go b/pkg/ts/util.go index ec64ec5cb..d72375983 100644 --- a/pkg/ts/util.go +++ b/pkg/ts/util.go @@ -80,10 +80,7 @@ func WaitHostReachable( return fmt.Errorf("host %s not reachable", addr.String()) } -// WatchNetmap invokes netmapChangedFn whenever the tailnet state changes: -// first with the initial status, then on self or peer updates. The full -// snapshot is fetched on demand via LocalClient.Status instead of reading -// the deprecated ipn.Notify.NetMap bus field. +// WatchNetmap invokes netmapChangedFn whenever the tailnet state changes. func WatchNetmap( ctx context.Context, lc *local.Client, @@ -118,8 +115,7 @@ func WatchNetmap( } // netmapChanged reports whether a bus notification indicates a tailnet -// state change, mirroring how upstream consumers react to InitialStatus, -// SelfChange, and peer deltas. +// state change. func netmapChanged(n ipn.Notify) bool { return n.InitialStatus != nil || n.SelfChange != nil || len(n.PeersChanged) > 0 || len(n.PeersRemoved) > 0 || From a4c19cec6619e7f0cbfa7d719dde02832aa32f43 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Thu, 27 Aug 2026 21:06:37 -0500 Subject: [PATCH 3/5] fix: subscribe to NotifyPeerChanges in WatchNetmap Peer-only membership changes (join/leave/rename without SelfChange) never refreshed netmap.json: the watch mask lacked ipn.NotifyPeerChanges, so tailscaled strips PeersChanged/ PeersRemoved/PeerChangedPatch before delivery, making the peer-delta checks in netmapChanged dead code. Adds ipn.NotifyPeerChanges to the WatchIPNBus mask in pkg/ts.WatchNetmap. --- pkg/ts/util.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/ts/util.go b/pkg/ts/util.go index d72375983..96eb248d5 100644 --- a/pkg/ts/util.go +++ b/pkg/ts/util.go @@ -88,7 +88,7 @@ func WatchNetmap( ) error { watcher, err := lc.WatchIPNBus( ctx, - ipn.NotifyInitialStatus|ipn.NotifyWatchEngineUpdates, + ipn.NotifyInitialStatus|ipn.NotifyWatchEngineUpdates|ipn.NotifyPeerChanges, ) if err != nil { return err From b892b105e74731b124c1131772bd58507f251568 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Thu, 27 Aug 2026 21:21:00 -0500 Subject: [PATCH 4/5] fix: decouple netmap status fetch from IPN bus drain WatchNetmap blocked in LocalClient.Status while calling watcher.Next() in the same loop. A burst of notifications (routine with NotifyPeerChanges now enabled) can outrun the IPN bus's 128-entry queue during that blocking fetch, causing tailscaled to close the watch ("IPN bus consumer fell behind") and killing the daemon/ workspace server, since neither caller retries WatchNetmap. watchNetmap now drains watcher.Next() on a dedicated goroutine and coalesces bursts into a single follow-up status fetch via a buffered(1) trigger channel, so notification consumption never stalls behind a status fetch. Adds a regression test simulating a 129- notification burst while a status fetch is blocked. --- pkg/ts/util.go | 56 +++++++++++++++++++--- pkg/ts/util_test.go | 111 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 161 insertions(+), 6 deletions(-) create mode 100644 pkg/ts/util_test.go diff --git a/pkg/ts/util.go b/pkg/ts/util.go index 96eb248d5..b575c6166 100644 --- a/pkg/ts/util.go +++ b/pkg/ts/util.go @@ -80,6 +80,11 @@ 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, @@ -95,22 +100,61 @@ func WatchNetmap( } defer func() { _ = watcher.Close() }() + 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 !netmapChanged(n) { continue } - status, err := lc.Status(ctx) - if err != nil { - return fmt.Errorf("fetch status: %w", err) + select { + case trigger <- struct{}{}: + default: } - netmapChangedFn(status) } } diff --git a/pkg/ts/util_test.go b/pkg/ts/util_test.go new file mode 100644 index 000000000..eb6840fef --- /dev/null +++ b/pkg/ts/util_test.go @@ -0,0 +1,111 @@ +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) + } +} + +// 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 + closed chan struct{} +} + +func (w *fakeIPNWatcher) Next() (ipn.Notify, error) { + i := w.idx.Load() + 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 + + notifs := make([]ipn.Notify, burst) + for i := range notifs { + notifs[i] = ipn.Notify{SelfChange: &tailcfg.Node{}} + } + watcher := &fakeIPNWatcher{notifs: notifs, 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 + n := fetchCount.Add(1) + if n == 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) + }() + + select { + case <-firstFetchStarted: + case <-time.After(5 * time.Second): + t.Fatal("first status fetch never started") + } + + 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") + } +} From 3969f4d2bdd7a532acc6f993c8fa09950155bc31 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Thu, 27 Aug 2026 22:42:20 -0500 Subject: [PATCH 5/5] fix: eliminate test race in burst-drain regression test Gate fakeIPNWatcher.Next after the first notification until the test observes the first fetchStatus call starting. Without this, the drain goroutine could exhaust all 129 notifications before watchNetmap's select loop ever ran, leaving no notification produced after the first one is consumed and no second trigger to coalesce -- an intermittent failure against correct code. --- pkg/ts/util_test.go | 45 +++++++++++++++++++++++++++++++-------------- 1 file changed, 31 insertions(+), 14 deletions(-) diff --git a/pkg/ts/util_test.go b/pkg/ts/util_test.go index eb6840fef..e24c5fdef 100644 --- a/pkg/ts/util_test.go +++ b/pkg/ts/util_test.go @@ -25,16 +25,37 @@ func waitUntil(t *testing.T, timeout time.Duration, msg string, cond func() bool } } +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 - closed chan 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 @@ -51,19 +72,18 @@ func (w *fakeIPNWatcher) Next() (ipn.Notify, error) { func TestWatchNetmapDrainsBurstWhileStatusFetchBlocked(t *testing.T) { const burst = 129 - notifs := make([]ipn.Notify, burst) - for i := range notifs { - notifs[i] = ipn.Notify{SelfChange: &tailcfg.Node{}} + watcher := &fakeIPNWatcher{ + notifs: selfChangeNotifs(burst), + afterFirst: make(chan struct{}), + closed: make(chan struct{}), } - watcher := &fakeIPNWatcher{notifs: notifs, 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 - n := fetchCount.Add(1) - if n == 1 { + if fetchCount.Add(1) == 1 { close(firstFetchStarted) <-unblockFirstFetch } @@ -78,11 +98,8 @@ func TestWatchNetmapDrainsBurstWhileStatusFetchBlocked(t *testing.T) { errc <- watchNetmap(context.Background(), watcher, fetchStatus, callback) }() - select { - case <-firstFetchStarted: - case <-time.After(5 * time.Second): - t.Fatal("first status fetch never started") - } + 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