diff --git a/control/controlclient/map.go b/control/controlclient/map.go index 34b5ecc55..9af3a75bd 100644 --- a/control/controlclient/map.go +++ b/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 diff --git a/control/controlclient/map_test.go b/control/controlclient/map_test.go index ebebc2e66..1c4dc6d78 100644 --- a/control/controlclient/map_test.go +++ b/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) {