diff --git a/ipn/backend.go b/ipn/backend.go index bc8320f89..f6d229a9a 100644 --- a/ipn/backend.go +++ b/ipn/backend.go @@ -158,6 +158,25 @@ const ( NotifyPeerPatches NotifyWatchOpt = 1 << 15 ) +// NotifyRateLimitIncompatibleBits is the set of new-style IPN bus +// subscription bits that cannot be combined with [NotifyRateLimit]. +// +// Those bits describe stateful delta streams. Randomly delaying or merging +// messages in those streams would break the consumer's ability to maintain a +// coherent local view. +const NotifyRateLimitIncompatibleBits = NotifyPeerChanges | NotifyNoNetMap | NotifyInitialStatus | NotifyPeerPatches + +// ValidateNotifyWatchOpt reports whether mask is a valid WatchIPNBus +// subscription mask. +func ValidateNotifyWatchOpt(mask NotifyWatchOpt) error { + if mask&NotifyRateLimit != 0 { + if bad := mask & NotifyRateLimitIncompatibleBits; bad != 0 { + return fmt.Errorf("NotifyRateLimit is incompatible with new-style IPN bus subscription bits %v", bad) + } + } + return nil +} + // Notify is a communication from a backend (e.g. tailscaled) to a frontend // (cmd/tailscale, iOS, macOS, Win Tasktray). // In any given notification, any or all of these may be nil, meaning diff --git a/ipn/backend_test.go b/ipn/backend_test.go index bfb6d6bc4..38159c821 100644 --- a/ipn/backend_test.go +++ b/ipn/backend_test.go @@ -40,3 +40,49 @@ func TestNotifyString(t *testing.T) { }) } } + +func TestValidateNotifyWatchOpt(t *testing.T) { + tests := []struct { + name string + mask NotifyWatchOpt + wantErr bool + }{ + { + name: "legacy-rate-limit-only", + mask: NotifyRateLimit, + }, + { + name: "peer-changes-without-rate-limit", + mask: NotifyPeerChanges | NotifyPeerPatches | NotifyNoNetMap | NotifyInitialStatus, + }, + { + name: "rate-limit-with-peer-changes", + mask: NotifyRateLimit | NotifyPeerChanges, + wantErr: true, + }, + { + name: "rate-limit-with-peer-patches", + mask: NotifyRateLimit | NotifyPeerPatches, + wantErr: true, + }, + { + name: "rate-limit-with-no-netmap", + mask: NotifyRateLimit | NotifyNoNetMap, + wantErr: true, + }, + { + name: "rate-limit-with-initial-status", + mask: NotifyRateLimit | NotifyInitialStatus, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := ValidateNotifyWatchOpt(tt.mask) + if gotErr := err != nil; gotErr != tt.wantErr { + t.Fatalf("ValidateNotifyWatchOpt(%v) error = %v; wantErr %v", tt.mask, err, tt.wantErr) + } + }) + } +} diff --git a/ipn/ipnlocal/bus.go b/ipn/ipnlocal/bus.go index 60e7af1bc..fbbfd7dda 100644 --- a/ipn/ipnlocal/bus.go +++ b/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 || diff --git a/ipn/ipnlocal/bus_test.go b/ipn/ipnlocal/bus_test.go index 68551b695..14aaae541 100644 --- a/ipn/ipnlocal/bus_test.go +++ b/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") diff --git a/ipn/ipnlocal/local.go b/ipn/ipnlocal/local.go index 20a22e29e..773656e6b 100644 --- a/ipn/ipnlocal/local.go +++ b/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 } } } diff --git a/ipn/ipnlocal/local_test.go b/ipn/ipnlocal/local_test.go index 237ed6d4e..8aaa20691 100644 --- a/ipn/ipnlocal/local_test.go +++ b/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: diff --git a/ipn/localapi/localapi.go b/ipn/localapi/localapi.go index 1acd98bb4..8fb83a594 100644 --- a/ipn/localapi/localapi.go +++ b/ipn/localapi/localapi.go @@ -902,6 +902,10 @@ func (h *Handler) serveWatchIPNBus(w http.ResponseWriter, r *http.Request) { } mask = ipn.NotifyWatchOpt(v) } + if err := ipn.ValidateNotifyWatchOpt(mask); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } // NotifyInitialNetMap is permitted alongside NotifyPeerChanges / // NotifyPeerPatches for backwards compatibility with clients that // set both (e.g. the Apple client). On platforms where diff --git a/ipn/localapi/localapi_test.go b/ipn/localapi/localapi_test.go index 1c1873a1d..cd51ab620 100644 --- a/ipn/localapi/localapi_test.go +++ b/ipn/localapi/localapi_test.go @@ -634,6 +634,7 @@ func TestServeWatchIPNBus(t *testing.T) { tests := []struct { desc string permitRead, permitWrite bool + mask ipn.NotifyWatchOpt wantStatus int }{ { @@ -654,6 +655,12 @@ func TestServeWatchIPNBus(t *testing.T) { permitWrite: true, wantStatus: http.StatusOK, }, + { + desc: "invalid-rate-limit-mask", + permitRead: true, + mask: ipn.NotifyRateLimit | ipn.NotifyPeerChanges, + wantStatus: http.StatusBadRequest, + }, } for _, tt := range tests { @@ -668,7 +675,11 @@ func TestServeWatchIPNBus(t *testing.T) { c := s.Client() ctx, cancel := context.WithCancel(context.Background()) - req, err := http.NewRequestWithContext(ctx, "GET", fmt.Sprintf("%s/localapi/v0/watch-ipn-bus?mask=%d", s.URL, ipn.NotifyInitialState), nil) + mask := tt.mask + if mask == 0 { + mask = ipn.NotifyInitialState + } + req, err := http.NewRequestWithContext(ctx, "GET", fmt.Sprintf("%s/localapi/v0/watch-ipn-bus?mask=%d", s.URL, mask), nil) if err != nil { t.Fatal(err) }