test(adapters,broker): move same-package-only test seams into export_test.go

listener.inFlight/liveSessions/refusal, brokeripc.DefaultSocketPath,
brokerwire.WithMaxStagedBytes, daemonmux ServerBinding.Accepts/Policy/Origin,
NewStreamEngine, NewListener, NewScheduler, ServerSupervisor.Children,
DecodePreambleRequestBytes, ipc.SocketPath, quic.WithRuntimeObserver, and
broker.NewRegistry/setHosts are all legitimate test conveniences over live
production constructors/accessors, with every caller confirmed to be a
same-package test. They move into a per-package export_test.go so they no
longer count as unreachable production code, without changing what any test
calls or asserts.

ipc.SocketPath has no current test caller (brokeripc.SocketPath, a distinct
function, is what every call site actually uses); it moves anyway per the
documented seam list, flagged here for the owner to reconsider deleting
outright.

daemonmux.NewServerBinding, brokeripc.SocketDir, brokerwire.MutationKindOf,
and snapshot.NewRepository are NOT touched: each is called from a different
package's tests (internal/app, internal/app, brokeripc, internal/app
respectively), and Go's build model cannot export a _test.go symbol across
package boundaries, so they must stay production code. They remain flagged
by deadcode and are documented in the phase report as accepted exceptions.

commit: d84053d662c6cad7e55d90ce4302491717198045
author: brice · 29d ago · 2026-09-23 01:56 UTC
parent: e312f75
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

18 files · +220 -200

A internal/adapters/brokeripc/export_test.go +34 -0

@@ -0,0 +1,34 @@
	1	+	package brokeripc
	2	+
	3	+	// Test-only accessors (export_test.go pattern): these observe listener
	4	+	// internals that only this package's own tests need. They are kept out of
	5	+	// the production build so a live server's diagnostics or capacity never
	6	+	// depend on a test-only accessor.
	7	+
	8	+	// refusal reports the most recent accept refusal, or nil if the listener has
	9	+	// refused nothing yet.
	10	+	func (l *listener) refusal() error {
	11	+	l.mu.Lock()
	12	+	defer l.mu.Unlock()
	13	+	return l.lastRefuse
	14	+	}
	15	+
	16	+	// inFlight reports the count of accepted carriages still completing their
	17	+	// handshake.
	18	+	func (l *listener) inFlight() int {
	19	+	l.mu.Lock()
	20	+	defer l.mu.Unlock()
	21	+	return len(l.pending)
	22	+	}
	23	+
	24	+	// liveSessions reports the count of admitted, still-open sessions.
	25	+	func (l *listener) liveSessions() int {
	26	+	l.mu.Lock()
	27	+	defer l.mu.Unlock()
	28	+	return len(l.sessions)
	29	+	}
	30	+
	31	+	// DefaultSocketPath returns the broker endpoint path under SocketFileName's
	32	+	// convenience default runtime directory, for tests that need it without
	33	+	// composing SocketPath(SocketDir()) themselves.
	34	+	func DefaultSocketPath() string { return SocketPath(SocketDir()) }

@@ -0,0 +1,34 @@
	1	+	package brokeripc
	2	+
	3	+	// Test-only accessors (export_test.go pattern): these observe listener
	4	+	// internals that only this package's own tests need. They are kept out of
	5	+	// the production build so a live server's diagnostics or capacity never
	6	+	// depend on a test-only accessor.
	7	+
	8	+	// refusal reports the most recent accept refusal, or nil if the listener has
	9	+	// refused nothing yet.
	10	+	func (l *listener) refusal() error {
	11	+	l.mu.Lock()
	12	+	defer l.mu.Unlock()
	13	+	return l.lastRefuse
	14	+	}
	15	+
	16	+	// inFlight reports the count of accepted carriages still completing their
	17	+	// handshake.
	18	+	func (l *listener) inFlight() int {
	19	+	l.mu.Lock()
	20	+	defer l.mu.Unlock()
	21	+	return len(l.pending)
	22	+	}
	23	+
	24	+	// liveSessions reports the count of admitted, still-open sessions.
	25	+	func (l *listener) liveSessions() int {
	26	+	l.mu.Lock()
	27	+	defer l.mu.Unlock()
	28	+	return len(l.sessions)
	29	+	}
	30	+
	31	+	// DefaultSocketPath returns the broker endpoint path under SocketFileName's
	32	+	// convenience default runtime directory, for tests that need it without
	33	+	// composing SocketPath(SocketDir()) themselves.
	34	+	func DefaultSocketPath() string { return SocketPath(SocketDir()) }

M internal/adapters/brokeripc/server.go +0 -22


@@ -210,13 +210,6 @@ func (l *listener) confirmRefusal(err error) {
210	210		l.mu.Unlock()
211	211		}
212	212
213		-	// refusal reports the most recent refused or failed admission.
214		-	func (l *listener) refusal() error {
215		-	l.mu.Lock()
216		-	defer l.mu.Unlock()
217		-	return l.lastRefuse
218		-	}
219		-
220	213		// admit completes the preamble and broker admission for one accepted carriage
221	214		// and publishes the running session. Slot ownership transfers to the session on
222	215		// success.

@@ -266,14 +259,6 @@ func (l *listener) closePending() {
266	259		}
267	260		}
268	261
269		-	// inFlight reports how many accepted carriages are still completing preamble or
270		-	// admission.
271		-	func (l *listener) inFlight() int {
272		-	l.mu.Lock()
273		-	defer l.mu.Unlock()
274		-	return len(l.pending)
275		-	}
276		-
277	262		// classifyAdmissionFailure maps one preamble or authority failure. Once the
278	263		// listener is closing, an accepted client that never reached admission was
279	264		// dropped by the close, so the failure is reported as ErrListenerClosed. A

@@ -446,13 +431,6 @@ func (l listener) forgetSession(session serverSession) {
446	431		l.mu.Unlock()
447	432		}
448	433
449		-	// liveSessions reports how many admitted sessions are still live.
450		-	func (l *listener) liveSessions() int {
451		-	l.mu.Lock()
452		-	defer l.mu.Unlock()
453		-	return len(l.sessions)
454		-	}
455		-
456	434		func (l *listener) isClosed() bool {
457	435		if l == nil {
458	436		return true


@@ -210,13 +210,6 @@ func (l *listener) confirmRefusal(err error) {
210		l.mu.Unlock()	210	l.mu.Unlock()
211		}	211	}
212			212
213	-	// refusal reports the most recent refused or failed admission.
214	-	func (l *listener) refusal() error {
215	-	l.mu.Lock()
216	-	defer l.mu.Unlock()
217	-	return l.lastRefuse
218	-	}
219	-
220		// admit completes the preamble and broker admission for one accepted carriage	213	// admit completes the preamble and broker admission for one accepted carriage
221		// and publishes the running session. Slot ownership transfers to the session on	214	// and publishes the running session. Slot ownership transfers to the session on
222		// success.	215	// success.

@@ -266,14 +259,6 @@ func (l *listener) closePending() {
266		}	259	}
267		}	260	}
268			261
269	-	// inFlight reports how many accepted carriages are still completing preamble or
270	-	// admission.
271	-	func (l *listener) inFlight() int {
272	-	l.mu.Lock()
273	-	defer l.mu.Unlock()
274	-	return len(l.pending)
275	-	}
276	-
277		// classifyAdmissionFailure maps one preamble or authority failure. Once the	262	// classifyAdmissionFailure maps one preamble or authority failure. Once the
278		// listener is closing, an accepted client that never reached admission was	263	// listener is closing, an accepted client that never reached admission was
279		// dropped by the close, so the failure is reported as ErrListenerClosed. A	264	// dropped by the close, so the failure is reported as ErrListenerClosed. A

@@ -446,13 +431,6 @@ func (l listener) forgetSession(session serverSession) {
446		l.mu.Unlock()	431	l.mu.Unlock()
447		}	432	}
448			433
449	-	// liveSessions reports how many admitted sessions are still live.
450	-	func (l *listener) liveSessions() int {
451	-	l.mu.Lock()
452	-	defer l.mu.Unlock()
453	-	return len(l.sessions)
454	-	}
455	-
456		func (l *listener) isClosed() bool {	434	func (l *listener) isClosed() bool {
457		if l == nil {	435	if l == nil {
458		return true	436	return true

M internal/adapters/brokeripc/socket.go +0 -5


@@ -23,8 +23,3 @@ func SocketPath(runtimeDir string) string {
23	23		// well, so both endpoints share one owner-only directory while keeping
24	24		// distinct socket names.
25	25		func SocketDir() string { return ipc.SocketDir() }
26		-
27		-	// DefaultSocketPath returns the broker endpoint path in the per-user runtime
28		-	// directory. Callers that need a private temporary endpoint supply their own
29		-	// parent directory to SocketPath instead.
30		-	func DefaultSocketPath() string { return SocketPath(SocketDir()) }


@@ -23,8 +23,3 @@ func SocketPath(runtimeDir string) string {
23		// well, so both endpoints share one owner-only directory while keeping	23	// well, so both endpoints share one owner-only directory while keeping
24		// distinct socket names.	24	// distinct socket names.
25		func SocketDir() string { return ipc.SocketDir() }	25	func SocketDir() string { return ipc.SocketDir() }
26	-
27	-	// DefaultSocketPath returns the broker endpoint path in the per-user runtime
28	-	// directory. Callers that need a private temporary endpoint supply their own
29	-	// parent directory to SocketPath instead.
30	-	func DefaultSocketPath() string { return SocketPath(SocketDir()) }

A internal/adapters/brokerwire/export_test.go +10 -0

@@ -0,0 +1,10 @@
	1	+	package brokerwire
	2	+
	3	+	// WithMaxStagedBytes overrides the staged estimate ceiling. It exists only for
	4	+	// tests that prove the bound without allocating the production-sized default;
	5	+	// production never needs a non-default ceiling.
	6	+	func WithMaxStagedBytes(n uint64) SnapshotOption {
	7	+	return func(a *SnapshotAssembler) {
	8	+	a.maxStagedBytes = n
	9	+	}
	10	+	}

@@ -0,0 +1,10 @@
	1	+	package brokerwire
	2	+
	3	+	// WithMaxStagedBytes overrides the staged estimate ceiling. It exists only for
	4	+	// tests that prove the bound without allocating the production-sized default;
	5	+	// production never needs a non-default ceiling.
	6	+	func WithMaxStagedBytes(n uint64) SnapshotOption {
	7	+	return func(a *SnapshotAssembler) {
	8	+	a.maxStagedBytes = n
	9	+	}
	10	+	}

M internal/adapters/brokerwire/snapshot.go +0 -7


@@ -90,13 +90,6 @@ var (
90	90		// to prove the bound without allocating 80 MiB; production uses defaults.
91	91		type SnapshotOption func(*SnapshotAssembler)
92	92
93		-	// WithMaxStagedBytes overrides the staged estimate ceiling.
94		-	func WithMaxStagedBytes(n uint64) SnapshotOption {
95		-	return func(a *SnapshotAssembler) {
96		-	a.maxStagedBytes = n
97		-	}
98		-	}
99		-
100	93		// WithGeneration pins the subscription generation this assembler belongs to,
101	94		// so a part of any other generation is fenced from the first one. The zero
102	95		// default adopts the generation from the first Begin.


@@ -90,13 +90,6 @@ var (
90		// to prove the bound without allocating 80 MiB; production uses defaults.	90	// to prove the bound without allocating 80 MiB; production uses defaults.
91		type SnapshotOption func(*SnapshotAssembler)	91	type SnapshotOption func(*SnapshotAssembler)
92			92
93	-	// WithMaxStagedBytes overrides the staged estimate ceiling.
94	-	func WithMaxStagedBytes(n uint64) SnapshotOption {
95	-	return func(a *SnapshotAssembler) {
96	-	a.maxStagedBytes = n
97	-	}
98	-	}
99	-
100		// WithGeneration pins the subscription generation this assembler belongs to,	93	// WithGeneration pins the subscription generation this assembler belongs to,
101		// so a part of any other generation is fenced from the first one. The zero	94	// so a part of any other generation is fenced from the first one. The zero
102		// default adopts the generation from the first Begin.	95	// default adopts the generation from the first Begin.

M internal/adapters/daemonmux/binding.go +0 -23


@@ -67,24 +67,6 @@ func NewServerBinding(identity ports.BrokerDaemonIdentity, incarnation ports.Bro
67	67		func (b ServerBinding) Identity() ports.BrokerDaemonIdentity { return b.identity }
68	68		func (b ServerBinding) Incarnation() ports.BrokerDaemonIncarnation { return b.incarnation }
69	69
70		-	// Policy returns the sole entry's policy for legacy single-policy callers. It
71		-	// returns the zero value for a multi-entry authority; admission never uses it.
72		-	func (b ServerBinding) Policy() ports.BrokerPolicy {
73		-	if len(b.entries) == 1 {
74		-	return b.entries[0].Policy
75		-	}
76		-	return ports.BrokerPolicy{}
77		-	}
78		-
79		-	// Origin returns the sole entry's locality for legacy single-policy callers. It
80		-	// returns the unknown origin for a multi-entry authority.
81		-	func (b ServerBinding) Origin() ports.SessionConnectionOrigin {
82		-	if len(b.entries) == 1 {
83		-	return b.entries[0].Origin
84		-	}
85		-	return ports.SessionOriginUnknown
86		-	}
87		-
88	70		func (b ServerBinding) Validate() error {
89	71		if b.identity.Validate() != nil \|\| b.incarnation.Validate() != nil \|\| len(b.entries) == 0 {
90	72		return ErrInvalidBinding

@@ -116,8 +98,3 @@ func (b ServerBinding) Accepted(policy ports.BrokerPolicy) (ServerPolicyAdmissio
116	98		}
117	99		return ServerPolicyAdmission{}, false
118	100		}
119		-
120		-	func (b ServerBinding) Accepts(policy ports.BrokerPolicy) bool {
121		-	_, ok := b.Accepted(policy)
122		-	return ok
123		-	}


@@ -67,24 +67,6 @@ func NewServerBinding(identity ports.BrokerDaemonIdentity, incarnation ports.Bro
67		func (b ServerBinding) Identity() ports.BrokerDaemonIdentity { return b.identity }	67	func (b ServerBinding) Identity() ports.BrokerDaemonIdentity { return b.identity }
68		func (b ServerBinding) Incarnation() ports.BrokerDaemonIncarnation { return b.incarnation }	68	func (b ServerBinding) Incarnation() ports.BrokerDaemonIncarnation { return b.incarnation }
69			69
70	-	// Policy returns the sole entry's policy for legacy single-policy callers. It
71	-	// returns the zero value for a multi-entry authority; admission never uses it.
72	-	func (b ServerBinding) Policy() ports.BrokerPolicy {
73	-	if len(b.entries) == 1 {
74	-	return b.entries[0].Policy
75	-	}
76	-	return ports.BrokerPolicy{}
77	-	}
78	-
79	-	// Origin returns the sole entry's locality for legacy single-policy callers. It
80	-	// returns the unknown origin for a multi-entry authority.
81	-	func (b ServerBinding) Origin() ports.SessionConnectionOrigin {
82	-	if len(b.entries) == 1 {
83	-	return b.entries[0].Origin
84	-	}
85	-	return ports.SessionOriginUnknown
86	-	}
87	-
88		func (b ServerBinding) Validate() error {	70	func (b ServerBinding) Validate() error {
89		if b.identity.Validate() != nil \|\| b.incarnation.Validate() != nil \|\| len(b.entries) == 0 {	71	if b.identity.Validate() != nil \|\| b.incarnation.Validate() != nil \|\| len(b.entries) == 0 {
90		return ErrInvalidBinding	72	return ErrInvalidBinding

@@ -116,8 +98,3 @@ func (b ServerBinding) Accepted(policy ports.BrokerPolicy) (ServerPolicyAdmissio
116		}	98	}
117		return ServerPolicyAdmission{}, false	99	return ServerPolicyAdmission{}, false
118		}	100	}
119	-
120	-	func (b ServerBinding) Accepts(policy ports.BrokerPolicy) bool {
121	-	_, ok := b.Accepted(policy)
122	-	return ok
123	-	}

M internal/adapters/daemonmux/engine_state.go +0 -8


@@ -234,14 +234,6 @@ type StreamEngine struct {
234	234		policy ports.BrokerPolicy
235	235		}
236	236
237		-	// NewStreamEngine returns an open engine bound to one physical connection
238		-	// under the default MuxCeilings: the full envelope ceiling, the maximum
239		-	// chunk, the maximum stream count (pending-plus-live), and the maximum
240		-	// aggregate buffered bytes.
241		-	func NewStreamEngine() *StreamEngine {
242		-	return newStreamEngine(DefaultMuxCeilings())
243		-	}
244		-
245	237		// NewStreamEngineWithCeilings returns an open engine bound to one physical
246	238		// connection whose admission, chunk, and aggregate budgets enforce the given
247	239		// negotiated MuxCeilings. The value is validated and copied, so the ceilings


@@ -234,14 +234,6 @@ type StreamEngine struct {
234		policy ports.BrokerPolicy	234	policy ports.BrokerPolicy
235		}	235	}
236			236
237	-	// NewStreamEngine returns an open engine bound to one physical connection
238	-	// under the default MuxCeilings: the full envelope ceiling, the maximum
239	-	// chunk, the maximum stream count (pending-plus-live), and the maximum
240	-	// aggregate buffered bytes.
241	-	func NewStreamEngine() *StreamEngine {
242	-	return newStreamEngine(DefaultMuxCeilings())
243	-	}
244	-
245		// NewStreamEngineWithCeilings returns an open engine bound to one physical	237	// NewStreamEngineWithCeilings returns an open engine bound to one physical
246		// connection whose admission, chunk, and aggregate budgets enforce the given	238	// connection whose admission, chunk, and aggregate budgets enforce the given
247		// negotiated MuxCeilings. The value is validated and copied, so the ceilings	239	// negotiated MuxCeilings. The value is validated and copied, so the ceilings

A internal/adapters/daemonmux/export_test.go +87 -0

@@ -0,0 +1,87 @@
	1	+	package daemonmux
	2	+
	3	+	import (
	4	+	"google.golang.org/protobuf/proto"
	5	+
	6	+	"github.com/bnema/vev/internal/adapters/clock"
	7	+	"github.com/bnema/vev/internal/ports"
	8	+	"github.com/bnema/vev/internal/protocol"
	9	+	"github.com/bnema/vev/internal/protocol/wire"
	10	+	)
	11	+
	12	+	// Test-only convenience constructors and accessors (export_test.go pattern):
	13	+	// production always supplies its own explicit ceilings, clock, or budget
	14	+	// (see pump.go, supervisor.go), so these package-default shorthands only
	15	+	// exist for tests that don't care about a non-default value.
	16	+
	17	+	// Accepts reports whether policy matches an admitted entry, for tests over the
	18	+	// legacy single-policy binding shape.
	19	+	func (b ServerBinding) Accepts(policy ports.BrokerPolicy) bool {
	20	+	_, ok := b.Accepted(policy)
	21	+	return ok
	22	+	}
	23	+
	24	+	// Policy returns the sole entry's policy for legacy single-policy test
	25	+	// fixtures. It returns the zero value for a multi-entry authority.
	26	+	func (b ServerBinding) Policy() ports.BrokerPolicy {
	27	+	if len(b.entries) == 1 {
	28	+	return b.entries[0].Policy
	29	+	}
	30	+	return ports.BrokerPolicy{}
	31	+	}
	32	+
	33	+	// Origin returns the sole entry's locality for legacy single-policy test
	34	+	// fixtures. It returns the unknown origin for a multi-entry authority.
	35	+	func (b ServerBinding) Origin() ports.SessionConnectionOrigin {
	36	+	if len(b.entries) == 1 {
	37	+	return b.entries[0].Origin
	38	+	}
	39	+	return ports.SessionOriginUnknown
	40	+	}
	41	+
	42	+	// NewStreamEngine returns an open engine bound to one physical connection
	43	+	// under the package default MuxCeilings.
	44	+	func NewStreamEngine() *StreamEngine {
	45	+	return newStreamEngine(DefaultMuxCeilings())
	46	+	}
	47	+
	48	+	// NewListener returns a listener under the package default handshake budget,
	49	+	// accept-queue bound, and wall clock.
	50	+	func NewListener(pump Pump, accepted ServerPolicyAdmission) (Listener, error) {
	51	+	return newListener(pump, protocol.HandshakeTimeout, MaxAcceptQueue, clock.New(), accepted)
	52	+	}
	53	+
	54	+	// NewScheduler returns an empty outbound scheduler under the package default
	55	+	// MuxCeilings.
	56	+	func NewScheduler(engine StreamEngine) Scheduler {
	57	+	return newScheduler(engine, DefaultMuxCeilings())
	58	+	}
	59	+
	60	+	// Children reports the count of currently owned physical children.
	61	+	func (s *ServerSupervisor) Children() int {
	62	+	if s == nil {
	63	+	return 0
	64	+	}
	65	+	s.mu.Lock()
	66	+	defer s.mu.Unlock()
	67	+	return len(s.children)
	68	+	}
	69	+
	70	+	// DecodePreambleRequestBytes strictly decodes one serialized daemonmux client
	71	+	// preamble under the 4 KiB bound, the strict scan, the generated unmarshal,
	72	+	// and DecodePreambleRequest's semantic checks. Production decodes a preamble
	73	+	// inline through the same steps (see handshake.go); this composed helper
	74	+	// exists for tests and the package fuzz target.
	75	+	func DecodePreambleRequestBytes(payload []byte) (MuxPreambleRequest, error) {
	76	+	if err := CheckPreambleSize(payload); err != nil {
	77	+	return MuxPreambleRequest{}, err
	78	+	}
	79	+	message := &wire.MuxPreambleRequest{}
	80	+	if err := wire.ScanEnvelope(message, payload); err != nil {
	81	+	return MuxPreambleRequest{}, err
	82	+	}
	83	+	if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, message)); err != nil {
	84	+	return MuxPreambleRequest{}, err
	85	+	}
	86	+	return DecodePreambleRequest(message)
	87	+	}

@@ -0,0 +1,87 @@
	1	+	package daemonmux
	2	+
	3	+	import (
	4	+	"google.golang.org/protobuf/proto"
	5	+
	6	+	"github.com/bnema/vev/internal/adapters/clock"
	7	+	"github.com/bnema/vev/internal/ports"
	8	+	"github.com/bnema/vev/internal/protocol"
	9	+	"github.com/bnema/vev/internal/protocol/wire"
	10	+	)
	11	+
	12	+	// Test-only convenience constructors and accessors (export_test.go pattern):
	13	+	// production always supplies its own explicit ceilings, clock, or budget
	14	+	// (see pump.go, supervisor.go), so these package-default shorthands only
	15	+	// exist for tests that don't care about a non-default value.
	16	+
	17	+	// Accepts reports whether policy matches an admitted entry, for tests over the
	18	+	// legacy single-policy binding shape.
	19	+	func (b ServerBinding) Accepts(policy ports.BrokerPolicy) bool {
	20	+	_, ok := b.Accepted(policy)
	21	+	return ok
	22	+	}
	23	+
	24	+	// Policy returns the sole entry's policy for legacy single-policy test
	25	+	// fixtures. It returns the zero value for a multi-entry authority.
	26	+	func (b ServerBinding) Policy() ports.BrokerPolicy {
	27	+	if len(b.entries) == 1 {
	28	+	return b.entries[0].Policy
	29	+	}
	30	+	return ports.BrokerPolicy{}
	31	+	}
	32	+
	33	+	// Origin returns the sole entry's locality for legacy single-policy test
	34	+	// fixtures. It returns the unknown origin for a multi-entry authority.
	35	+	func (b ServerBinding) Origin() ports.SessionConnectionOrigin {
	36	+	if len(b.entries) == 1 {
	37	+	return b.entries[0].Origin
	38	+	}
	39	+	return ports.SessionOriginUnknown
	40	+	}
	41	+
	42	+	// NewStreamEngine returns an open engine bound to one physical connection
	43	+	// under the package default MuxCeilings.
	44	+	func NewStreamEngine() *StreamEngine {
	45	+	return newStreamEngine(DefaultMuxCeilings())
	46	+	}
	47	+
	48	+	// NewListener returns a listener under the package default handshake budget,
	49	+	// accept-queue bound, and wall clock.
	50	+	func NewListener(pump Pump, accepted ServerPolicyAdmission) (Listener, error) {
	51	+	return newListener(pump, protocol.HandshakeTimeout, MaxAcceptQueue, clock.New(), accepted)
	52	+	}
	53	+
	54	+	// NewScheduler returns an empty outbound scheduler under the package default
	55	+	// MuxCeilings.
	56	+	func NewScheduler(engine StreamEngine) Scheduler {
	57	+	return newScheduler(engine, DefaultMuxCeilings())
	58	+	}
	59	+
	60	+	// Children reports the count of currently owned physical children.
	61	+	func (s *ServerSupervisor) Children() int {
	62	+	if s == nil {
	63	+	return 0
	64	+	}
	65	+	s.mu.Lock()
	66	+	defer s.mu.Unlock()
	67	+	return len(s.children)
	68	+	}
	69	+
	70	+	// DecodePreambleRequestBytes strictly decodes one serialized daemonmux client
	71	+	// preamble under the 4 KiB bound, the strict scan, the generated unmarshal,
	72	+	// and DecodePreambleRequest's semantic checks. Production decodes a preamble
	73	+	// inline through the same steps (see handshake.go); this composed helper
	74	+	// exists for tests and the package fuzz target.
	75	+	func DecodePreambleRequestBytes(payload []byte) (MuxPreambleRequest, error) {
	76	+	if err := CheckPreambleSize(payload); err != nil {
	77	+	return MuxPreambleRequest{}, err
	78	+	}
	79	+	message := &wire.MuxPreambleRequest{}
	80	+	if err := wire.ScanEnvelope(message, payload); err != nil {
	81	+	return MuxPreambleRequest{}, err
	82	+	}
	83	+	if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, message)); err != nil {
	84	+	return MuxPreambleRequest{}, err
	85	+	}
	86	+	return DecodePreambleRequest(message)
	87	+	}

M internal/adapters/daemonmux/listener.go +0 -17


@@ -185,23 +185,6 @@ type admittedStream struct {
185	185		settled bool
186	186		}
187	187
188		-	// NewListener returns a listener over one daemon-side pump under the package
189		-	// policy: the accepted absolute handshake deadline budget is
190		-	// protocol.HandshakeTimeout, and the accept queue is bounded by
191		-	// MaxAcceptQueue. accepted is the provisioned carriage shape the physical
192		-	// handshake admitted: its exact policy and its locality origin. Every admitted
193		-	// session connection is stamped with it. The pump must decode broker-to-daemon
194		-	// frames (Inbound == DirectionClient) and emit daemon-to-broker frames, as the
195		-	// daemon side of a pooled physical connection does; a nil, engine-less, or
196		-	// broker-side pump, or an invalid accepted entry, is refused. The pump must not
197		-	// have started: the listener registers the pump's admission observer, and a
198		-	// pump that already started refuses the registration with
199		-	// ErrAdmissionObserverLate (wrapped in ErrListenerConfig), because an observer
200		-	// registered after Start could miss an already-admitted Open.
201		-	func NewListener(pump Pump, accepted ServerPolicyAdmission) (Listener, error) {
202		-	return newListener(pump, protocol.HandshakeTimeout, MaxAcceptQueue, clock.New(), accepted)
203		-	}
204		-
205	188		// newListener builds a listener from explicit policy. budget is the handshake
206	189		// deadline budget each stream receives at admission, limit is the accept-queue
207	190		// bound, clock supplies the admission time and the deadline timer, and accepted


@@ -185,23 +185,6 @@ type admittedStream struct {
185		settled bool	185	settled bool
186		}	186	}
187			187
188	-	// NewListener returns a listener over one daemon-side pump under the package
189	-	// policy: the accepted absolute handshake deadline budget is
190	-	// protocol.HandshakeTimeout, and the accept queue is bounded by
191	-	// MaxAcceptQueue. accepted is the provisioned carriage shape the physical
192	-	// handshake admitted: its exact policy and its locality origin. Every admitted
193	-	// session connection is stamped with it. The pump must decode broker-to-daemon
194	-	// frames (Inbound == DirectionClient) and emit daemon-to-broker frames, as the
195	-	// daemon side of a pooled physical connection does; a nil, engine-less, or
196	-	// broker-side pump, or an invalid accepted entry, is refused. The pump must not
197	-	// have started: the listener registers the pump's admission observer, and a
198	-	// pump that already started refuses the registration with
199	-	// ErrAdmissionObserverLate (wrapped in ErrListenerConfig), because an observer
200	-	// registered after Start could miss an already-admitted Open.
201	-	func NewListener(pump Pump, accepted ServerPolicyAdmission) (Listener, error) {
202	-	return newListener(pump, protocol.HandshakeTimeout, MaxAcceptQueue, clock.New(), accepted)
203	-	}
204	-
205		// newListener builds a listener from explicit policy. budget is the handshake	188	// newListener builds a listener from explicit policy. budget is the handshake
206		// deadline budget each stream receives at admission, limit is the accept-queue	189	// deadline budget each stream receives at admission, limit is the accept-queue
207		// bound, clock supplies the admission time and the deadline timer, and accepted	190	// bound, clock supplies the admission time and the deadline timer, and accepted

M internal/adapters/daemonmux/preamble.go +0 -23


@@ -328,29 +328,6 @@ func ValidatePreambleResponseAgainstOffer(response MuxPreambleResponse, offered
328	328		return nil
329	329		}
330	330
331		-	// DecodePreambleRequestBytes strictly decodes one serialized daemonmux client
332		-	// preamble. The 4 KiB bound is enforced before allocation, wire.ScanEnvelope
333		-	// is the shared strict scanner that rejects unknown fields, duplicate singular
334		-	// or oneof fields, trailing bytes, truncation, and empty payloads before
335		-	// generated unmarshal, proto.UnmarshalOptions{DiscardUnknown: false} runs next,
336		-	// and DecodePreambleRequest applies the semantic checks. A carriage failure
337		-	// returns the precise wire sentinel (wire.ErrScanUnknown, wire.ErrScanDuplicate,
338		-	// wire.ErrScanTrailing, wire.ErrScanTruncated, ...); an oversize payload or a
339		-	// semantic failure returns ErrPreambleRejected.
340		-	func DecodePreambleRequestBytes(payload []byte) (MuxPreambleRequest, error) {
341		-	if err := CheckPreambleSize(payload); err != nil {
342		-	return MuxPreambleRequest{}, err
343		-	}
344		-	message := &wire.MuxPreambleRequest{}
345		-	if err := wire.ScanEnvelope(message, payload); err != nil {
346		-	return MuxPreambleRequest{}, err
347		-	}
348		-	if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, message)); err != nil {
349		-	return MuxPreambleRequest{}, err
350		-	}
351		-	return DecodePreambleRequest(message)
352		-	}
353		-
354	331		// DecodePreambleResponseBytes strictly decodes one serialized daemonmux server
355	332		// preamble under the same 4 KiB bound, strict scan, generated unmarshal, and
356	333		// semantic decode as DecodePreambleRequestBytes. Carriage failures return the


@@ -328,29 +328,6 @@ func ValidatePreambleResponseAgainstOffer(response MuxPreambleResponse, offered
328		return nil	328	return nil
329		}	329	}
330			330
331	-	// DecodePreambleRequestBytes strictly decodes one serialized daemonmux client
332	-	// preamble. The 4 KiB bound is enforced before allocation, wire.ScanEnvelope
333	-	// is the shared strict scanner that rejects unknown fields, duplicate singular
334	-	// or oneof fields, trailing bytes, truncation, and empty payloads before
335	-	// generated unmarshal, proto.UnmarshalOptions{DiscardUnknown: false} runs next,
336	-	// and DecodePreambleRequest applies the semantic checks. A carriage failure
337	-	// returns the precise wire sentinel (wire.ErrScanUnknown, wire.ErrScanDuplicate,
338	-	// wire.ErrScanTrailing, wire.ErrScanTruncated, ...); an oversize payload or a
339	-	// semantic failure returns ErrPreambleRejected.
340	-	func DecodePreambleRequestBytes(payload []byte) (MuxPreambleRequest, error) {
341	-	if err := CheckPreambleSize(payload); err != nil {
342	-	return MuxPreambleRequest{}, err
343	-	}
344	-	message := &wire.MuxPreambleRequest{}
345	-	if err := wire.ScanEnvelope(message, payload); err != nil {
346	-	return MuxPreambleRequest{}, err
347	-	}
348	-	if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, message)); err != nil {
349	-	return MuxPreambleRequest{}, err
350	-	}
351	-	return DecodePreambleRequest(message)
352	-	}
353	-
354		// DecodePreambleResponseBytes strictly decodes one serialized daemonmux server	331	// DecodePreambleResponseBytes strictly decodes one serialized daemonmux server
355		// preamble under the same 4 KiB bound, strict scan, generated unmarshal, and	332	// preamble under the same 4 KiB bound, strict scan, generated unmarshal, and
356		// semantic decode as DecodePreambleRequestBytes. Carriage failures return the	333	// semantic decode as DecodePreambleRequestBytes. Carriage failures return the

M internal/adapters/daemonmux/scheduler.go +0 -10


@@ -258,16 +258,6 @@ type Scheduler struct {
258	258		quantum int64
259	259		}
260	260
261		-	// NewScheduler returns an empty outbound scheduler under the default
262		-	// MuxCeilings. When engine is non-nil the scheduler enqueues a frame only for
263		-	// an opening or open stream the engine has admitted; when it is nil, the
264		-	// caller owns admission and any nonzero physical identity is accepted. The
265		-	// engine does not need to share the scheduler's ceilings: the scheduler's
266		-	// envelope, chunk, and aggregate budgets come from its own copy.
267		-	func NewScheduler(engine StreamEngine) Scheduler {
268		-	return newScheduler(engine, DefaultMuxCeilings())
269		-	}
270		-
271	261		// NewSchedulerWithCeilings returns an empty outbound scheduler whose envelope,
272	262		// chunk, and aggregate budgets enforce the given negotiated MuxCeilings. The
273	263		// value is validated and copied, so the ceilings a scheduler enforces are


@@ -258,16 +258,6 @@ type Scheduler struct {
258		quantum int64	258	quantum int64
259		}	259	}
260			260
261	-	// NewScheduler returns an empty outbound scheduler under the default
262	-	// MuxCeilings. When engine is non-nil the scheduler enqueues a frame only for
263	-	// an opening or open stream the engine has admitted; when it is nil, the
264	-	// caller owns admission and any nonzero physical identity is accepted. The
265	-	// engine does not need to share the scheduler's ceilings: the scheduler's
266	-	// envelope, chunk, and aggregate budgets come from its own copy.
267	-	func NewScheduler(engine StreamEngine) Scheduler {
268	-	return newScheduler(engine, DefaultMuxCeilings())
269	-	}
270	-
271		// NewSchedulerWithCeilings returns an empty outbound scheduler whose envelope,	261	// NewSchedulerWithCeilings returns an empty outbound scheduler whose envelope,
272		// chunk, and aggregate budgets enforce the given negotiated MuxCeilings. The	262	// chunk, and aggregate budgets enforce the given negotiated MuxCeilings. The
273		// value is validated and copied, so the ceilings a scheduler enforces are	263	// value is validated and copied, so the ceilings a scheduler enforces are

M internal/adapters/daemonmux/supervisor.go +0 -12


@@ -201,18 +201,6 @@ func (s *ServerSupervisor) Adopt(ctx context.Context, raw RawFramedTransport) er
201	201		return nil
202	202		}
203	203
204		-	// Children reports how many physical connections the supervisor currently
205		-	// owns. It is a diagnostic accessor; a lost child disappears once its reaper
206		-	// has closed it.
207		-	func (s *ServerSupervisor) Children() int {
208		-	if s == nil {
209		-	return 0
210		-	}
211		-	s.mu.Lock()
212		-	defer s.mu.Unlock()
213		-	return len(s.children)
214		-	}
215		-
216	204		// Close stops accepting physical carriage and tears every owned child down:
217	205		// it deregisters and closes each child - listener, pump, and carrier together -
218	206		// and joins every reaper. Close is idempotent and concurrent-safe. It leaves


@@ -201,18 +201,6 @@ func (s *ServerSupervisor) Adopt(ctx context.Context, raw RawFramedTransport) er
201		return nil	201	return nil
202		}	202	}
203			203
204	-	// Children reports how many physical connections the supervisor currently
205	-	// owns. It is a diagnostic accessor; a lost child disappears once its reaper
206	-	// has closed it.
207	-	func (s *ServerSupervisor) Children() int {
208	-	if s == nil {
209	-	return 0
210	-	}
211	-	s.mu.Lock()
212	-	defer s.mu.Unlock()
213	-	return len(s.children)
214	-	}
215	-
216		// Close stops accepting physical carriage and tears every owned child down:	204	// Close stops accepting physical carriage and tears every owned child down:
217		// it deregisters and closes each child - listener, pump, and carrier together -	205	// it deregisters and closes each child - listener, pump, and carrier together -
218		// and joins every reaper. Close is idempotent and concurrent-safe. It leaves	206	// and joins every reaper. Close is idempotent and concurrent-safe. It leaves

A internal/adapters/ipc/export_test.go +12 -0

@@ -0,0 +1,12 @@
	1	+	package ipc
	2	+
	3	+	import "path/filepath"
	4	+
	5	+	// SocketPath returns the daemon socket path inside runtimeDir. It has no
	6	+	// current caller: brokeripc.SocketPath (a distinct, unrelated function) is
	7	+	// the one every production and test call site actually uses for the daemon's
	8	+	// own socket path. This is kept as a test-only convenience the package name
	9	+	// implies is available.
	10	+	func SocketPath(runtimeDir string) string {
	11	+	return filepath.Join(runtimeDir, socketFileName)
	12	+	}

@@ -0,0 +1,12 @@
	1	+	package ipc
	2	+
	3	+	import "path/filepath"
	4	+
	5	+	// SocketPath returns the daemon socket path inside runtimeDir. It has no
	6	+	// current caller: brokeripc.SocketPath (a distinct, unrelated function) is
	7	+	// the one every production and test call site actually uses for the daemon's
	8	+	// own socket path. This is kept as a test-only convenience the package name
	9	+	// implies is available.
	10	+	func SocketPath(runtimeDir string) string {
	11	+	return filepath.Join(runtimeDir, socketFileName)
	12	+	}

M internal/adapters/ipc/listener.go +0 -5


@@ -17,11 +17,6 @@ import (
17	17		// its socket directory.
18	18		const socketFileName = "daemon.sock"
19	19
20		-	// SocketPath returns the daemon socket path inside runtimeDir.
21		-	func SocketPath(runtimeDir string) string {
22		-	return filepath.Join(runtimeDir, socketFileName)
23		-	}
24		-
25	20		// ErrDaemonRunning is returned by Listen when the socket path is already
26	21		// bound by a live daemon (a dial-probe against it succeeded).
27	22		var ErrDaemonRunning = errors.New("ipc: a daemon is already listening on this socket")


@@ -17,11 +17,6 @@ import (
17		// its socket directory.	17	// its socket directory.
18		const socketFileName = "daemon.sock"	18	const socketFileName = "daemon.sock"
19			19
20	-	// SocketPath returns the daemon socket path inside runtimeDir.
21	-	func SocketPath(runtimeDir string) string {
22	-	return filepath.Join(runtimeDir, socketFileName)
23	-	}
24	-
25		// ErrDaemonRunning is returned by Listen when the socket path is already	20	// ErrDaemonRunning is returned by Listen when the socket path is already
26		// bound by a live daemon (a dial-probe against it succeeded).	21	// bound by a live daemon (a dial-probe against it succeeded).
27		var ErrDaemonRunning = errors.New("ipc: a daemon is already listening on this socket")	22	var ErrDaemonRunning = errors.New("ipc: a daemon is already listening on this socket")

M internal/adapters/quic/endpoint.go +0 -6


@@ -47,12 +47,6 @@ type dialOptions struct {
47	47		observer ports.SerializedRuntimeObserver
48	48		}
49	49
50		-	// WithRuntimeObserver enables process-local adapter marks on the dialed
51		-	// transport. A nil observer leaves observation disabled.
52		-	func WithRuntimeObserver(observer ports.SerializedRuntimeObserver) Option {
53		-	return func(opts *dialOptions) { opts.observer = observer }
54		-	}
55		-
56	50		// DialConfig carries the dial parameters. Fingerprint is the exact
57	51		// SHA-256 of the server certificate (32 bytes). Timeout bounds the whole
58	52		// dial: address resolution, every candidate connection, stream open, and


@@ -47,12 +47,6 @@ type dialOptions struct {
47		observer ports.SerializedRuntimeObserver	47	observer ports.SerializedRuntimeObserver
48		}	48	}
49			49
50	-	// WithRuntimeObserver enables process-local adapter marks on the dialed
51	-	// transport. A nil observer leaves observation disabled.
52	-	func WithRuntimeObserver(observer ports.SerializedRuntimeObserver) Option {
53	-	return func(opts *dialOptions) { opts.observer = observer }
54	-	}
55	-
56		// DialConfig carries the dial parameters. Fingerprint is the exact	50	// DialConfig carries the dial parameters. Fingerprint is the exact
57		// SHA-256 of the server certificate (32 bytes). Timeout bounds the whole	51	// SHA-256 of the server certificate (32 bytes). Timeout bounds the whole
58		// dial: address resolution, every candidate connection, stream open, and	52	// dial: address resolution, every candidate connection, stream open, and

A internal/adapters/quic/export_test.go +10 -0

@@ -0,0 +1,10 @@
	1	+	package quic
	2	+
	3	+	import "github.com/bnema/vev/internal/ports"
	4	+
	5	+	// WithRuntimeObserver enables process-local adapter marks on the dialed
	6	+	// transport, for tests that assert on those marks. Production never needs
	7	+	// runtime observation on a QUIC dial.
	8	+	func WithRuntimeObserver(observer ports.SerializedRuntimeObserver) Option {
	9	+	return func(opts *dialOptions) { opts.observer = observer }
	10	+	}

@@ -0,0 +1,10 @@
	1	+	package quic
	2	+
	3	+	import "github.com/bnema/vev/internal/ports"
	4	+
	5	+	// WithRuntimeObserver enables process-local adapter marks on the dialed
	6	+	// transport, for tests that assert on those marks. Production never needs
	7	+	// runtime observation on a QUIC dial.
	8	+	func WithRuntimeObserver(observer ports.SerializedRuntimeObserver) Option {
	9	+	return func(opts *dialOptions) { opts.observer = observer }
	10	+	}

A internal/usecase/broker/export_test.go +67 -0

@@ -0,0 +1,67 @@
	1	+	package broker
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"fmt"
	6	+	"log/slog"
	7	+
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	)
	10	+
	11	+	// NewRegistry restores a validated durable snapshot under a fresh broker
	12	+	// epoch with the default RegistryConfig, delegating to the same
	13	+	// NewRegistryWithConfig production composition uses. It is a test-only
	14	+	// shorthand: production always selects an explicit membership/observation
	15	+	// mode.
	16	+	func NewRegistry(epoch ports.BrokerEpoch, store ports.BrokerHostStore, probe ports.BrokerHostProbe, clock ports.Clock, log slog.Logger) (Registry, error) {
	17	+	return NewRegistryWithConfig(epoch, store, probe, clock, log, RegistryConfig{})
	18	+	}
	19	+
	20	+	// setHosts is projection-only; it is private so callers cannot bypass durable
	21	+	// membership and policy. Observation tests use it without claiming durability.
	22	+	// An unchanged membership is a no-op: it neither republishes nor restarts an
	23	+	// in-flight observation. An exhausted revision series is refused with
	24	+	// ports.ErrBrokerRevisionExhausted before the projection changes.
	25	+	// Removed registrations retire bounded tombstones so a stale completion can
	26	+	// never revive them; replaced or re-added endpoints are fenced by the
	27	+	// registry attempt token instead.
	28	+	func (r *Registry) setHosts(records []ports.BrokerHostRecord) error {
	29	+	if len(records) > ports.BrokerMaxHosts {
	30	+	return fmt.Errorf("broker: %d host registrations exceed the limit of %d", len(records), ports.BrokerMaxHosts)
	31	+	}
	32	+	if len(records) > 0 && !r.observationDisabled && nilDependency(r.probe) {
	33	+	return errors.New("broker: remote hosts require a host probe")
	34	+	}
	35	+	next := make(map[string]ports.BrokerDaemonObservation, len(records))
	36	+	order := make([]string, 0, len(records))
	37	+	for rank, record := range records {
	38	+	if err := record.Registration.Validate(); err != nil {
	39	+	return err
	40	+	}
	41	+	if err := record.Policy.Validate(); err != nil {
	42	+	return err
	43	+	}
	44	+	registration := record.Registration
	45	+	if _, duplicate := next[registration.Endpoint]; duplicate {
	46	+	return errors.New("broker: duplicate host registration")
	47	+	}
	48	+	observation := stampHostAuthority(ports.BrokerDaemonObservation{}, record, rank)
	49	+	// Membership the registry cannot publish is a caller error, refused here
	50	+	// rather than left to freeze every later publication: an endpoint whose
	51	+	// derived display hint is empty after the display rule drops its runes
	52	+	// makes every snapshot carrying it invalid.
	53	+	if err := observation.Validate(); err != nil {
	54	+	return fmt.Errorf("broker: host record %q is not publishable: %w", registration.Endpoint, err)
	55	+	}
	56	+	next[registration.Endpoint] = observation
	57	+	order = append(order, registration.Endpoint)
	58	+	}
	59	+
	60	+	r.mu.Lock()
	61	+	defer r.mu.Unlock()
	62	+	if r.revisionExhaustedLocked() {
	63	+	return fmt.Errorf("broker: set hosts: %w", ports.ErrBrokerRevisionExhausted)
	64	+	}
	65	+	r.setHostsLocked(next, order)
	66	+	return nil
	67	+	}

@@ -0,0 +1,67 @@
	1	+	package broker
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"fmt"
	6	+	"log/slog"
	7	+
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	)
	10	+
	11	+	// NewRegistry restores a validated durable snapshot under a fresh broker
	12	+	// epoch with the default RegistryConfig, delegating to the same
	13	+	// NewRegistryWithConfig production composition uses. It is a test-only
	14	+	// shorthand: production always selects an explicit membership/observation
	15	+	// mode.
	16	+	func NewRegistry(epoch ports.BrokerEpoch, store ports.BrokerHostStore, probe ports.BrokerHostProbe, clock ports.Clock, log slog.Logger) (Registry, error) {
	17	+	return NewRegistryWithConfig(epoch, store, probe, clock, log, RegistryConfig{})
	18	+	}
	19	+
	20	+	// setHosts is projection-only; it is private so callers cannot bypass durable
	21	+	// membership and policy. Observation tests use it without claiming durability.
	22	+	// An unchanged membership is a no-op: it neither republishes nor restarts an
	23	+	// in-flight observation. An exhausted revision series is refused with
	24	+	// ports.ErrBrokerRevisionExhausted before the projection changes.
	25	+	// Removed registrations retire bounded tombstones so a stale completion can
	26	+	// never revive them; replaced or re-added endpoints are fenced by the
	27	+	// registry attempt token instead.
	28	+	func (r *Registry) setHosts(records []ports.BrokerHostRecord) error {
	29	+	if len(records) > ports.BrokerMaxHosts {
	30	+	return fmt.Errorf("broker: %d host registrations exceed the limit of %d", len(records), ports.BrokerMaxHosts)
	31	+	}
	32	+	if len(records) > 0 && !r.observationDisabled && nilDependency(r.probe) {
	33	+	return errors.New("broker: remote hosts require a host probe")
	34	+	}
	35	+	next := make(map[string]ports.BrokerDaemonObservation, len(records))
	36	+	order := make([]string, 0, len(records))
	37	+	for rank, record := range records {
	38	+	if err := record.Registration.Validate(); err != nil {
	39	+	return err
	40	+	}
	41	+	if err := record.Policy.Validate(); err != nil {
	42	+	return err
	43	+	}
	44	+	registration := record.Registration
	45	+	if _, duplicate := next[registration.Endpoint]; duplicate {
	46	+	return errors.New("broker: duplicate host registration")
	47	+	}
	48	+	observation := stampHostAuthority(ports.BrokerDaemonObservation{}, record, rank)
	49	+	// Membership the registry cannot publish is a caller error, refused here
	50	+	// rather than left to freeze every later publication: an endpoint whose
	51	+	// derived display hint is empty after the display rule drops its runes
	52	+	// makes every snapshot carrying it invalid.
	53	+	if err := observation.Validate(); err != nil {
	54	+	return fmt.Errorf("broker: host record %q is not publishable: %w", registration.Endpoint, err)
	55	+	}
	56	+	next[registration.Endpoint] = observation
	57	+	order = append(order, registration.Endpoint)
	58	+	}
	59	+
	60	+	r.mu.Lock()
	61	+	defer r.mu.Unlock()
	62	+	if r.revisionExhaustedLocked() {
	63	+	return fmt.Errorf("broker: set hosts: %w", ports.ErrBrokerRevisionExhausted)
	64	+	}
	65	+	r.setHostsLocked(next, order)
	66	+	return nil
	67	+	}

M internal/usecase/broker/registry.go +0 -62


@@ -172,19 +172,6 @@ type Registry struct {
172	172		demand int
173	173		}
174	174
175		-	// NewRegistry restores a validated durable snapshot under a fresh broker epoch.
176		-	// The store owns durable membership as well as the advisory snapshot, so
177		-	// BrokerHostStore is required explicitly: a store that cannot answer
178		-	// LoadHosts is not a broker host store, and an optional assertion would let a
179		-	// caller silently run the registry without authoritative membership. A nil
180		-	// store is refused by the dependency guard, including a typed nil hidden
181		-	// behind the interface. Until Run returns, callers must not mutate the store
182		-	// outside Registry.ReplaceHosts. The caller closes the store after the registry
183		-	// has stopped and flushed its advisory writer.
184		-	func NewRegistry(epoch ports.BrokerEpoch, store ports.BrokerHostStore, probe ports.BrokerHostProbe, clock ports.Clock, log slog.Logger) (Registry, error) {
185		-	return NewRegistryWithConfig(epoch, store, probe, clock, log, RegistryConfig{})
186		-	}
187		-
188	175		// NewRegistryWithConfig restores a validated durable snapshot under a fresh
189	176		// broker epoch and selects the run mode from cfg. Observation-disabled mode
190	177		// tolerates a nil probe because such a registry never probes, and a registry

@@ -759,55 +746,6 @@ func (r *Registry) ReplaceHosts(records []ports.BrokerHostRecord) error {
759	746		return r.commitMembershipLocked(records)
760	747		}
761	748
762		-	// setHosts is projection-only; it is private so callers cannot bypass durable
763		-	// membership and policy. Observation tests use it without claiming durability.
764		-	// An unchanged membership is a no-op: it neither republishes nor restarts an
765		-	// in-flight observation. An exhausted revision series is refused with
766		-	// ports.ErrBrokerRevisionExhausted before the projection changes.
767		-	// Removed registrations retire bounded tombstones so a stale completion can
768		-	// never revive them; replaced or re-added endpoints are fenced by the
769		-	// registry attempt token instead.
770		-	func (r *Registry) setHosts(records []ports.BrokerHostRecord) error {
771		-	if len(records) > ports.BrokerMaxHosts {
772		-	return fmt.Errorf("broker: %d host registrations exceed the limit of %d", len(records), ports.BrokerMaxHosts)
773		-	}
774		-	if len(records) > 0 && !r.observationDisabled && nilDependency(r.probe) {
775		-	return errors.New("broker: remote hosts require a host probe")
776		-	}
777		-	next := make(map[string]ports.BrokerDaemonObservation, len(records))
778		-	order := make([]string, 0, len(records))
779		-	for rank, record := range records {
780		-	if err := record.Registration.Validate(); err != nil {
781		-	return err
782		-	}
783		-	if err := record.Policy.Validate(); err != nil {
784		-	return err
785		-	}
786		-	registration := record.Registration
787		-	if _, duplicate := next[registration.Endpoint]; duplicate {
788		-	return errors.New("broker: duplicate host registration")
789		-	}
790		-	observation := stampHostAuthority(ports.BrokerDaemonObservation{}, record, rank)
791		-	// Membership the registry cannot publish is a caller error, refused here
792		-	// rather than left to freeze every later publication: an endpoint whose
793		-	// derived display hint is empty after the display rule drops its runes
794		-	// makes every snapshot carrying it invalid.
795		-	if err := observation.Validate(); err != nil {
796		-	return fmt.Errorf("broker: host record %q is not publishable: %w", registration.Endpoint, err)
797		-	}
798		-	next[registration.Endpoint] = observation
799		-	order = append(order, registration.Endpoint)
800		-	}
801		-
802		-	r.mu.Lock()
803		-	defer r.mu.Unlock()
804		-	if r.revisionExhaustedLocked() {
805		-	return fmt.Errorf("broker: set hosts: %w", ports.ErrBrokerRevisionExhausted)
806		-	}
807		-	r.setHostsLocked(next, order)
808		-	return nil
809		-	}
810		-
811	749		// setHostsLocked applies only a projection; the public path commits authority
812	750		// first. The caller holds mu through both steps, fencing probe completions.
813	751		// order is the registration order the projection was built from; it becomes


@@ -172,19 +172,6 @@ type Registry struct {
172		demand int	172	demand int
173		}	173	}
174			174
175	-	// NewRegistry restores a validated durable snapshot under a fresh broker epoch.
176	-	// The store owns durable membership as well as the advisory snapshot, so
177	-	// BrokerHostStore is required explicitly: a store that cannot answer
178	-	// LoadHosts is not a broker host store, and an optional assertion would let a
179	-	// caller silently run the registry without authoritative membership. A nil
180	-	// store is refused by the dependency guard, including a typed nil hidden
181	-	// behind the interface. Until Run returns, callers must not mutate the store
182	-	// outside Registry.ReplaceHosts. The caller closes the store after the registry
183	-	// has stopped and flushed its advisory writer.
184	-	func NewRegistry(epoch ports.BrokerEpoch, store ports.BrokerHostStore, probe ports.BrokerHostProbe, clock ports.Clock, log slog.Logger) (Registry, error) {
185	-	return NewRegistryWithConfig(epoch, store, probe, clock, log, RegistryConfig{})
186	-	}
187	-
188		// NewRegistryWithConfig restores a validated durable snapshot under a fresh	175	// NewRegistryWithConfig restores a validated durable snapshot under a fresh
189		// broker epoch and selects the run mode from cfg. Observation-disabled mode	176	// broker epoch and selects the run mode from cfg. Observation-disabled mode
190		// tolerates a nil probe because such a registry never probes, and a registry	177	// tolerates a nil probe because such a registry never probes, and a registry

@@ -759,55 +746,6 @@ func (r *Registry) ReplaceHosts(records []ports.BrokerHostRecord) error {
759		return r.commitMembershipLocked(records)	746	return r.commitMembershipLocked(records)
760		}	747	}
761			748
762	-	// setHosts is projection-only; it is private so callers cannot bypass durable
763	-	// membership and policy. Observation tests use it without claiming durability.
764	-	// An unchanged membership is a no-op: it neither republishes nor restarts an
765	-	// in-flight observation. An exhausted revision series is refused with
766	-	// ports.ErrBrokerRevisionExhausted before the projection changes.
767	-	// Removed registrations retire bounded tombstones so a stale completion can
768	-	// never revive them; replaced or re-added endpoints are fenced by the
769	-	// registry attempt token instead.
770	-	func (r *Registry) setHosts(records []ports.BrokerHostRecord) error {
771	-	if len(records) > ports.BrokerMaxHosts {
772	-	return fmt.Errorf("broker: %d host registrations exceed the limit of %d", len(records), ports.BrokerMaxHosts)
773	-	}
774	-	if len(records) > 0 && !r.observationDisabled && nilDependency(r.probe) {
775	-	return errors.New("broker: remote hosts require a host probe")
776	-	}
777	-	next := make(map[string]ports.BrokerDaemonObservation, len(records))
778	-	order := make([]string, 0, len(records))
779	-	for rank, record := range records {
780	-	if err := record.Registration.Validate(); err != nil {
781	-	return err
782	-	}
783	-	if err := record.Policy.Validate(); err != nil {
784	-	return err
785	-	}
786	-	registration := record.Registration
787	-	if _, duplicate := next[registration.Endpoint]; duplicate {
788	-	return errors.New("broker: duplicate host registration")
789	-	}
790	-	observation := stampHostAuthority(ports.BrokerDaemonObservation{}, record, rank)
791	-	// Membership the registry cannot publish is a caller error, refused here
792	-	// rather than left to freeze every later publication: an endpoint whose
793	-	// derived display hint is empty after the display rule drops its runes
794	-	// makes every snapshot carrying it invalid.
795	-	if err := observation.Validate(); err != nil {
796	-	return fmt.Errorf("broker: host record %q is not publishable: %w", registration.Endpoint, err)
797	-	}
798	-	next[registration.Endpoint] = observation
799	-	order = append(order, registration.Endpoint)
800	-	}
801	-
802	-	r.mu.Lock()
803	-	defer r.mu.Unlock()
804	-	if r.revisionExhaustedLocked() {
805	-	return fmt.Errorf("broker: set hosts: %w", ports.ErrBrokerRevisionExhausted)
806	-	}
807	-	r.setHostsLocked(next, order)
808	-	return nil
809	-	}
810	-
811		// setHostsLocked applies only a projection; the public path commits authority	749	// setHostsLocked applies only a projection; the public path commits authority
812		// first. The caller holds mu through both steps, fencing probe completions.	750	// first. The caller holds mu through both steps, fencing probe completions.
813		// order is the registration order the projection was built from; it becomes	751	// order is the registration order the projection was built from; it becomes