fix: pass OnSubscribe to HA MultiAgent (#9947)

coadler · web-flow · commit 4da1223a807e · 2023-09-29T13:37:17.000-05:00
Fixes #9929
diff --git a/enterprise/tailnet/coordinator.go b/enterprise/tailnet/coordinator.go
@@ -57,8 +57,9 @@ func (c *haCoordinator) ServeMultiAgent(id uuid.UUID) agpl.MultiAgentConn {
 		ID:                id,
 		AgentIsLegacyFunc: c.agentIsLegacy,
 		OnSubscribe:       c.clientSubscribeToAgent,
+		OnUnsubscribe:     c.clientUnsubscribeFromAgent,
 		OnNodeUpdate:      c.clientNodeUpdate,
-		OnRemove:          func(enq agpl.Queue) { c.clientDisconnected(enq.UniqueID()) },
+		OnRemove:          c.clientDisconnected,
 	}).Init()
 	c.addClient(id, m)
 	return m
@@ -101,6 +102,22 @@ func (c *haCoordinator) clientSubscribeToAgent(enq agpl.Queue, agentID uuid.UUID
 	return nil, nil
 }
 
+func (c *haCoordinator) clientUnsubscribeFromAgent(enq agpl.Queue, agentID uuid.UUID) error {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	connectionSockets, ok := c.agentToConnectionSockets[agentID]
+	if !ok {
+		return nil
+	}
+	delete(connectionSockets, enq.UniqueID())
+	if len(connectionSockets) == 0 {
+		delete(c.agentToConnectionSockets, agentID)
+	}
+
+	return nil
+}
+
 type haCoordinator struct {
 	id        uuid.UUID
 	log       slog.Logger
@@ -161,7 +178,7 @@ func (c *haCoordinator) ServeClient(conn net.Conn, id, agentID uuid.UUID) error
 	defer tc.Close()
 
 	c.addClient(id, tc)
-	defer c.clientDisconnected(id)
+	defer c.clientDisconnected(tc)
 
 	agentNode, err := c.clientSubscribeToAgent(tc, agentID)
 	if err != nil {
@@ -200,26 +217,24 @@ func (c *haCoordinator) initOrSetAgentConnectionSocketLocked(agentID uuid.UUID,
 	c.clientsToAgents[enq.UniqueID()][agentID] = c.agentSockets[agentID]
 }
 
-func (c *haCoordinator) clientDisconnected(id uuid.UUID) {
+func (c *haCoordinator) clientDisconnected(enq agpl.Queue) {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 
-	for agentID := range c.clientsToAgents[id] {
-		// Clean all traces of this connection from the map.
-		delete(c.nodes, id)
+	for agentID := range c.clientsToAgents[enq.UniqueID()] {
 		connectionSockets, ok := c.agentToConnectionSockets[agentID]
 		if !ok {
-			return
+			continue
 		}
-		delete(connectionSockets, id)
-		if len(connectionSockets) != 0 {
-			return
+		delete(connectionSockets, enq.UniqueID())
+		if len(connectionSockets) == 0 {
+			delete(c.agentToConnectionSockets, agentID)
 		}
-		delete(c.agentToConnectionSockets, agentID)
 	}
 
-	delete(c.clients, id)
-	delete(c.clientsToAgents, id)
+	delete(c.nodes, enq.UniqueID())
+	delete(c.clients, enq.UniqueID())
+	delete(c.clientsToAgents, enq.UniqueID())
 }
 
 func (c *haCoordinator) handleNextClientMessage(id uuid.UUID, decoder *json.Decoder) error {
