fix: replace deprecated tailscale n.NetMap usage

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)
 		}

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
+}

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")
+	}
+}

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)
 		}
 	}()