ipn: enforce lossless IPN bus delta streams

New-style IPN bus subscribers consume stateful delta streams. Reject
NotifyRateLimit when it is combined with those subscription bits so
tailscaled cannot merge or delay messages that clients need to apply in
order.

Also stop silently dropping notifications when a watcher falls behind.
Remove the watcher, replace its stale queue with one terminal ErrMessage
notification, and close the watch.

Updates #20062

Change-Id: Id9d402ea76f4011cd23f122adf62f30dd4b6f90b
Signed-off-by: Brad Fitzpatrick <bradfitz@tailscale.com>

ipn/ipnlocal/bus.go

@@ -218,7 +218,11 @@ func mergePeerChangeForIpnBus(old, new *tailcfg.PeerChange) *tailcfg.PeerChange
 // should be sent on the IPN bus immediately (e.g. to GUIs) without
 // rate limiting it for a few seconds.
 //
-// PeerChanges and Engine are the only "boring" (rate-limitable) fields.
+// This is only used for legacy [ipn.NotifyRateLimit] subscribers. New-style
+// subscriptions that receive delta streams are rejected by
+// [ipn.ValidateNotifyWatchOpt] when combined with NotifyRateLimit.
+//
+// Legacy NetMap and Engine are the only "boring" (rate-limitable) fields.
 func isNotableNotify(n *ipn.Notify) bool {
 	if n == nil {
 		return false
@@ -233,6 +237,10 @@ func isNotableNotify(n *ipn.Notify) bool {
 		n.LoginFinished != nil ||
 		n.SelfChange != nil ||
 		n.InitialStatus != nil ||
+		len(n.PeerChangedPatch) > 0 ||
+		len(n.PeersChanged) > 0 ||
+		len(n.PeersRemoved) > 0 ||
+		len(n.UserProfiles) > 0 ||
 		!n.DriveShares.IsNil() ||
 		n.Health != nil ||
 		len(n.IncomingFiles) > 0 ||

ipn/ipnlocal/bus_test.go

@@ -30,10 +30,10 @@ func TestIsNotableNotify(t *testing.T) {
 		{"empty", &ipn.Notify{}, false},
 		{"version", &ipn.Notify{Version: "foo"}, false},
 		{"netmap", &ipn.Notify{NetMap: new(netmap.NetworkMap)}, false},
-		{"peerchanges", &ipn.Notify{PeerChangedPatch: []*tailcfg.PeerChange{{}}}, false},
-		{"peerschanged", &ipn.Notify{PeersChanged: []*tailcfg.Node{{}}}, false},
-		{"peersremoved", &ipn.Notify{PeersRemoved: []tailcfg.NodeID{1}}, false},
-		{"userprofiles", &ipn.Notify{UserProfiles: map[tailcfg.UserID]tailcfg.UserProfileView{1: (&tailcfg.UserProfile{}).View()}}, false},
+		{"peerchanges", &ipn.Notify{PeerChangedPatch: []*tailcfg.PeerChange{{}}}, true},
+		{"peerschanged", &ipn.Notify{PeersChanged: []*tailcfg.Node{{}}}, true},
+		{"peersremoved", &ipn.Notify{PeersRemoved: []tailcfg.NodeID{1}}, true},
+		{"userprofiles", &ipn.Notify{UserProfiles: map[tailcfg.UserID]tailcfg.UserProfileView{1: (&tailcfg.UserProfile{}).View()}}, true},
 		{"engine", &ipn.Notify{Engine: new(ipn.EngineStatus)}, false},
 		{"selfchange", &ipn.Notify{SelfChange: &tailcfg.Node{}}, true},
 	}
@@ -126,20 +126,18 @@ func (st *rateLimitingBusSenderTester) advance(d time.Duration) {
 }
 
 func TestRateLimitingBusSender(t *testing.T) {
-	// Both share NodeID 1 so merge collapses to a single PeerChange and
-	// the later one (nm2) wins.
-	nm1 := &ipn.Notify{PeerChangedPatch: []*tailcfg.PeerChange{{NodeID: 1, DERPRegion: 1}}}
-	nm2 := &ipn.Notify{PeerChangedPatch: []*tailcfg.PeerChange{{NodeID: 1, DERPRegion: 2}}}
+	ver1 := &ipn.Notify{Version: "1"}
+	ver2 := &ipn.Notify{Version: "2"}
 	eng1 := &ipn.Notify{Engine: new(ipn.EngineStatus)}
 	eng2 := &ipn.Notify{Engine: new(ipn.EngineStatus)}
 
 	t.Run("unbuffered", func(t *testing.T) {
 		st := &rateLimitingBusSenderTester{tb: t}
-		st.send(nm1)
-		st.send(nm2)
+		st.send(ver1)
+		st.send(ver2)
 		st.send(eng1)
 		st.send(eng2)
-		if !slices.Equal(st.got, []*ipn.Notify{nm1, nm2, eng1, eng2}) {
+		if !slices.Equal(st.got, []*ipn.Notify{ver1, ver2, eng1, eng2}) {
 			t.Errorf("got %d items; want 4 specific ones, unmodified", len(st.got))
 		}
 	})
@@ -152,8 +150,8 @@ func TestRateLimitingBusSender(t *testing.T) {
 		if len(st.got) != 1 {
 			t.Fatalf("got %d items; expected 1 (first to flush immediately)", len(st.got))
 		}
-		st.send(nm1)
-		st.send(nm2)
+		st.send(ver1)
+		st.send(ver2)
 		st.send(eng1)
 		st.send(eng2)
 		if len(st.got) != 1 {
@@ -168,8 +166,8 @@ func TestRateLimitingBusSender(t *testing.T) {
 			t.Fatalf("got %d items; want 2", len(st.got))
 		}
 		gotn := st.got[1]
-		if !reflect.DeepEqual(gotn.PeerChangedPatch, nm2.PeerChangedPatch) {
-			t.Errorf("got wrong PeerChangedPatch; got %v want %v", gotn.PeerChangedPatch, nm2.PeerChangedPatch)
+		if gotn.Version != ver1.Version {
+			t.Errorf("got wrong Version; got %q want %q", gotn.Version, ver1.Version)
 		}
 		if gotn.Engine != eng2.Engine {
 			t.Errorf("got wrong Engine; got %p", gotn.Engine)
@@ -206,15 +204,18 @@ func TestRateLimitingBusSender(t *testing.T) {
 
 		incoming := make(chan *ipn.Notify, 2)
 		go func() {
-			incoming <- nm1
+			incoming <- ver1
 			waitSend()
-			incoming <- nm2
+			incoming <- eng2
 			waitSend()
 			st.advance(5 * time.Second)
 			select {
 			case n := <-flushc:
-				if !reflect.DeepEqual(n.PeerChangedPatch, nm2.PeerChangedPatch) {
-					t.Errorf("got wrong PeerChangedPatch; got %v want %v", n.PeerChangedPatch, nm2.PeerChangedPatch)
+				if n.Version != ver1.Version {
+					t.Errorf("got wrong Version; got %q want %q", n.Version, ver1.Version)
+				}
+				if n.Engine != eng2.Engine {
+					t.Errorf("got wrong Engine; got %p", n.Engine)
 				}
 			case <-time.After(10 * time.Second):
 				t.Error("timeout")

ipn/ipnlocal/local.go

@@ -3587,17 +3587,24 @@ func applyConfigToHostinfo(hi *tailcfg.Hostinfo, c *conffile.Config) {
 // called with non-nil pointers. The caller must not modify roNotify. If
 // fn returns false, the watch also stops.
 //
-// Failure to consume many notifications in a row will result in dropped
-// notifications. There is currently (2022-11-22) no mechanism provided to
-// detect when a message has been dropped.
+// Failure to consume many notifications in a row will result in one final
+// notification with ErrMessage set, followed by the watch closing.
 func (b *LocalBackend) WatchNotifications(ctx context.Context, mask ipn.NotifyWatchOpt, onWatchAdded func(), fn func(roNotify *ipn.Notify) (keepGoing bool)) {
 	b.WatchNotificationsAs(ctx, nil, mask, onWatchAdded, fn)
 }
 
+const watchIPNBusFellBehindMessage = "IPN bus consumer fell behind; closing watch"
+
 // WatchNotificationsAs is like [LocalBackend.WatchNotifications] but takes an [ipnauth.Actor]
 // as an additional parameter. If non-nil, the specified callback is invoked
 // only for notifications relevant to this actor.
 func (b *LocalBackend) WatchNotificationsAs(ctx context.Context, actor ipnauth.Actor, mask ipn.NotifyWatchOpt, onWatchAdded func(), fn func(roNotify *ipn.Notify) (keepGoing bool)) {
+	if err := ipn.ValidateNotifyWatchOpt(mask); err != nil {
+		msg := err.Error()
+		fn(&ipn.Notify{Version: version.Long(), ErrMessage: &msg})
+		return
+	}
+
 	ch := make(chan *ipn.Notify, 128)
 	sessionID := rands.HexString(16)
 	if mask&ipn.NotifyHealthActions == 0 {
@@ -3916,7 +3923,32 @@ func (b *LocalBackend) sendToLocked(n ipn.Notify, recipient notificationTarget)
 		select {
 		case sess.ch <- nForSess:
 		default:
-			// Drop the notification if the channel is full.
+			b.closeLaggingWatchSessionLocked(sess)
+		}
+	}
+}
+
+// closeLaggingWatchSessionLocked removes sess from the active watcher set and
+// arranges for its consumer to receive one final terminal error notification.
+//
+// b.mu must be held.
+func (b *LocalBackend) closeLaggingWatchSessionLocked(sess *watchSession) {
+	delete(b.notifyWatchers, sess.sessionID)
+
+	// The session already fell behind, so the queued delta stream is not
+	// trustworthy. Drop queued messages and replace them with a terminal
+	// notification.
+	for {
+		select {
+		case <-sess.ch:
+		default:
+			msg := watchIPNBusFellBehindMessage
+			sess.ch <- &ipn.Notify{
+				Version:    version.Long(),
+				ErrMessage: &msg,
+			}
+			close(sess.ch)
+			return
 		}
 	}
 }

ipn/ipnlocal/local_test.go

@@ -2116,6 +2116,66 @@ func TestWatchNotificationsCallbacks(t *testing.T) {
 	}
 }
 
+func TestWatchNotificationsClosesSlowConsumer(t *testing.T) {
+	b := newTestLocalBackend(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	watchAdded := make(chan struct{})
+	firstNotify := make(chan struct{}, 1)
+	releaseFirstNotify := make(chan struct{})
+	terminalMessage := make(chan string, 1)
+	done := make(chan struct{})
+
+	go func() {
+		defer close(done)
+		b.WatchNotificationsAs(ctx, nil, 0, func() { close(watchAdded) }, func(n *ipn.Notify) bool {
+			if n.ErrMessage != nil {
+				terminalMessage <- *n.ErrMessage
+				return true
+			}
+			select {
+			case firstNotify <- struct{}{}:
+				<-releaseFirstNotify
+			default:
+			}
+			return true
+		})
+	}()
+	<-watchAdded
+
+	state := ipn.Running
+	b.send(ipn.Notify{State: &state})
+	select {
+	case <-firstNotify:
+	case <-time.After(10 * time.Second):
+		t.Fatal("timeout waiting for first notification")
+	}
+
+	// The watcher's callback is blocked on the first notification. Fill the
+	// 128-slot queue, then send one more notification to force overflow.
+	for range 129 {
+		b.send(ipn.Notify{State: &state})
+	}
+
+	close(releaseFirstNotify)
+
+	select {
+	case got := <-terminalMessage:
+		if got != watchIPNBusFellBehindMessage {
+			t.Fatalf("terminal message = %q; want %q", got, watchIPNBusFellBehindMessage)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("timeout waiting for terminal notification")
+	}
+
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("timeout waiting for watcher to close")
+	}
+}
+
 // TestNotifyForSessionPeerVisibility verifies the per-session masking
 // logic in [LocalBackend.notifyForSessionLocked] for the
 // NotifyPeerChanges / NotifyPeerPatches flag pair: