From 15726213f917ea1cc58c638542928369249deee2 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Sun, 27 Sep 2026 20:30:17 -0400 Subject: [PATCH] feat(ws): opt-in includeRepeats streams later hearings over new paths Duplicate observations are dropped before the hub today, so a live map only sees each packet's first hearing per observer. Clients that configure includeRepeats also get later hearings whose path is new; nothing is stored, repeats skip the count query, and they are dropped first when the broadcast channel is half full. --- internal/hub/hub.go | 62 ++++++++++++- internal/hub/hub_test.go | 109 ++++++++++++++++++++++ internal/hub/sent.go | 59 ++++++++++++ internal/ingest/ingest.go | 7 +- internal/ingest/observer_key_test.go | 2 +- internal/ingest/packet.go | 30 ++++-- internal/ingest/repeats_test.go | 132 +++++++++++++++++++++++++++ internal/ws/handler.go | 19 ++-- internal/ws/handler_test.go | 65 +++++++++++++ 9 files changed, 465 insertions(+), 20 deletions(-) create mode 100644 internal/hub/sent.go create mode 100644 internal/ingest/repeats_test.go diff --git a/internal/hub/hub.go b/internal/hub/hub.go index d4086a1..09c2207 100644 --- a/internal/hub/hub.go +++ b/internal/hub/hub.go @@ -21,6 +21,7 @@ import ( "log/slog" "slices" "sync/atomic" + "time" ) // EventType identifies the kind of server-push event. These match the @@ -54,6 +55,8 @@ type Event struct { IATA string PayloadType uint8 ChannelHash string // hex string, non-empty only for channelMessage events + // Repeat marks a later hearing of a stored observation; only IncludeRepeats clients get it. + Repeat bool } // Scope mirrors the client-side subscribe message. All fields are optional: @@ -84,12 +87,15 @@ type Client struct { ResolvePath bool // IncludeObserverKey works like ResolvePath, for observerPublicKey. IncludeObserverKey bool + // IncludeRepeats works like ResolvePath, for Repeat events. + IncludeRepeats bool } // ClientOptions are the connection-wide settings a "configure" message sets. type ClientOptions struct { ResolvePath bool IncludeObserverKey bool + IncludeRepeats bool } // payloadFor falls back to a narrower variant if the event lacks one. @@ -153,6 +159,12 @@ type Hub struct { // observerKeyClients counts registered clients with IncludeObserverKey set. observerKeyClients atomic.Int64 + // repeatClients counts registered clients with IncludeRepeats set. + repeatClients atomic.Int64 + // repeatDrops counts repeats refused because the broadcast channel was busy. + repeatDrops atomic.Int64 + repeatDropLogAt atomic.Int64 // unix nanos of the last drop log line + sent *sentPaths } // subscribeMsg carries a client registration, a scope subscription, or a @@ -185,6 +197,7 @@ func New() *Hub { unsubscribe: make(chan unsubscribeMsg, 64), remove: make(chan *Client, 64), broadcast: make(chan Event, 512), + sent: newSentPaths(sentPathsTTL, sentPathsMax), } } @@ -226,6 +239,17 @@ func (h *Hub) ObserverKeyWanted() bool { return h.observerKeyClients.Load() > 0 } +// RepeatsWanted lets ingest skip repeat work when nobody wants it. +func (h *Hub) RepeatsWanted() bool { + return h.repeatClients.Load() > 0 +} + +// MarkSent records a hearing's path and reports whether it was not already sent recently, +// so broker copies and same-path duplicates go out once. +func (h *Hub) MarkSent(packetHash, observerID, path []byte) bool { + return h.sent.mark(packetHash, observerID, path, time.Now()) +} + // Remove deregisters a client and closes its Send channel. // Safe to call from any goroutine (e.g. the WS handler's defer). func (h *Hub) Remove(c *Client) { @@ -241,6 +265,31 @@ func (h *Hub) Broadcast(e Event) { } } +// BroadcastRepeat enqueues a repeat event, dropping it once the broadcast channel is half +// full so repeats never crowd out first hearings. +func (h *Hub) BroadcastRepeat(e Event) { + e.Repeat = true + if len(h.broadcast) >= cap(h.broadcast)/2 { + h.dropRepeat() + return + } + select { + case h.broadcast <- e: + default: + h.dropRepeat() + } +} + +// dropRepeat counts a dropped repeat and logs the running total at most once a minute. +func (h *Hub) dropRepeat() { + total := h.repeatDrops.Add(1) + now := time.Now().UnixNano() + last := h.repeatDropLogAt.Load() + if now-last >= int64(time.Minute) && h.repeatDropLogAt.CompareAndSwap(last, now) { + slog.Warn("hub: broadcast channel busy, dropping repeat events", "component", "hub", "dropped_total", total) + } +} + // Run is the hub's single-goroutine event loop. Call it in a dedicated // goroutine: go hub.Run(). // @@ -270,8 +319,16 @@ func (h *Hub) Run() { h.observerKeyClients.Add(-1) } } + if msg.client.IncludeRepeats != msg.options.IncludeRepeats { + if msg.options.IncludeRepeats { + h.repeatClients.Add(1) + } else { + h.repeatClients.Add(-1) + } + } msg.client.ResolvePath = msg.options.ResolvePath msg.client.IncludeObserverKey = msg.options.IncludeObserverKey + msg.client.IncludeRepeats = msg.options.IncludeRepeats } default: // AddScope path — client must already be registered. @@ -291,13 +348,16 @@ func (h *Hub) Run() { if c.IncludeObserverKey { h.observerKeyClients.Add(-1) } + if c.IncludeRepeats { + h.repeatClients.Add(-1) + } close(c.Send) close(c.laggedCH) } case evt := <-h.broadcast: for c := range clients { - if !c.matches(evt) { + if (evt.Repeat && !c.IncludeRepeats) || !c.matches(evt) { continue } outEvt := evt diff --git a/internal/hub/hub_test.go b/internal/hub/hub_test.go index 3a99223..94be968 100644 --- a/internal/hub/hub_test.go +++ b/internal/hub/hub_test.go @@ -434,3 +434,112 @@ func TestHub_ObserverKeyWanted_TracksOptIns(t *testing.T) { h.Remove(b) waitFor(false) } + +func TestHub_Repeat_OnlyReachesOptedInClients(t *testing.T) { + h := runHub(t) + plain, opted := h.NewClient(), h.NewClient() + h.AddScope(plain, "all", Scope{}) + h.AddScope(opted, "all", Scope{}) + h.Configure(opted, ClientOptions{IncludeRepeats: true}) + time.Sleep(10 * time.Millisecond) + + read := func(c *Client) (Event, bool) { + select { + case evt := <-c.Send: + return evt, true + case <-time.After(50 * time.Millisecond): + return Event{}, false + } + } + h.BroadcastRepeat(Event{Type: EventPacketObservation}) + if evt, ok := read(opted); !ok || !evt.Repeat { + t.Fatalf("opted-in client: got %+v, %t", evt, ok) + } + if evt, ok := read(plain); ok { + t.Fatalf("plain client got repeat %+v", evt) + } + h.Broadcast(Event{Type: EventPacketObservation}) + for _, c := range []*Client{plain, opted} { + if evt, ok := read(c); !ok || evt.Repeat { + t.Fatalf("normal event: got %+v, %t", evt, ok) + } + } +} + +func TestHub_RepeatsWanted_TracksOptIns(t *testing.T) { + h := runHub(t) + waitFor := func(want bool) { + t.Helper() + deadline := time.Now().Add(time.Second) + for h.RepeatsWanted() != want { + if time.Now().After(deadline) { + t.Fatalf("RepeatsWanted = %t, want %t", !want, want) + } + time.Sleep(time.Millisecond) + } + } + a, b := h.NewClient(), h.NewClient() + if h.RepeatsWanted() { + t.Fatal("no client opted in yet") + } + h.Configure(a, ClientOptions{IncludeRepeats: true}) + h.Configure(a, ClientOptions{IncludeRepeats: true}) // repeat must not double count + h.Configure(b, ClientOptions{IncludeRepeats: true}) + waitFor(true) + h.Configure(a, ClientOptions{ResolvePath: true}) + h.Remove(b) + waitFor(false) +} + +func TestHub_BroadcastRepeat_DropsFirstWhenBusy(t *testing.T) { + h := New() // not running, so the broadcast channel only fills + h.BroadcastRepeat(Event{Type: EventPacketObservation}) + if len(h.broadcast) != 1 || h.repeatDrops.Load() != 0 { + t.Fatalf("repeat on an idle hub: queued %d, dropped %d", len(h.broadcast), h.repeatDrops.Load()) + } + for len(h.broadcast) < cap(h.broadcast)/2 { + h.Broadcast(Event{Type: EventPacketObservation}) + } + queued := len(h.broadcast) + h.BroadcastRepeat(Event{Type: EventPacketObservation}) + h.BroadcastRepeat(Event{Type: EventPacketObservation}) + if len(h.broadcast) != queued || h.repeatDrops.Load() != 2 { + t.Fatalf("repeats past half full: queued %d (want %d), dropped %d", len(h.broadcast), queued, h.repeatDrops.Load()) + } + h.Broadcast(Event{Type: EventPacketObservation}) + if len(h.broadcast) != queued+1 { + t.Fatal("normal event not enqueued past half full") + } +} + +func TestSentPaths(t *testing.T) { + start := time.Now() + s := newSentPaths(time.Minute, 3) + hash, observer := []byte{1, 2}, []byte{3} + if !s.mark(hash, observer, []byte{0xaa}, start) { + t.Fatal("first hearing refused") + } + if s.mark(hash, observer, []byte{0xaa}, start.Add(time.Second)) { + t.Fatal("exact copy within the window allowed") + } + if !s.mark(hash, observer, []byte{0xaa, 0xbb}, start.Add(time.Second)) { + t.Fatal("new path refused") + } + if !s.mark(hash, []byte{4}, []byte{0xaa}, start.Add(time.Second)) { + t.Fatal("other observer refused") + } + if !s.mark(hash, observer, []byte{0xaa}, start.Add(time.Minute)) { + t.Fatal("copy after the window refused") + } + + s = newSentPaths(time.Hour, 3) + for i := range 4 { + s.mark(hash, observer, []byte{byte(i)}, start) + } + if len(s.at) != 3 || len(s.order) != 3 { + t.Fatalf("size bound: %d keys, %d queued", len(s.at), len(s.order)) + } + if !s.mark(hash, observer, []byte{0}, start) { + t.Fatal("oldest key not evicted at the size bound") + } +} diff --git a/internal/hub/sent.go b/internal/hub/sent.go new file mode 100644 index 0000000..f1e0411 --- /dev/null +++ b/internal/hub/sent.go @@ -0,0 +1,59 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package hub + +import ( + "hash/maphash" + "sync" + "time" +) + +const ( + sentPathsTTL = 2 * time.Minute + sentPathsMax = 200_000 +) + +// sentPaths remembers recently sent (packet, observer, path) hearings. It is shared by all +// broker workers; keys are hashed so memory stays small at the size bound. +type sentPaths struct { + mu sync.Mutex + seed maphash.Seed + ttl time.Duration + max int + at map[uint64]time.Time + order []sentPath // oldest first +} + +type sentPath struct { + key uint64 + at time.Time +} + +func newSentPaths(ttl time.Duration, max int) *sentPaths { + return &sentPaths{seed: maphash.MakeSeed(), ttl: ttl, max: max, at: make(map[uint64]time.Time)} +} + +// mark records the hearing and reports whether it was not already recorded within the TTL. +func (s *sentPaths) mark(packetHash, observerID, path []byte, now time.Time) bool { + var h maphash.Hash + h.SetSeed(s.seed) + for _, part := range [][]byte{packetHash, observerID, path} { + h.WriteByte(byte(len(part))) + h.Write(part) + } + key := h.Sum64() + + s.mu.Lock() + defer s.mu.Unlock() + for len(s.order) > 0 && (now.Sub(s.order[0].at) >= s.ttl || len(s.order) >= s.max) { + delete(s.at, s.order[0].key) + s.order = s.order[1:] + } + if _, ok := s.at[key]; ok { + return false + } + s.at[key] = now + s.order = append(s.order, sentPath{key, now}) + return true +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 4961344..f19e403 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -385,7 +385,8 @@ func (w *Worker) broadcast(eventType hub.EventType, iata string, payloadType uin // passed in rather than computed here because the caller already has the // path-hash resolution results in hand from other per-packet work (known // route detection, capability detection) — this adds no extra DB calls. -func (w *Worker) broadcastPacketObservation(iata string, payloadType uint8, evt packetObservationEvent, resolvedPath []api.ResolvedHop, observerKey string) { +// Repeats go through the hub's drop-first path. +func (w *Worker) broadcastPacketObservation(iata string, payloadType uint8, evt packetObservationEvent, resolvedPath []api.ResolvedHop, observerKey string, repeat bool) { base, err := json.Marshal(evt) if err != nil { w.log.Error("failed to marshal packetObservation event", "error", err) @@ -414,6 +415,10 @@ func (w *Worker) broadcastPacketObservation(iata string, payloadType uint8, evt out.PayloadWithKey = nil } } + if repeat { + w.hub.BroadcastRepeat(out) + return + } w.hub.Broadcast(out) } diff --git a/internal/ingest/observer_key_test.go b/internal/ingest/observer_key_test.go index afd04b0..c202121 100644 --- a/internal/ingest/observer_key_test.go +++ b/internal/ingest/observer_key_test.go @@ -27,7 +27,7 @@ func TestBroadcastPacketObservation_ObserverKeyOptIn(t *testing.T) { broadcast := func() hub.Event { t.Helper() - w.broadcastPacketObservation("YVR", 4, packetObservationEvent{}, []api.ResolvedHop{{}}, key) + w.broadcastPacketObservation("YVR", 4, packetObservationEvent{}, []api.ResolvedHop{{}}, key, false) for { select { case evt := <-client.Send: diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index e34b55d..ad1bfcb 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -80,6 +80,8 @@ type packetObservationEvent struct { ObservationCount int64 `json:"observationCount"` Scope *string `json:"scope,omitempty"` Summary *string `json:"summary,omitempty"` // same advert name as REST list/backfill rows + // Only set on later hearings sent to includeRepeats clients; those carry observationCount 0. + IsRepeat bool `json:"isRepeat,omitempty"` } `json:"packet"` Observation struct { ObserverID string `json:"observerId"` @@ -880,15 +882,23 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) - if inserted { - w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID, matchedScope, pubkeyBytes, float32(parseNumber(envelope.SNR))) + // A duplicate is streamed, never stored, to includeRepeats clients when its path is new. + repeat := !inserted && w.hub.RepeatsWanted() && w.hub.MarkSent(packetHash[:], id[:], packet.Path) + if inserted || repeat { + if inserted { + w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID, matchedScope, pubkeyBytes, float32(parseNumber(envelope.SNR))) + if w.hub.RepeatsWanted() { + w.hub.MarkSent(packetHash[:], id[:], packet.Path) // so broker copies of it aren't repeats + } + } evt := packetObservationEvent{} evt.PacketHash = hex.EncodeToString(packetHash[:]) evt.Packet.PayloadType = packet.PayloadType() evt.Packet.PayloadTypeName = packet.PayloadTypeString() evt.Packet.RouteType = packet.RouteType() evt.Packet.RouteTypeName = api.RouteTypeName(int16(packet.RouteType())) - evt.Packet.IsFirstObservation = isNew + evt.Packet.IsFirstObservation = isNew && !repeat + evt.Packet.IsRepeat = repeat evt.Packet.Summary = summary evt.Observation.ObserverID = id.String() evt.Observation.ObserverName = observerName @@ -902,19 +912,21 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ evt.Observation.PathLength.HashSize = packet.PathHashSize() evt.Observation.PathLength.HopCount = packet.PathHashCount() evt.Observation.PropagationTimeMs = 0 // not yet calculated - count, err := w.db.GetPacketObservationCount(ctx, packetHash[:]) - if err != nil { - w.log.Error("failed to get observation count", "error", err) - count = 0 + if !repeat { + count, err := w.db.GetPacketObservationCount(ctx, packetHash[:]) + if err != nil { + w.log.Error("failed to get observation count", "error", err) + count = 0 + } + evt.Packet.ObservationCount = count } - evt.Packet.ObservationCount = count if matchedScope != nil { evt.Packet.Scope = matchedScope } resolvedPath := api.BuildResolvedPath(hashes, resolved) evt.Observation.ResolvedSource = resolvedSource evt.Observation.ResolvedDestination = resolvedDestination - w.broadcastPacketObservation(iata, packet.PayloadType(), evt, resolvedPath, hex.EncodeToString(pubkeyBytes)) + w.broadcastPacketObservation(iata, packet.PayloadType(), evt, resolvedPath, hex.EncodeToString(pubkeyBytes), repeat) } } diff --git a/internal/ingest/repeats_test.go b/internal/ingest/repeats_test.go new file mode 100644 index 0000000..5766c09 --- /dev/null +++ b/internal/ingest/repeats_test.go @@ -0,0 +1,132 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package ingest + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/hub" +) + +type countQueryDB struct { + *stubDB + countQueries int +} + +func (s *countQueryDB) GetPacketObservationCount(context.Context, []byte) (int64, error) { + s.countQueries++ + return 1, nil +} + +type repeatHarness struct { + t *testing.T + ctx context.Context + w *Worker + db *countQueryDB + client *hub.Client +} + +func newRepeatHarness(t *testing.T, includeRepeats bool) *repeatHarness { + w, base := newTestWorker() + db := &countQueryDB{stubDB: base} + w.db = db + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + t.Cleanup(cancel) + go w.hub.Run() + client := w.hub.NewClient() + t.Cleanup(func() { w.hub.Remove(client) }) + w.hub.AddScope(client, "all", hub.Scope{Events: []hub.EventType{hub.EventPacketObservation, hub.EventObserverStatus}}) + if includeRepeats { + w.hub.Configure(client, hub.ClientOptions{IncludeRepeats: true}) + for !w.hub.RepeatsWanted() { + time.Sleep(time.Millisecond) + } + } + waitForSummarySubscriber(t, ctx, w.hub, client) + return &repeatHarness{t: t, ctx: ctx, w: w, db: db, client: client} +} + +// hear feeds one hearing of the same group text over path and returns the packetObservation +// payloads it produced. +func (r *repeatHarness) hear(inserted bool, path ...byte) []json.RawMessage { + r.t.Helper() + packet := buildGrpTxtPacket(r.t, 0x1a, make([]byte, 16)) + packet.Path = path + packet.PathLength = byte(len(path)) + r.db.observationInserted = inserted + r.w.handlePacket(r.ctx, "YOW", "0102", packetEnvelope(r.t, packet)) + + // Hub delivery is FIFO, so everything handlePacket sent arrives before this marker. + r.w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus}) + var got []json.RawMessage + for { + select { + case evt := <-r.client.Send: + if evt.Type == hub.EventObserverStatus { + return got + } + got = append(got, evt.Payload) + case <-r.ctx.Done(): + r.t.Fatal("marker event not delivered") + } + } +} + +func packetFields(t *testing.T, raw json.RawMessage) map[string]json.RawMessage { + t.Helper() + var evt struct { + Packet map[string]json.RawMessage `json:"packet"` + } + if err := json.Unmarshal(raw, &evt); err != nil { + t.Fatal(err) + } + return evt.Packet +} + +func TestHandlePacket_Repeats(t *testing.T) { + r := newRepeatHarness(t, true) + + first := r.hear(true, 0x11) + if len(first) != 1 || strings.Contains(string(first[0]), "isRepeat") { + t.Fatalf("first hearing: %s", first) + } + if r.db.countQueries != 1 { + t.Fatalf("first hearing ran %d count queries", r.db.countQueries) + } + if got := r.hear(false, 0x11); len(got) != 0 { + t.Fatalf("broker copy of the first hearing sent: %s", got) + } + + repeat := r.hear(false, 0x22, 0x11) + if len(repeat) != 1 { + t.Fatalf("new path sent %d events", len(repeat)) + } + fields := packetFields(t, repeat[0]) + if string(fields["isRepeat"]) != "true" || string(fields["isFirstObservation"]) != "false" || string(fields["observationCount"]) != "0" { + t.Fatalf("repeat packet fields: %v", fields) + } + if r.db.countQueries != 1 { + t.Fatalf("repeat ran a count query (%d total)", r.db.countQueries) + } + if got := r.hear(false, 0x22, 0x11); len(got) != 0 { + t.Fatalf("same path sent twice: %s", got) + } +} + +func TestHandlePacket_RepeatsOffByDefault(t *testing.T) { + r := newRepeatHarness(t, false) + if got := r.hear(true, 0x11); len(got) != 1 || strings.Contains(string(got[0]), "isRepeat") { + t.Fatalf("first hearing: %s", got) + } + if got := r.hear(false, 0x22, 0x11); len(got) != 0 { + t.Fatalf("repeat sent with nobody opted in: %s", got) + } + if r.db.countQueries != 1 { + t.Fatalf("duplicate ran a count query (%d total)", r.db.countQueries) + } +} diff --git a/internal/ws/handler.go b/internal/ws/handler.go index c51a4b4..dd9c545 100644 --- a/internal/ws/handler.go +++ b/internal/ws/handler.go @@ -10,15 +10,17 @@ // Client → Server: // subscribe { v, type, id, scope } → server replies subscribed { v, type, id, subscriptionId } // unsubscribe { v, type, id, subscriptionId } -// configure { v, type, id, resolvePath, includeObserverKey } -// → server replies configured { v, type, id, resolvePath, includeObserverKey } +// configure { v, type, id, resolvePath, includeObserverKey, includeRepeats } +// → server replies configured { v, type, id, resolvePath, includeObserverKey, includeRepeats } // ping { v, type, id } → server replies pong { v, type, id } // // configure's flags are connection-wide settings, not per-subscription. // resolvePath adds per-hop resolvedPath data to packetObservation events, -// includeObserverKey adds observation.observerPublicKey. Each configure -// sets both to exactly the values sent, so an omitted flag turns off. -// Both default false. +// includeObserverKey adds observation.observerPublicKey, and includeRepeats +// also streams later hearings of an already-stored observation over a new +// path, as packetObservation events with packet.isRepeat true and +// observationCount 0. Each configure sets all three to exactly the values +// sent, so an omitted flag turns off. All default false. // // Server → Client events (unsolicited): // packetObservation, observerStatus, nodeUpdate, channelMessage @@ -180,6 +182,7 @@ type clientMessage struct { // Only read for "configure" messages. See package doc. ResolvePath bool `json:"resolvePath,omitempty"` IncludeObserverKey bool `json:"includeObserverKey,omitempty"` + IncludeRepeats bool `json:"includeRepeats,omitempty"` } // subscribeScope mirrors the scope object in the subscribe message. @@ -264,13 +267,13 @@ func handleClientMessage(ctx context.Context, client *hub.Client, reader api.Rea } case "configure": - h.Configure(client, hub.ClientOptions{ResolvePath: msg.ResolvePath, IncludeObserverKey: msg.IncludeObserverKey}) + h.Configure(client, hub.ClientOptions{ResolvePath: msg.ResolvePath, IncludeObserverKey: msg.IncludeObserverKey, IncludeRepeats: msg.IncludeRepeats}) reply, _ := json.Marshal(map[string]any{ "v": 1, "type": "configured", "id": msg.ID, - "resolvePath": msg.ResolvePath, "includeObserverKey": msg.IncludeObserverKey, + "resolvePath": msg.ResolvePath, "includeObserverKey": msg.IncludeObserverKey, "includeRepeats": msg.IncludeRepeats, }) if slog.Default().Enabled(context.Background(), slog.LevelDebug) { - slog.Debug(fmt.Sprintf("ws[%s]: configured resolvePath=%t includeObserverKey=%t", connID, msg.ResolvePath, msg.IncludeObserverKey), "component", "ws") + slog.Debug(fmt.Sprintf("ws[%s]: configured resolvePath=%t includeObserverKey=%t includeRepeats=%t", connID, msg.ResolvePath, msg.IncludeObserverKey, msg.IncludeRepeats), "component", "ws") } if err := conn.Write(ctx, websocket.MessageText, reply); err != nil { slog.Warn(fmt.Sprintf("ws[%s]: failed to send configured reply", connID), "component", "ws", "error", err) diff --git a/internal/ws/handler_test.go b/internal/ws/handler_test.go index 176a3e6..3e9c590 100644 --- a/internal/ws/handler_test.go +++ b/internal/ws/handler_test.go @@ -132,3 +132,68 @@ func TestConfigureIncludeObserverKey(t *testing.T) { t.Errorf("opted-in client got %s", got) } } + +func TestConfigureIncludeRepeats(t *testing.T) { + h := hub.New() + go h.Run() + server := httptest.NewServer(Handler(h, nil, 5, 100, nil)) + t.Cleanup(server.Close) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + conn, _, err := websocket.Dial(ctx, "ws"+strings.TrimPrefix(server.URL, "http"), nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.CloseNow() }) + read := func() map[string]json.RawMessage { + t.Helper() + _, data, err := conn.Read(ctx) + if err != nil { + t.Fatal(err) + } + var msg map[string]json.RawMessage + if err := json.Unmarshal(data, &msg); err != nil { + t.Fatal(err) + } + return msg + } + send := func(msg string) { + t.Helper() + if err := conn.Write(ctx, websocket.MessageText, []byte(msg)); err != nil { + t.Fatal(err) + } + } + configure := func(msg, want string) { + t.Helper() + send(msg) + if reply := read(); string(reply["type"]) != `"configured"` || string(reply["includeRepeats"]) != want { + t.Fatalf("configured reply: %v", reply) + } + time.Sleep(20 * time.Millisecond) // let the hub apply it + } + // Each repeat is followed by a normal marker, so a skipped repeat shows up as the marker. + next := func() string { + t.Helper() + h.BroadcastRepeat(hub.Event{Type: hub.EventPacketObservation, Payload: json.RawMessage(`"repeat"`)}) + h.Broadcast(hub.Event{Type: hub.EventPacketObservation, Payload: json.RawMessage(`"marker"`)}) + got := string(read()["data"]) + if got == `"repeat"` { + read() // marker + } + return got + } + + read() // hello + configure(`{"v":1,"type":"configure","id":"c","includeRepeats":true}`, "true") + send(`{"v":1,"type":"subscribe","id":"s","scope":{}}`) + read() // subscribed + time.Sleep(20 * time.Millisecond) + if got := next(); got != `"repeat"` { + t.Fatalf("opted-in client got %s", got) + } + configure(`{"v":1,"type":"configure","id":"c","resolvePath":true}`, "false") + if got := next(); got != `"marker"` { + t.Fatalf("client still received a repeat after opting out: %s", got) + } +}