control/controlclient/map.go

@@ -207,6 +207,7 @@ func (ms *mapSession) updateDiscoForNode(id tailcfg.NodeID, key key.NodePublic,
 		ms.processQueue.Wait()
 		return ErrChangeQueueClosed
 	}
+	defer ms.cqmu.Unlock()
 
 	resp := responseWithSource{
 		response: &tailcfg.MapResponse{
@@ -220,12 +221,24 @@ func (ms *mapSession) updateDiscoForNode(id tailcfg.NodeID, key key.NodePublic,
 		},
 		viaTSMP: true,
 	}
-	ms.changeQueue <- resp
-	ms.cqmu.Unlock()
-	return nil
+	return ms.addRespToQueue(resp)
 }
 
+// HandleNonKeepAliveMapResponse handles a non-KeepAlive MapResponse (full or
+// incremental).
+//
+// All fields that are valid on a KeepAlive MapResponse have already been
+// handled.
+//
+// Debug messages are handled first, followed by pushing the response onto a
+// queue for new updates handled sequentially.
 func (ms *mapSession) HandleNonKeepAliveMapResponse(ctx context.Context, resp *tailcfg.MapResponse) error {
+	if debug := resp.Debug; debug != nil {
+		if err := ms.onDebug(ctx, debug); err != nil {
+			return err
+		}
+	}
+
 	ms.cqmu.Lock()
 
 	if ms.changeQueueClosed {
@@ -234,14 +247,23 @@ func (ms *mapSession) HandleNonKeepAliveMapResponse(ctx context.Context, resp *t
 		return ErrChangeQueueClosed
 	}
 
+	defer ms.cqmu.Unlock()
+
 	change := responseWithSource{
 		response: resp,
 		viaTSMP:  false,
 	}
 
-	ms.changeQueue <- change
-	ms.cqmu.Unlock()
-	return nil
+	return ms.addRespToQueue(change)
+}
+
+func (ms *mapSession) addRespToQueue(resp responseWithSource) error {
+	select {
+	case ms.changeQueue <- resp:
+		return nil
+	case <-ms.sessionAliveCtx.Done():
+		return ErrChangeQueueClosed
+	}
 }
 
 // handleNonKeepAliveMapResponse handles a non-KeepAlive MapResponse (full or
@@ -253,12 +275,6 @@ func (ms *mapSession) HandleNonKeepAliveMapResponse(ctx context.Context, resp *t
 // TODO(bradfitz): make this handle all fields later. For now (2023-08-20) this
 // is [re]factoring progress enough.
 func (ms *mapSession) handleNonKeepAliveMapResponse(ctx context.Context, resp *tailcfg.MapResponse, viaTSMP bool) error {
-	if debug := resp.Debug; debug != nil {
-		if err := ms.onDebug(ctx, debug); err != nil {
-			return err
-		}
-	}
-
 	if DevKnob.StripEndpoints() {
 		for _, p := range resp.Peers {
 			p.Endpoints = nil

control/controlclient/map_test.go

@@ -865,6 +865,53 @@ func TestUpdateDiscoForNodeCallback(t *testing.T) {
 				nu.lastTSMPKey, nu.lastTSMPDisco)
 		}
 	})
+
+	t.Run("test_deadlock", func(t *testing.T) {
+		nu := &rememberLastNetmapUpdater{
+			done: make(chan any, 1),
+		}
+		ms := newTestMapSession(t, nu)
+		// Very barebones onDebug func that will let us exercise sleep command
+		// from control and potentially induce deadlocks.
+		ms.onDebug = func(ctx context.Context, d *tailcfg.Debug) error {
+			time.Sleep(time.Duration(d.SleepSeconds * float64(time.Second)))
+			return nil
+		}
+
+		oldKey := key.NewDisco()
+
+		// Insert existing node
+		node := tailcfg.Node{
+			ID:       1,
+			Key:      key.NewNode().Public(),
+			DiscoKey: oldKey.Public(),
+			Online:   new(false),
+			LastSeen: new(time.Unix(1, 0)),
+		}
+
+		if nm := ms.netmapForResponse(&tailcfg.MapResponse{
+			Peers: []*tailcfg.Node{&node},
+		}); len(nm.Peers) != 1 {
+			t.Fatalf("node not inserted")
+		}
+
+		sleep1 := &tailcfg.MapResponse{
+			Debug: &tailcfg.Debug{
+				SleepSeconds: 1.0,
+			},
+		}
+		ms.HandleNonKeepAliveMapResponse(t.Context(), sleep1)
+
+		// Resembles the disco key advert subscriber running in a separate context.
+		go func() {
+			newKey := key.NewDisco()
+			ms.updateDiscoForNode(node.ID, node.Key, newKey.Public(), time.Now(), false)
+		}()
+
+		ms.Close()
+
+		<-nu.done
+	})
 }
 
 func TestUpdateDiscoForNodeCallbackWithFullNetmap(t *testing.T) {