diff --git a/internal/hub/hub.go b/internal/hub/hub.go index 3acc5b2..1f8866d 100644 --- a/internal/hub/hub.go +++ b/internal/hub/hub.go @@ -109,7 +109,7 @@ func scopeMatches(s Scope, e Event) bool { if len(s.PayloadTypes) > 0 && !slices.Contains(s.PayloadTypes, e.PayloadType) { return false } - if len(s.ChannelHashes) > 0 && !slices.Contains(s.ChannelHashes, e.ChannelHash) { + if e.Type == EventChannelMessage && len(s.ChannelHashes) > 0 && !slices.Contains(s.ChannelHashes, e.ChannelHash) { return false } return true @@ -119,15 +119,26 @@ func scopeMatches(s Scope, e Event) bool { type Hub struct { subscribe chan subscribeMsg unsubscribe chan unsubscribeMsg - configure chan configureMsg remove chan *Client broadcast chan Event } +// subscribeMsg carries a client registration, a scope subscription, or a +// configure (resolvePath toggle) request — all three go through this single +// channel, not separate ones, specifically so that Go's same-channel FIFO +// guarantee orders them relative to NewClient's registration message. A +// separate "configure" channel raced against registration: select() has no +// cross-channel ordering guarantee, so a configure sent immediately after +// NewClient could be (and empirically was, ~50% of the time) processed by +// Run() before the registration message, silently no-oping since the client +// wasn't in the clients map yet. type subscribeMsg struct { client *Client scope Scope subscriptionID string + + isConfigure bool + resolvePath bool } type unsubscribeMsg struct { @@ -148,7 +159,6 @@ func New() *Hub { return &Hub{ subscribe: make(chan subscribeMsg, 64), unsubscribe: make(chan unsubscribeMsg, 64), - configure: make(chan configureMsg, 64), remove: make(chan *Client, 64), broadcast: make(chan Event, 512), } @@ -187,7 +197,7 @@ func (h *Hub) RemoveScope(c *Client, id string) { // the connection's lifetime — takes effect on the next broadcast after the // hub processes it. func (h *Hub) SetResolvePath(c *Client, enabled bool) { - h.configure <- configureMsg{client: c, resolvePath: enabled} + h.subscribe <- subscribeMsg{client: c, isConfigure: true, resolvePath: enabled} } // Remove deregisters a client and closes its Send channel. @@ -220,10 +230,16 @@ func (h *Hub) Run() { select { case msg := <-h.subscribe: - if msg.subscriptionID == "" { + switch { + case msg.subscriptionID == "" && !msg.isConfigure: // Registration with no scope yet (NewClient path). clients[msg.client] = struct{}{} - } else { + case msg.isConfigure: + // SetResolvePath path — client must already be registered. + if _, ok := clients[msg.client]; ok { + msg.client.ResolvePath = msg.resolvePath + } + default: // AddScope path — client must already be registered. if _, ok := clients[msg.client]; ok { msg.client.subscriptions[msg.subscriptionID] = msg.scope @@ -235,11 +251,6 @@ func (h *Hub) Run() { delete(msg.client.subscriptions, msg.subscriptionID) } - case msg := <-h.configure: - if _, ok := clients[msg.client]; ok { - msg.client.ResolvePath = msg.resolvePath - } - case c := <-h.remove: if _, ok := clients[c]; ok { delete(clients, c) diff --git a/internal/hub/hub_test.go b/internal/hub/hub_test.go index d2bee3f..5ced37b 100644 --- a/internal/hub/hub_test.go +++ b/internal/hub/hub_test.go @@ -61,6 +61,24 @@ func TestScopeMatches_ChannelHashFilter(t *testing.T) { } } +func TestScopeMatches_ChannelHashesOnlyAppliesToChannelMessage(t *testing.T) { + s := Scope{ + Events: []EventType{EventPacketObservation, EventChannelMessage}, + ChannelHashes: []string{"11"}, + } + // packetObservation has no channel hash — must still match despite the + // channelHashes filter, since that filter only applies to channelMessage. + if !scopeMatches(s, Event{Type: EventPacketObservation, ChannelHash: ""}) { + t.Error("expected packetObservation to match even with channelHashes set") + } + if !scopeMatches(s, Event{Type: EventChannelMessage, ChannelHash: "11"}) { + t.Error("expected channelMessage with matching hash to match") + } + if scopeMatches(s, Event{Type: EventChannelMessage, ChannelHash: "ff"}) { + t.Error("expected channelMessage with non-matching hash to be filtered") + } +} + func TestScopeMatches_AllFiltersPass(t *testing.T) { s := Scope{ Events: []EventType{EventPacketObservation},