31711a6 fix(broker): read snapshots passively so vev ls never triggers a probe · vev

M internal/adapters/brokeripc/client.go +1 -0


@@ -503,6 +503,7 @@ func (c *client) Subscribe() (ports.BrokerSubscription, error) {
503	503		c.mu.Unlock()
504	504		if err := c.sendAsync(brokerwire.Subscribe{
505	505		Epoch: c.scope.Epoch, Connection: c.scope.Connection, Generation: generation,
	506	+	Passive: c.cfg.PassiveSubscribe,
506	507		}); err != nil {
507	508		// The generation was already consumed by the connection's subscription
508	509		// tracker, so the previous subscription's generation is no longer


@@ -503,6 +503,7 @@ func (c *client) Subscribe() (ports.BrokerSubscription, error) {
503	c.mu.Unlock()	503		c.mu.Unlock()
504	if err := c.sendAsync(brokerwire.Subscribe{	504		if err := c.sendAsync(brokerwire.Subscribe{
505	Epoch: c.scope.Epoch, Connection: c.scope.Connection, Generation: generation,	505		Epoch: c.scope.Epoch, Connection: c.scope.Connection, Generation: generation,
		506	+	Passive: c.cfg.PassiveSubscribe,
506	}); err != nil {	507		}); err != nil {
507	// The generation was already consumed by the connection's subscription	508		// The generation was already consumed by the connection's subscription
508	// tracker, so the previous subscription's generation is no longer	509		// tracker, so the previous subscription's generation is no longer

M internal/adapters/brokeripc/config.go +4 -0


@@ -46,6 +46,10 @@ type Config struct {
46	46		// MaxPendingOperations bounds mutating operations this side is waiting for
47	47		// on one connection. It never exceeds the brokerwire per-connection bound.
48	48		MaxPendingOperations int
	49	+	// PassiveSubscribe marks every snapshot subscription of a client
	50	+	// connection as a read-only reader: it receives publications but never
	51	+	// counts as observation demand, so a CLI read never triggers a probe.
	52	+	PassiveSubscribe bool
49	53		}
50	54
51	55		// withDefaults fills every unset bound with its package default.


@@ -46,6 +46,10 @@ type Config struct {
46	// MaxPendingOperations bounds mutating operations this side is waiting for	46		// MaxPendingOperations bounds mutating operations this side is waiting for
47	// on one connection. It never exceeds the brokerwire per-connection bound.	47		// on one connection. It never exceeds the brokerwire per-connection bound.
48	MaxPendingOperations int	48		MaxPendingOperations int
		49	+	// PassiveSubscribe marks every snapshot subscription of a client
		50	+	// connection as a read-only reader: it receives publications but never
		51	+	// counts as observation demand, so a CLI read never triggers a probe.
		52	+	PassiveSubscribe bool
49	}	53		}
50		54
51	// withDefaults fills every unset bound with its package default.	55		// withDefaults fills every unset bound with its package default.

M internal/adapters/brokeripc/helpers_test.go +27 -4


@@ -396,9 +396,11 @@ type fakeCore struct {
396	396		updated []updateHostPolicyCall
397	397		reconcile []string
398	398		openErr error
399		-	closed bool
400		-	done chan struct{}
401		-	once sync.Once
	399	+	// subscribes records each core subscription's reader kind (true = passive).
	400	+	subscribes []bool
	401	+	closed bool
	402	+	done chan struct{}
	403	+	once sync.Once
402	404
403	405		// Membership result hooks. Their zero values keep the deterministic
404	406		// defaults that mirror the real registry: a fresh generation-1 registration

@@ -431,7 +433,28 @@ func (c *fakeCore) Err() error { return nil }
431	433
432	434		func (c *fakeCore) Snapshot() ports.BrokerSnapshot { return c.hub.current() }
433	435
434		-	func (c *fakeCore) Subscribe() (ports.BrokerSubscription, error) { return c.hub.subscribe(), nil }
	436	+	func (c *fakeCore) Subscribe() (ports.BrokerSubscription, error) {
	437	+	return c.recordSubscribe(false), nil
	438	+	}
	439	+
	440	+	// SubscribePassive is the optional passive-reader capability the server uses
	441	+	// for a Subscribe flagged passive.
	442	+	func (c *fakeCore) SubscribePassive() (ports.BrokerSubscription, error) {
	443	+	return c.recordSubscribe(true), nil
	444	+	}
	445	+
	446	+	func (c fakeCore) recordSubscribe(passive bool) fakeSubscription {
	447	+	c.mu.Lock()
	448	+	c.subscribes = append(c.subscribes, passive)
	449	+	c.mu.Unlock()
	450	+	return c.hub.subscribe()
	451	+	}
	452	+
	453	+	func (c *fakeCore) subscribeKinds() []bool {
	454	+	c.mu.Lock()
	455	+	defer c.mu.Unlock()
	456	+	return append([]bool(nil), c.subscribes...)
	457	+	}
435	458
436	459		// fakePreviewSubscription is one server-side preview subscription whose
437	460		// Changed channel a test drives explicitly.


@@ -396,9 +396,11 @@ type fakeCore struct {
396		updated []updateHostPolicyCall	396		updated []updateHostPolicyCall
397		reconcile []string	397		reconcile []string
398		openErr error	398		openErr error
399	-	closed bool	399	+	// subscribes records each core subscription's reader kind (true = passive).
400	-	done chan struct{}	400	+	subscribes []bool
401	-	once sync.Once	401	+	closed bool
			402	+	done chan struct{}
			403	+	once sync.Once
402			404
403		// Membership result hooks. Their zero values keep the deterministic	405		// Membership result hooks. Their zero values keep the deterministic
404		// defaults that mirror the real registry: a fresh generation-1 registration	406		// defaults that mirror the real registry: a fresh generation-1 registration

@@ -431,7 +433,28 @@ func (c *fakeCore) Err() error { return nil }
431			433
432		func (c *fakeCore) Snapshot() ports.BrokerSnapshot { return c.hub.current() }	434		func (c *fakeCore) Snapshot() ports.BrokerSnapshot { return c.hub.current() }
433			435
434	-	func (c *fakeCore) Subscribe() (ports.BrokerSubscription, error) { return c.hub.subscribe(), nil }	436	+	func (c *fakeCore) Subscribe() (ports.BrokerSubscription, error) {
			437	+	return c.recordSubscribe(false), nil
			438	+	}
			439	+
			440	+	// SubscribePassive is the optional passive-reader capability the server uses
			441	+	// for a Subscribe flagged passive.
			442	+	func (c *fakeCore) SubscribePassive() (ports.BrokerSubscription, error) {
			443	+	return c.recordSubscribe(true), nil
			444	+	}
			445	+
			446	+	func (c fakeCore) recordSubscribe(passive bool) fakeSubscription {
			447	+	c.mu.Lock()
			448	+	c.subscribes = append(c.subscribes, passive)
			449	+	c.mu.Unlock()
			450	+	return c.hub.subscribe()
			451	+	}
			452	+
			453	+	func (c *fakeCore) subscribeKinds() []bool {
			454	+	c.mu.Lock()
			455	+	defer c.mu.Unlock()
			456	+	return append([]bool(nil), c.subscribes...)
			457	+	}
435			458
436		// fakePreviewSubscription is one server-side preview subscription whose	459		// fakePreviewSubscription is one server-side preview subscription whose
437		// Changed channel a test drives explicitly.	460		// Changed channel a test drives explicitly.

M internal/adapters/brokeripc/mutation.go +13 -2


@@ -38,12 +38,23 @@ type publisher struct {
38	38		wake chan struct{}
39	39		}
40	40
	41	+	// passiveSubscriber is the optional core capability behind a passive
	42	+	// Subscribe: a read-only snapshot subscription that never counts as
	43	+	// observation demand. A core without it serves every subscription as demand.
	44	+	type passiveSubscriber interface {
	45	+	SubscribePassive() (ports.BrokerSubscription, error)
	46	+	}
	47	+
41	48		// retargetPublisher replaces any active publisher with one for the new
42	49		// generation and publishes the current snapshot immediately. A failed
43	50		// subscription ends the superseded series first, so no publisher survives to
44	51		// publish under a generation the connection tracker already retired.
45		-	func (s *serverSession) retargetPublisher(generation brokerwire.SubscriptionGeneration) error {
46		-	sub, err := s.core.Subscribe()
	52	+	func (s *serverSession) retargetPublisher(generation brokerwire.SubscriptionGeneration, passive bool) error {
	53	+	subscribe := s.core.Subscribe
	54	+	if reader, ok := s.core.(passiveSubscriber); passive && ok {
	55	+	subscribe = reader.SubscribePassive
	56	+	}
	57	+	sub, err := subscribe()
47	58		if err != nil {
48	59		s.stopPublisher()
49	60		detail := errorDetail(err)

 	wake       chan struct{}
 }
+// passiveSubscriber is the optional core capability behind a passive
+// Subscribe: a read-only snapshot subscription that never counts as
+// observation demand. A core without it serves every subscription as demand.
+type passiveSubscriber interface {
+	SubscribePassive() (ports.BrokerSubscription, error)
+}
 // retargetPublisher replaces any active publisher with one for the new
 // generation and publishes the current snapshot immediately. A failed
 // subscription ends the superseded series first, so no publisher survives to
 // publish under a generation the connection tracker already retired.
-func (s *serverSession) retargetPublisher(generation brokerwire.SubscriptionGeneration) error {
+func (s *serverSession) retargetPublisher(generation brokerwire.SubscriptionGeneration, passive bool) error {
-	sub, err := s.core.Subscribe()
+	subscribe := s.core.Subscribe
+	if reader, ok := s.core.(passiveSubscriber); passive && ok {
+		subscribe = reader.SubscribePassive
+	}
+	sub, err := subscribe()
 	if err != nil {
 		s.stopPublisher()
 		detail := errorDetail(err)

M internal/adapters/brokeripc/server.go +1 -1


@@ -664,7 +664,7 @@ func (s *serverSession) dispatch(message brokerwire.ClientMessage) error {
664	664		if err := s.conn.Subscribe(m.Generation); err != nil {
665	665		return errors.Join(ErrProtocol, err)
666	666		}
667		-	return s.retargetPublisher(m.Generation)
	667	+	return s.retargetPublisher(m.Generation, m.Passive)
668	668		case brokerwire.Resync:
669	669		if !s.scopeMatches(m.Epoch, m.Connection) {
670	670		return s.refuseScope()


@@ -664,7 +664,7 @@ func (s *serverSession) dispatch(message brokerwire.ClientMessage) error {
664		if err := s.conn.Subscribe(m.Generation); err != nil {	664		if err := s.conn.Subscribe(m.Generation); err != nil {
665		return errors.Join(ErrProtocol, err)	665		return errors.Join(ErrProtocol, err)
666		}	666		}
667	-	return s.retargetPublisher(m.Generation)	667	+	return s.retargetPublisher(m.Generation, m.Passive)
668		case brokerwire.Resync:	668		case brokerwire.Resync:
669		if !s.scopeMatches(m.Epoch, m.Connection) {	669		if !s.scopeMatches(m.Epoch, m.Connection) {
670		return s.refuseScope()	670		return s.refuseScope()

M internal/adapters/brokeripc/session_test.go +29 -3


@@ -39,7 +39,7 @@ func TestResubscribePublishesCurrentSnapshotPerGeneration(t *testing.T) {
39	39		generation := brokerwire.SubscriptionGeneration(index)
40	40		core.hub.publish(testSnapshot(epoch, ports.BrokerRevision(index)))
41	41		require.NoError(t, session.conn.Subscribe(generation))
42		-	require.NoError(t, session.retargetPublisher(generation))
	42	+	require.NoError(t, session.retargetPublisher(generation, false))
43	43		}
44	44
45	45		// Each publisher owns its wake, so a later generation can never be left

@@ -79,14 +79,14 @@ func TestSupersededPublisherOwnsItsWake(t *testing.T) {
79	79
80	80		core.hub.publish(testSnapshot(epoch, 1))
81	81		require.NoError(t, session.conn.Subscribe(1))
82		-	require.NoError(t, session.retargetPublisher(1))
	82	+	require.NoError(t, session.retargetPublisher(1, false))
83	83		session.subMu.Lock()
84	84		first := session.pub
85	85		session.subMu.Unlock()
86	86
87	87		core.hub.publish(testSnapshot(epoch, 2))
88	88		require.NoError(t, session.conn.Subscribe(2))
89		-	require.NoError(t, session.retargetPublisher(2))
	89	+	require.NoError(t, session.retargetPublisher(2, false))
90	90		session.subMu.Lock()
91	91		second := session.pub
92	92		session.subMu.Unlock()

@@ -229,6 +229,32 @@ func TestSubscribePublishesSnapshot(t *testing.T) {
229	229		}
230	230		}
231	231
	232	+	// TestSubscribeCarriesPassiveReader proves a client configured as a passive
	233	+	// reader reaches the core's passive subscription, while a default client
	234	+	// subscribes as a watching client that records observation demand.
	235	+	func TestSubscribeCarriesPassiveReader(t *testing.T) {
	236	+	tests := []struct {
	237	+	name string
	238	+	passive bool
	239	+	}{
	240	+	{name: "watching client", passive: false},
	241	+	{name: "passive reader", passive: true},
	242	+	}
	243	+	for _, tt := range tests {
	244	+	t.Run(tt.name, func(t *testing.T) {
	245	+	e := startEndpoint(t, Config{})
	246	+	client := e.dialWith(Config{PassiveSubscribe: tt.passive})
	247	+	e.accept()
	248	+	sub, err := client.Subscribe()
	249	+	require.NoError(t, err)
	250	+	t.Cleanup(sub.Close)
	251	+	core := e.authority.last()
	252	+	require.Eventually(t, func() bool { return len(core.subscribeKinds()) == 1 }, 5time.Second, 10time.Millisecond)
	253	+	require.Equal(t, []bool{tt.passive}, core.subscribeKinds())
	254	+	})
	255	+	}
	256	+	}
	257	+
232	258		// TestSubscribeGenerationAdvancesAndReplaces proves a second Subscribe advances
233	259		// the generation series (refusing reuse) and replaces the previous
234	260		// subscription.

 			generation := brokerwire.SubscriptionGeneration(index)
 			core.hub.publish(testSnapshot(epoch, ports.BrokerRevision(index)))
 			require.NoError(t, session.conn.Subscribe(generation))
-			require.NoError(t, session.retargetPublisher(generation))
+			require.NoError(t, session.retargetPublisher(generation, false))
 		}
 		// Each publisher owns its wake, so a later generation can never be left
 	core.hub.publish(testSnapshot(epoch, 1))
 	require.NoError(t, session.conn.Subscribe(1))
-	require.NoError(t, session.retargetPublisher(1))
+	require.NoError(t, session.retargetPublisher(1, false))
 	session.subMu.Lock()
 	first := session.pub
 	session.subMu.Unlock()
 	core.hub.publish(testSnapshot(epoch, 2))
 	require.NoError(t, session.conn.Subscribe(2))
-	require.NoError(t, session.retargetPublisher(2))
+	require.NoError(t, session.retargetPublisher(2, false))
 	session.subMu.Lock()
 	second := session.pub
 	session.subMu.Unlock()
 	}
 }
+// TestSubscribeCarriesPassiveReader proves a client configured as a passive
+// reader reaches the core's passive subscription, while a default client
+// subscribes as a watching client that records observation demand.
+func TestSubscribeCarriesPassiveReader(t *testing.T) {
+	tests := []struct {
+		name    string
+		passive bool
+	}{
+		{name: "watching client", passive: false},
+		{name: "passive reader", passive: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			e := startEndpoint(t, Config{})
+			client := e.dialWith(Config{PassiveSubscribe: tt.passive})
+			e.accept()
+			sub, err := client.Subscribe()
+			require.NoError(t, err)
+			t.Cleanup(sub.Close)
+			core := e.authority.last()
+			require.Eventually(t, func() bool { return len(core.subscribeKinds()) == 1 }, 5*time.Second, 10*time.Millisecond)
+			require.Equal(t, []bool{tt.passive}, core.subscribeKinds())
+		})
+	}
+}
 // TestSubscribeGenerationAdvancesAndReplaces proves a second Subscribe advances
 // the generation series (refusing reuse) and replaces the previous
 // subscription.

M internal/adapters/brokerwire/codec.go +2 -2


@@ -116,7 +116,7 @@ func encodeClientEnvelope(message ClientMessage, maxChunkBytes uint64) (*wire.Br
116	116		return nil, ErrInvalidMessage
117	117		}
118	118		return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Subscribe{Subscribe: &wire.Subscribe{
119		-	Scope: scopeToWire(m.Epoch, m.Connection), Generation: m.Generation,
	119	+	Scope: scopeToWire(m.Epoch, m.Connection), Generation: m.Generation, Passive: m.Passive,
120	120		}}}, nil
121	121		case *Subscribe:
122	122		if m == nil {

@@ -491,7 +491,7 @@ func subscribeFromWire(message *wire.Subscribe) (Subscribe, error) {
491	491		if err != nil {
492	492		return Subscribe{}, ErrInvalidMessage
493	493		}
494		-	return Subscribe{Epoch: epoch, Connection: connection, Generation: message.GetGeneration()}, nil
	494	+	return Subscribe{Epoch: epoch, Connection: connection, Generation: message.GetGeneration(), Passive: message.GetPassive()}, nil
495	495		}
496	496
497	497		func resyncFromWire(message *wire.Resync) (Resync, error) {


@@ -116,7 +116,7 @@ func encodeClientEnvelope(message ClientMessage, maxChunkBytes uint64) (*wire.Br
116		return nil, ErrInvalidMessage	116		return nil, ErrInvalidMessage
117		}	117		}
118		return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Subscribe{Subscribe: &wire.Subscribe{	118		return &wire.BrokerClientEnvelope{Payload: &wire.BrokerClientEnvelope_Subscribe{Subscribe: &wire.Subscribe{
119	-	Scope: scopeToWire(m.Epoch, m.Connection), Generation: m.Generation,	119	+	Scope: scopeToWire(m.Epoch, m.Connection), Generation: m.Generation, Passive: m.Passive,
120		}}}, nil	120		}}}, nil
121		case *Subscribe:	121		case *Subscribe:
122		if m == nil {	122		if m == nil {

@@ -491,7 +491,7 @@ func subscribeFromWire(message *wire.Subscribe) (Subscribe, error) {
491		if err != nil {	491		if err != nil {
492		return Subscribe{}, ErrInvalidMessage	492		return Subscribe{}, ErrInvalidMessage
493		}	493		}
494	-	return Subscribe{Epoch: epoch, Connection: connection, Generation: message.GetGeneration()}, nil	494	+	return Subscribe{Epoch: epoch, Connection: connection, Generation: message.GetGeneration(), Passive: message.GetPassive()}, nil
495		}	495		}
496			496
497		func resyncFromWire(message *wire.Resync) (Resync, error) {	497		func resyncFromWire(message *wire.Resync) (Resync, error) {

M internal/adapters/brokerwire/codec_test.go +10 -0


@@ -803,6 +803,16 @@ func TestBrokerGoldenVectors(t *testing.T) {
803	803		back, err := DecodeClient(rawSub, testEnvelopeCeiling, testChunkCeiling)
804	804		require.NoError(t, err)
805	805		require.Equal(t, sub, back)
	806	+
	807	+	// The passive reader flag survives the round trip, and an active
	808	+	// subscription encodes without it (proto3 omits false).
	809	+	passive := sub
	810	+	passive.Passive = true
	811	+	rawPassive := mustEncodeClient(t, passive)
	812	+	back, err = DecodeClient(rawPassive, testEnvelopeCeiling, testChunkCeiling)
	813	+	require.NoError(t, err)
	814	+	require.Equal(t, passive, back)
	815	+	require.Greater(t, len(rawPassive), len(rawSub))
806	816		}
807	817
808	818		// TestBrokerBounds proves stateless bound refusals.


@@ -803,6 +803,16 @@ func TestBrokerGoldenVectors(t *testing.T) {
803	back, err := DecodeClient(rawSub, testEnvelopeCeiling, testChunkCeiling)	803		back, err := DecodeClient(rawSub, testEnvelopeCeiling, testChunkCeiling)
804	require.NoError(t, err)	804		require.NoError(t, err)
805	require.Equal(t, sub, back)	805		require.Equal(t, sub, back)
		806	+
		807	+	// The passive reader flag survives the round trip, and an active
		808	+	// subscription encodes without it (proto3 omits false).
		809	+	passive := sub
		810	+	passive.Passive = true
		811	+	rawPassive := mustEncodeClient(t, passive)
		812	+	back, err = DecodeClient(rawPassive, testEnvelopeCeiling, testChunkCeiling)
		813	+	require.NoError(t, err)
		814	+	require.Equal(t, passive, back)
		815	+	require.Greater(t, len(rawPassive), len(rawSub))
806	}	816		}
807		817
808	// TestBrokerBounds proves stateless bound refusals.	818		// TestBrokerBounds proves stateless bound refusals.

M internal/adapters/brokerwire/messages.go +3 -0


@@ -44,6 +44,9 @@ type Subscribe struct {
44	44		Epoch ports.BrokerEpoch
45	45		Connection ports.BrokerConnectionID
46	46		Generation SubscriptionGeneration
	47	+	// Passive marks a read-only snapshot reader that must never count as
	48	+	// observation demand.
	49	+	Passive bool
47	50		}
48	51
49	52		func (Subscribe) brokerClientMessage() {}


@@ -44,6 +44,9 @@ type Subscribe struct {
44	Epoch ports.BrokerEpoch	44		Epoch ports.BrokerEpoch
45	Connection ports.BrokerConnectionID	45		Connection ports.BrokerConnectionID
46	Generation SubscriptionGeneration	46		Generation SubscriptionGeneration
		47	+	// Passive marks a read-only snapshot reader that must never count as
		48	+	// observation demand.
		49	+	Passive bool
47	}	50		}
48		51
49	func (Subscribe) brokerClientMessage() {}	52		func (Subscribe) brokerClientMessage() {}

M internal/app/broker_production_connect.go +1 -1


@@ -16,7 +16,7 @@ const productionBrokerStartupTimeout = 10 * time.Second
16	16		// connectExistingBroker dials without ensuring or spawning a broker.
17	17		func connectExistingBroker(ctx context.Context) (ports.BrokerService, error) {
18	18		path := brokeripc.SocketPath(productionBrokerLayout().Runtime)
19		-	service, err := brokeripc.NewConnector(path, brokeripc.Config{}).Connect(ctx)
	19	+	service, err := brokeripc.NewConnector(path, brokeripc.Config{PassiveSubscribe: true}).Connect(ctx)
20	20		if err != nil {
21	21		if backendAbsent(err) {
22	22		return nil, absentError(path)


@@ -16,7 +16,7 @@ const productionBrokerStartupTimeout = 10 * time.Second
16		// connectExistingBroker dials without ensuring or spawning a broker.	16		// connectExistingBroker dials without ensuring or spawning a broker.
17		func connectExistingBroker(ctx context.Context) (ports.BrokerService, error) {	17		func connectExistingBroker(ctx context.Context) (ports.BrokerService, error) {
18		path := brokeripc.SocketPath(productionBrokerLayout().Runtime)	18		path := brokeripc.SocketPath(productionBrokerLayout().Runtime)
19	-	service, err := brokeripc.NewConnector(path, brokeripc.Config{}).Connect(ctx)	19	+	service, err := brokeripc.NewConnector(path, brokeripc.Config{PassiveSubscribe: true}).Connect(ctx)
20		if err != nil {	20		if err != nil {
21		if backendAbsent(err) {	21		if backendAbsent(err) {
22		return nil, absentError(path)	22		return nil, absentError(path)

M internal/protocol/wire/broker.pb.go +17 -5 generated file: collapsed

load diff

M internal/protocol/wire/schema/broker.proto +4 -0


@@ -116,6 +116,10 @@ message Register {}
116	116		message Subscribe {
117	117		BrokerScope scope = 1;
118	118		uint64 generation = 2;
	119	+	// passive marks a read-only snapshot reader (vev ls, vev host list). It
	120	+	// receives publications but never counts as observation demand, so reading
	121	+	// state never triggers a probe.
	122	+	bool passive = 3;
119	123		}
120	124
121	125		// Resync asks for a full snapshot publication after an epoch change or a


@@ -116,6 +116,10 @@ message Register {}
116	message Subscribe {	116		message Subscribe {
117	BrokerScope scope = 1;	117		BrokerScope scope = 1;
118	uint64 generation = 2;	118		uint64 generation = 2;
		119	+	// passive marks a read-only snapshot reader (vev ls, vev host list). It
		120	+	// receives publications but never counts as observation demand, so reading
		121	+	// state never triggers a probe.
		122	+	bool passive = 3;
119	}	123		}
120		124
121	// Resync asks for a full snapshot publication after an epoch change or a	125		// Resync asks for a full snapshot publication after an epoch change or a

M internal/usecase/broker/service.go +19 -3


@@ -230,14 +230,27 @@ func (s *Service) Snapshot() ports.BrokerSnapshot { return s.registry.Snapshot()
230	230		// watches the snapshot, the registry observes on the demand cadence. Without
231	231		// a subscriber it performs no scheduled observations. Close balances demand.
232	232		func (s *Service) Subscribe() (ports.BrokerSubscription, error) {
	233	+	return s.subscribe(true)
	234	+	}
	235	+
	236	+	// SubscribePassive is Subscribe for a read-only snapshot reader (vev ls, vev
	237	+	// host list): it receives every publication but records no demand, so reading
	238	+	// state never triggers an observation.
	239	+	func (s *Service) SubscribePassive() (ports.BrokerSubscription, error) {
	240	+	return s.subscribe(false)
	241	+	}
	242	+
	243	+	func (s *Service) subscribe(demand bool) (ports.BrokerSubscription, error) {
233	244		s.mu.Lock()
234	245		if s.closed {
235	246		s.mu.Unlock()
236	247		return nil, ports.BrokerAdmissionClosed
237	248		}
238		-	sub := &serviceSubscription{service: s, inner: s.registry.Subscribe()}
	249	+	sub := &serviceSubscription{service: s, inner: s.registry.Subscribe(), demand: demand}
239	250		s.subs[sub] = struct{}{}
240		-	s.registry.SetDemand(true)
	251	+	if demand {
	252	+	s.registry.SetDemand(true)
	253	+	}
241	254		s.mu.Unlock()
242	255		return sub, nil
243	256		}

@@ -462,6 +475,7 @@ func (s *Service) openContext(ctx context.Context) (context.Context, func()) {
462	475		type serviceSubscription struct {
463	476		service *Service
464	477		inner ports.BrokerSubscription
	478	+	demand bool
465	479		once sync.Once
466	480		}
467	481

@@ -472,7 +486,9 @@ func (s *serviceSubscription) Close() {
472	486		s.inner.Close()
473	487		s.service.mu.Lock()
474	488		delete(s.service.subs, s)
475		-	s.service.registry.SetDemand(false)
	489	+	if s.demand {
	490	+	s.service.registry.SetDemand(false)
	491	+	}
476	492		s.service.mu.Unlock()
477	493		})
478	494		}

 // watches the snapshot, the registry observes on the demand cadence. Without
 // a subscriber it performs no scheduled observations. Close balances demand.
 func (s *Service) Subscribe() (ports.BrokerSubscription, error) {
+	return s.subscribe(true)
+}
+// SubscribePassive is Subscribe for a read-only snapshot reader (vev ls, vev
+// host list): it receives every publication but records no demand, so reading
+// state never triggers an observation.
+func (s *Service) SubscribePassive() (ports.BrokerSubscription, error) {
+	return s.subscribe(false)
+}
+func (s *Service) subscribe(demand bool) (ports.BrokerSubscription, error) {
 	s.mu.Lock()
 	if s.closed {
 		s.mu.Unlock()
 		return nil, ports.BrokerAdmissionClosed
 	}
-	sub := &serviceSubscription{service: s, inner: s.registry.Subscribe()}
+	sub := &serviceSubscription{service: s, inner: s.registry.Subscribe(), demand: demand}
 	s.subs[sub] = struct{}{}
-	s.registry.SetDemand(true)
+	if demand {
+		s.registry.SetDemand(true)
+	}
 	s.mu.Unlock()
 	return sub, nil
 }
 type serviceSubscription struct {
 	service *Service
 	inner   ports.BrokerSubscription
+	demand  bool
 	once    sync.Once
 }
 		s.inner.Close()
 		s.service.mu.Lock()
 		delete(s.service.subs, s)
-		s.service.registry.SetDemand(false)
+		if s.demand {
+			s.service.registry.SetDemand(false)
+		}
 		s.service.mu.Unlock()
 	})
 }

M internal/usecase/broker/service_test.go +42 -0


@@ -507,6 +507,48 @@ func TestServiceSubscribeDrivesRegistryDemand(t *testing.T) {
507	507		require.Zero(t, registryDemandCount(registry))
508	508		}
509	509
	510	+	// TestServiceSubscribeDemandByReader pins the Amendment 2 rule "reading state
	511	+	// forces nothing": a passive subscription (vev ls, vev host list) receives
	512	+	// publications but never records demand, while a watching client does.
	513	+	func TestServiceSubscribeDemandByReader(t *testing.T) {
	514	+	tests := []struct {
	515	+	name string
	516	+	passive bool
	517	+	wantDemand int
	518	+	}{
	519	+	{name: "watching client records demand", passive: false, wantDemand: 1},
	520	+	{name: "passive reader records none", passive: true, wantDemand: 0},
	521	+	}
	522	+	for _, tt := range tests {
	523	+	t.Run(tt.name, func(t *testing.T) {
	524	+	authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
	525	+	admitted, err := authority.AdmitClient(context.Background())
	526	+	require.NoError(t, err)
	527	+	t.Cleanup(func() { require.NoError(t, admitted.Close()) })
	528	+	service := admitted.(*Service)
	529	+
	530	+	subscribe := service.Subscribe
	531	+	if tt.passive {
	532	+	subscribe = service.SubscribePassive
	533	+	}
	534	+	sub, err := subscribe()
	535	+	require.NoError(t, err)
	536	+	require.Equal(t, tt.wantDemand, registryDemandCount(registry))
	537	+
	538	+	require.NoError(t, registry.setHosts(hostRecords(registration(t, "user@host:22", 1))))
	539	+	select {
	540	+	case <-sub.Changed():
	541	+	case <-time.After(time.Second):
	542	+	t.Fatal("every subscription must observe registry publications")
	543	+	}
	544	+
	545	+	sub.Close()
	546	+	sub.Close()
	547	+	require.Zero(t, registryDemandCount(registry), "closing must balance exactly the demand it recorded")
	548	+	})
	549	+	}
	550	+	}
	551	+
510	552		func TestServiceSnapshotDelegates(t *testing.T) {
511	553		authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
512	554		service, err := authority.AdmitClient(context.Background())


@@ -507,6 +507,48 @@ func TestServiceSubscribeDrivesRegistryDemand(t *testing.T) {
507	require.Zero(t, registryDemandCount(registry))	507		require.Zero(t, registryDemandCount(registry))
508	}	508		}
509		509
		510	+	// TestServiceSubscribeDemandByReader pins the Amendment 2 rule "reading state
		511	+	// forces nothing": a passive subscription (vev ls, vev host list) receives
		512	+	// publications but never records demand, while a watching client does.
		513	+	func TestServiceSubscribeDemandByReader(t *testing.T) {
		514	+	tests := []struct {
		515	+	name string
		516	+	passive bool
		517	+	wantDemand int
		518	+	}{
		519	+	{name: "watching client records demand", passive: false, wantDemand: 1},
		520	+	{name: "passive reader records none", passive: true, wantDemand: 0},
		521	+	}
		522	+	for _, tt := range tests {
		523	+	t.Run(tt.name, func(t *testing.T) {
		524	+	authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
		525	+	admitted, err := authority.AdmitClient(context.Background())
		526	+	require.NoError(t, err)
		527	+	t.Cleanup(func() { require.NoError(t, admitted.Close()) })
		528	+	service := admitted.(*Service)
		529	+
		530	+	subscribe := service.Subscribe
		531	+	if tt.passive {
		532	+	subscribe = service.SubscribePassive
		533	+	}
		534	+	sub, err := subscribe()
		535	+	require.NoError(t, err)
		536	+	require.Equal(t, tt.wantDemand, registryDemandCount(registry))
		537	+
		538	+	require.NoError(t, registry.setHosts(hostRecords(registration(t, "user@host:22", 1))))
		539	+	select {
		540	+	case <-sub.Changed():
		541	+	case <-time.After(time.Second):
		542	+	t.Fatal("every subscription must observe registry publications")
		543	+	}
		544	+
		545	+	sub.Close()
		546	+	sub.Close()
		547	+	require.Zero(t, registryDemandCount(registry), "closing must balance exactly the demand it recorded")
		548	+	})
		549	+	}
		550	+	}
		551	+
510	func TestServiceSnapshotDelegates(t *testing.T) {	552		func TestServiceSnapshotDelegates(t *testing.T) {
511	authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)	553		authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
512	service, err := authority.AdmitClient(context.Background())	554		service, err := authority.AdmitClient(context.Background())