feat(ws): opt-in includeRepeats streams later hearings over new paths (#178)

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

@@ -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

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")
+	}
+}

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
+}

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)
 }
 

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:

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)
 	}
 }
 

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)
+	}
+}

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)

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)
+	}
+}