fix(broker): observe daemons over pooled physical transports

Local and remote observation probes dialed a fresh physical transport on
every probe, so demand-aware re-probing (1s local, 2s remote) and the
local re-probe after a stream completes opened extra daemonmux carriages
next to the pooled one, and every remote probe paid a full SSH/QUIC
bootstrap. Probes now borrow the pool's published transport for the
authenticated identity and exact policy (Pool.SharedPhysical) and dial
only when nothing is pooled or the borrowed transport retired mid-probe.
Borrowing never reserves the entry or refreshes its idle deadline.

Restores TestOfflineBrokerSharedTransportKeepsAttachmentsIndependent,
TestWebGatewayTabsShareOneBrokerWithIndependentStreamsAndServices and
TestWebGatewayTabClosingIsIsolated.

commit: 818dc4840f4532a22482c12d83df0adbf1a9bbd5
author: brice · 30d ago · 2026-09-22 20:33 UTC
parent: 9e386a7
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +205 -4

M internal/app/broker_offline.go +9 -0


@@ -433,6 +433,15 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
433	433		if err != nil {
434	434		return err
435	435		}
	436	+	// Observation borrows pooled transports rather than dialing its own.
	437	+	if concreteRemoteProbe != nil {
	438	+	concreteRemoteProbe.shared.share(pool)
	439	+	}
	440	+	if registryConfig.Local != nil {
	441	+	if local, ok := registryConfig.Local.Probe.(*localRouteProbe); ok {
	442	+	local.shared.share(pool)
	443	+	}
	444	+	}
436	445		if err := supervisor.RegisterCloseable("pool", pool); err != nil {
437	446		return err
438	447		}


@@ -433,6 +433,15 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
433	if err != nil {	433		if err != nil {
434	return err	434		return err
435	}	435		}
		436	+	// Observation borrows pooled transports rather than dialing its own.
		437	+	if concreteRemoteProbe != nil {
		438	+	concreteRemoteProbe.shared.share(pool)
		439	+	}
		440	+	if registryConfig.Local != nil {
		441	+	if local, ok := registryConfig.Local.Probe.(*localRouteProbe); ok {
		442	+	local.shared.share(pool)
		443	+	}
		444	+	}
436	if err := supervisor.RegisterCloseable("pool", pool); err != nil {	445		if err := supervisor.RegisterCloseable("pool", pool); err != nil {
437	return err	446		return err
438	}	447		}

M internal/app/broker_offline_local.go +11 -4


@@ -59,6 +59,7 @@ type localRouteProbe struct {
59	59		policy ports.BrokerPolicy
60	60		loadIdentity localIdentityLoader
61	61		connector ports.BrokerEndpointConnector
	62	+	shared pooledPhysicals
62	63		}
63	64
64	65		var _ broker.LocalProbe = (*localRouteProbe)(nil)

@@ -152,6 +153,16 @@ func (p *localRouteProbe) ProbeLocal(ctx context.Context) (ports.BrokerDaemonObs
152	153		if err := endpoint.Validate(); err != nil {
153	154		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, fmt.Errorf("vev: broker local probe: %w", err)
154	155		}
	156	+	request, err := p.request()
	157	+	if err != nil {
	158	+	return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err
	159	+	}
	160	+	if observation, handled, err := p.shared.observe(ctx, identity, p.policy, request); handled {
	161	+	if err != nil {
	162	+	return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err
	163	+	}
	164	+	return observation, nil
	165	+	}
155	166		physical, err := p.connector.Connect(ctx, endpoint)
156	167		if err != nil {
157	168		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err

@@ -163,10 +174,6 @@ func (p *localRouteProbe) ProbeLocal(ctx context.Context) (ports.BrokerDaemonObs
163	174		if physical.Identity() != endpoint.ExpectedIdentity.Identity \|\| !physical.Policy().Compatible(endpoint.Policy) {
164	175		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, errors.New("vev: broker local probe authenticated binding mismatch")
165	176		}
166		-	request, err := p.request()
167		-	if err != nil {
168		-	return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err
169		-	}
170	177		observation, err := observeDaemonCatalogue(ctx, physical, request)
171	178		if err != nil {
172	179		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err

 	policy       ports.BrokerPolicy
 	loadIdentity localIdentityLoader
 	connector    ports.BrokerEndpointConnector
+	shared       pooledPhysicals
 }
 var _ broker.LocalProbe = (*localRouteProbe)(nil)
 	if err := endpoint.Validate(); err != nil {
 		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, fmt.Errorf("vev: broker local probe: %w", err)
 	}
+	request, err := p.request()
+	if err != nil {
+		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err
+	}
+	if observation, handled, err := p.shared.observe(ctx, identity, p.policy, request); handled {
+		if err != nil {
+			return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err
+		}
+		return observation, nil
+	}
 	physical, err := p.connector.Connect(ctx, endpoint)
 	if err != nil {
 		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err
 	if physical.Identity() != endpoint.ExpectedIdentity.Identity || !physical.Policy().Compatible(endpoint.Policy) {
 		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, errors.New("vev: broker local probe authenticated binding mismatch")
 	}
-	request, err := p.request()
-	if err != nil {
-		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err
-	}
 	observation, err := observeDaemonCatalogue(ctx, physical, request)
 	if err != nil {
 		return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityUnreachable}, err

M internal/app/broker_remote_probe.go +47 -0


@@ -7,6 +7,7 @@ import (
7	7		"encoding/json"
8	8		"errors"
9	9		"fmt"
	10	+	"sync/atomic"
10	11
11	12		"github.com/bnema/vev/internal/domain"
12	13		"github.com/bnema/vev/internal/ports"

@@ -34,6 +35,44 @@ type brokerRemoteProbe struct {
34	35		binder ports.BrokerIdentityBinder
35	36		connector ports.BrokerEndpointConnector
36	37		hosts ports.BrokerHostAuthorityReader
	38	+	shared pooledPhysicals
	39	+	}
	40	+
	41	+	// sharedPhysicals lends the pooled transport of an attached or warm daemon.
	42	+	type sharedPhysicals interface {
	43	+	SharedPhysical(ports.BrokerDaemonIdentity, ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool)
	44	+	}
	45	+
	46	+	// pooledPhysicals lets an observation probe borrow the broker pool's physical
	47	+	// transport instead of dialing (and, for a remote, bootstrapping SSH/QUIC)
	48	+	// again: the broker exists to share connections. The pool is published after
	49	+	// composition because it binds identities through the registry that runs the
	50	+	// probes. A probe dials only when nothing is pooled, or when the borrowed
	51	+	// transport retired under it.
	52	+	type pooledPhysicals struct{ pool atomic.Value }
	53	+
	54	+	func (s *pooledPhysicals) share(pool sharedPhysicals) { s.pool.Store(pool) }
	55	+
	56	+	// observe reports handled=false when the probe must dial its own transport.
	57	+	func (s *pooledPhysicals) observe(ctx context.Context, identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy, request ports.BrokerOpenStreamRequest) (ports.BrokerDaemonObservation, bool, error) {
	58	+	pool, _ := s.pool.Load().(sharedPhysicals)
	59	+	if pool == nil \|\| identity == "" {
	60	+	return ports.BrokerDaemonObservation{}, false, nil
	61	+	}
	62	+	physical, ok := pool.SharedPhysical(identity, policy)
	63	+	if !ok {
	64	+	return ports.BrokerDaemonObservation{}, false, nil
	65	+	}
	66	+	observation, err := observeDaemonCatalogue(ctx, physical, request)
	67	+	if err == nil \|\| ctx.Err() != nil {
	68	+	return observation, true, err
	69	+	}
	70	+	select {
	71	+	case <-physical.Done():
	72	+	return ports.BrokerDaemonObservation{}, false, nil
	73	+	default:
	74	+	return observation, true, err
	75	+	}
37	76		}
38	77
39	78		var _ ports.BrokerHostProbe = (*brokerRemoteProbe)(nil)

@@ -50,6 +89,14 @@ func (p *brokerRemoteProbe) Probe(ctx context.Context, registration domain.Remot
50	89		if err != nil {
51	90		return ports.BrokerDaemonObservation{}, err
52	91		}
	92	+	// Borrow the pooled transport of an attached or warm daemon instead of
	93	+	// bootstrapping SSH/QUIC again. A borrowed transport that retires mid-probe
	94	+	// falls back to a fresh dial.
	95	+	if endpoint.ExpectedIdentity.Bound {
	96	+	if observation, handled, err := p.shared.observe(ctx, endpoint.ExpectedIdentity.Identity, endpoint.Policy, request); handled {
	97	+	return observation, err
	98	+	}
	99	+	}
53	100		physical, err := p.connector.Connect(ctx, endpoint)
54	101		if err != nil {
55	102		return ports.BrokerDaemonObservation{}, err


@@ -7,6 +7,7 @@ import (
7	"encoding/json"	7		"encoding/json"
8	"errors"	8		"errors"
9	"fmt"	9		"fmt"
		10	+	"sync/atomic"
10		11
11	"github.com/bnema/vev/internal/domain"	12		"github.com/bnema/vev/internal/domain"
12	"github.com/bnema/vev/internal/ports"	13		"github.com/bnema/vev/internal/ports"

@@ -34,6 +35,44 @@ type brokerRemoteProbe struct {
34	binder ports.BrokerIdentityBinder	35		binder ports.BrokerIdentityBinder
35	connector ports.BrokerEndpointConnector	36		connector ports.BrokerEndpointConnector
36	hosts ports.BrokerHostAuthorityReader	37		hosts ports.BrokerHostAuthorityReader
		38	+	shared pooledPhysicals
		39	+	}
		40	+
		41	+	// sharedPhysicals lends the pooled transport of an attached or warm daemon.
		42	+	type sharedPhysicals interface {
		43	+	SharedPhysical(ports.BrokerDaemonIdentity, ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool)
		44	+	}
		45	+
		46	+	// pooledPhysicals lets an observation probe borrow the broker pool's physical
		47	+	// transport instead of dialing (and, for a remote, bootstrapping SSH/QUIC)
		48	+	// again: the broker exists to share connections. The pool is published after
		49	+	// composition because it binds identities through the registry that runs the
		50	+	// probes. A probe dials only when nothing is pooled, or when the borrowed
		51	+	// transport retired under it.
		52	+	type pooledPhysicals struct{ pool atomic.Value }
		53	+
		54	+	func (s *pooledPhysicals) share(pool sharedPhysicals) { s.pool.Store(pool) }
		55	+
		56	+	// observe reports handled=false when the probe must dial its own transport.
		57	+	func (s *pooledPhysicals) observe(ctx context.Context, identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy, request ports.BrokerOpenStreamRequest) (ports.BrokerDaemonObservation, bool, error) {
		58	+	pool, _ := s.pool.Load().(sharedPhysicals)
		59	+	if pool == nil \|\| identity == "" {
		60	+	return ports.BrokerDaemonObservation{}, false, nil
		61	+	}
		62	+	physical, ok := pool.SharedPhysical(identity, policy)
		63	+	if !ok {
		64	+	return ports.BrokerDaemonObservation{}, false, nil
		65	+	}
		66	+	observation, err := observeDaemonCatalogue(ctx, physical, request)
		67	+	if err == nil \|\| ctx.Err() != nil {
		68	+	return observation, true, err
		69	+	}
		70	+	select {
		71	+	case <-physical.Done():
		72	+	return ports.BrokerDaemonObservation{}, false, nil
		73	+	default:
		74	+	return observation, true, err
		75	+	}
37	}	76		}
38		77
39	var _ ports.BrokerHostProbe = (*brokerRemoteProbe)(nil)	78		var _ ports.BrokerHostProbe = (*brokerRemoteProbe)(nil)

@@ -50,6 +89,14 @@ func (p *brokerRemoteProbe) Probe(ctx context.Context, registration domain.Remot
50	if err != nil {	89		if err != nil {
51	return ports.BrokerDaemonObservation{}, err	90		return ports.BrokerDaemonObservation{}, err
52	}	91		}
		92	+	// Borrow the pooled transport of an attached or warm daemon instead of
		93	+	// bootstrapping SSH/QUIC again. A borrowed transport that retires mid-probe
		94	+	// falls back to a fresh dial.
		95	+	if endpoint.ExpectedIdentity.Bound {
		96	+	if observation, handled, err := p.shared.observe(ctx, endpoint.ExpectedIdentity.Identity, endpoint.Policy, request); handled {
		97	+	return observation, err
		98	+	}
		99	+	}
53	physical, err := p.connector.Connect(ctx, endpoint)	100		physical, err := p.connector.Connect(ctx, endpoint)
54	if err != nil {	101		if err != nil {
55	return ports.BrokerDaemonObservation{}, err	102		return ports.BrokerDaemonObservation{}, err

A internal/app/broker_remote_probe_shared_test.go +110 -0

@@ -0,0 +1,110 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"testing"
	7	+
	8	+	"github.com/bnema/vev/internal/domain"
	9	+	"github.com/bnema/vev/internal/ports"
	10	+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
	11	+	"github.com/bnema/vev/internal/protocol"
	12	+	"github.com/stretchr/testify/mock"
	13	+	"github.com/stretchr/testify/require"
	14	+	)
	15	+
	16	+	// sharedPhysicalsFunc adapts a closure to the probe's pool lending seam.
	17	+	type sharedPhysicalsFunc func(ports.BrokerDaemonIdentity, ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool)
	18	+
	19	+	func (f sharedPhysicalsFunc) SharedPhysical(identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
	20	+	return f(identity, policy)
	21	+	}
	22	+
	23	+	// TestBrokerRemoteProbeBorrowsWarmTransport proves observation reuses the
	24	+	// broker pool's attached or warm transport instead of bootstrapping SSH/QUIC
	25	+	// again, and redials only when the borrowed transport has retired.
	26	+	func TestBrokerRemoteProbeBorrowsWarmTransport(t *testing.T) {
	27	+	registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
	28	+	policy := remoteBrokerPolicy("quic")
	29	+	const identity ports.BrokerDaemonIdentity = "remote-daemon"
	30	+	dialErr := errors.New("dial refused")
	31	+	tests := []struct {
	32	+	name string
	33	+	bound bool
	34	+	shared bool
	35	+	openErr error
	36	+	retired bool
	37	+	wantDials int
	38	+	wantErr error
	39	+	}{
	40	+	{name: "warm transport observes without dialing", bound: true, shared: true},
	41	+	{name: "unbound identity dials", bound: false, shared: true, wantDials: 1, wantErr: dialErr},
	42	+	{name: "no pooled transport dials", bound: true, wantDials: 1, wantErr: dialErr},
	43	+	{name: "live transport refusal is reported without redial", bound: true, shared: true, openErr: errors.New("stream refused")},
	44	+	{name: "retired transport redials", bound: true, shared: true, openErr: errors.New("physical closed"), retired: true, wantDials: 1, wantErr: dialErr},
	45	+	}
	46	+	for _, tt := range tests {
	47	+	t.Run(tt.name, func(t *testing.T) {
	48	+	record := ports.BrokerHostRecord{Registration: registration, Policy: policy}
	49	+	if tt.bound {
	50	+	record.Identity = identity
	51	+	}
	52	+	routes := portsmocks.NewMockBrokerRouteAuthority(t)
	53	+	routes.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).Return(ports.BrokerDialTarget{
	54	+	Fence: ports.BrokerEndpointFence{Registration: registration}, Policy: policy, Address: "opaque", StartMode: ports.BrokerDaemonExistingOnly,
	55	+	ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: record.Identity, Bound: tt.bound},
	56	+	}, nil)
	57	+	connector := portsmocks.NewMockBrokerEndpointConnector(t)
	58	+	if tt.wantDials > 0 {
	59	+	connector.EXPECT().Connect(mock.Anything, mock.Anything).Return(nil, dialErr).Times(tt.wantDials)
	60	+	}
	61	+	physical := portsmocks.NewMockBrokerPhysicalConnection(t)
	62	+	done := make(chan struct{})
	63	+	if tt.retired {
	64	+	close(done)
	65	+	}
	66	+	if tt.bound && tt.shared {
	67	+	if tt.openErr != nil {
	68	+	physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(nil, tt.openErr)
	69	+	physical.EXPECT().Done().Return(done)
	70	+	} else {
	71	+	logical := portsmocks.NewMockBrokerLogicalConnection(t)
	72	+	var sent protocol.CommandRequest
	73	+	logical.EXPECT().SendClient(mock.Anything).RunAndReturn(func(message protocol.ClientMessage) error {
	74	+	sent = message.(protocol.CommandRequest)
	75	+	return nil
	76	+	})
	77	+	logical.EXPECT().ReceiveServer().RunAndReturn(func() (protocol.ServerMessage, error) {
	78	+	return protocol.CommandResult{RequestID: sent.RequestID, Outcome: protocol.CommandSucceeded, Output: testLocalCatalogue(t)}, nil
	79	+	})
	80	+	logical.EXPECT().Close().Return(nil)
	81	+	physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(logical, nil)
	82	+	physical.EXPECT().Identity().Return(identity)
	83	+	physical.EXPECT().Incarnation().Return(ports.BrokerDaemonIncarnation{1})
	84	+	}
	85	+	}
	86	+	probe := &brokerRemoteProbe{
	87	+	epoch: 7, routes: routes, binder: portsmocks.NewMockBrokerIdentityBinder(t), connector: connector,
	88	+	hosts: &routeTestHosts{record: record, found: true},
	89	+	}
	90	+	probe.shared.share(sharedPhysicalsFunc(func(gotIdentity ports.BrokerDaemonIdentity, gotPolicy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
	91	+	if !tt.shared \|\| gotIdentity != identity \|\| gotPolicy != policy {
	92	+	return nil, false
	93	+	}
	94	+	return physical, true
	95	+	}))
	96	+
	97	+	observation, err := probe.Probe(context.Background(), registration)
	98	+	switch {
	99	+	case tt.wantErr != nil:
	100	+	require.ErrorIs(t, err, tt.wantErr)
	101	+	case tt.openErr != nil:
	102	+	require.ErrorIs(t, err, tt.openErr)
	103	+	default:
	104	+	require.NoError(t, err)
	105	+	require.Equal(t, identity, observation.Identity)
	106	+	require.True(t, observation.InventoryKnown)
	107	+	}
	108	+	})
	109	+	}
	110	+	}

@@ -0,0 +1,110 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"testing"
	7	+
	8	+	"github.com/bnema/vev/internal/domain"
	9	+	"github.com/bnema/vev/internal/ports"
	10	+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
	11	+	"github.com/bnema/vev/internal/protocol"
	12	+	"github.com/stretchr/testify/mock"
	13	+	"github.com/stretchr/testify/require"
	14	+	)
	15	+
	16	+	// sharedPhysicalsFunc adapts a closure to the probe's pool lending seam.
	17	+	type sharedPhysicalsFunc func(ports.BrokerDaemonIdentity, ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool)
	18	+
	19	+	func (f sharedPhysicalsFunc) SharedPhysical(identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
	20	+	return f(identity, policy)
	21	+	}
	22	+
	23	+	// TestBrokerRemoteProbeBorrowsWarmTransport proves observation reuses the
	24	+	// broker pool's attached or warm transport instead of bootstrapping SSH/QUIC
	25	+	// again, and redials only when the borrowed transport has retired.
	26	+	func TestBrokerRemoteProbeBorrowsWarmTransport(t *testing.T) {
	27	+	registration := domain.RemoteRegistration{Endpoint: "user@example.test", Incarnation: [16]byte{1}, Generation: 1}
	28	+	policy := remoteBrokerPolicy("quic")
	29	+	const identity ports.BrokerDaemonIdentity = "remote-daemon"
	30	+	dialErr := errors.New("dial refused")
	31	+	tests := []struct {
	32	+	name string
	33	+	bound bool
	34	+	shared bool
	35	+	openErr error
	36	+	retired bool
	37	+	wantDials int
	38	+	wantErr error
	39	+	}{
	40	+	{name: "warm transport observes without dialing", bound: true, shared: true},
	41	+	{name: "unbound identity dials", bound: false, shared: true, wantDials: 1, wantErr: dialErr},
	42	+	{name: "no pooled transport dials", bound: true, wantDials: 1, wantErr: dialErr},
	43	+	{name: "live transport refusal is reported without redial", bound: true, shared: true, openErr: errors.New("stream refused")},
	44	+	{name: "retired transport redials", bound: true, shared: true, openErr: errors.New("physical closed"), retired: true, wantDials: 1, wantErr: dialErr},
	45	+	}
	46	+	for _, tt := range tests {
	47	+	t.Run(tt.name, func(t *testing.T) {
	48	+	record := ports.BrokerHostRecord{Registration: registration, Policy: policy}
	49	+	if tt.bound {
	50	+	record.Identity = identity
	51	+	}
	52	+	routes := portsmocks.NewMockBrokerRouteAuthority(t)
	53	+	routes.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).Return(ports.BrokerDialTarget{
	54	+	Fence: ports.BrokerEndpointFence{Registration: registration}, Policy: policy, Address: "opaque", StartMode: ports.BrokerDaemonExistingOnly,
	55	+	ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: record.Identity, Bound: tt.bound},
	56	+	}, nil)
	57	+	connector := portsmocks.NewMockBrokerEndpointConnector(t)
	58	+	if tt.wantDials > 0 {
	59	+	connector.EXPECT().Connect(mock.Anything, mock.Anything).Return(nil, dialErr).Times(tt.wantDials)
	60	+	}
	61	+	physical := portsmocks.NewMockBrokerPhysicalConnection(t)
	62	+	done := make(chan struct{})
	63	+	if tt.retired {
	64	+	close(done)
	65	+	}
	66	+	if tt.bound && tt.shared {
	67	+	if tt.openErr != nil {
	68	+	physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(nil, tt.openErr)
	69	+	physical.EXPECT().Done().Return(done)
	70	+	} else {
	71	+	logical := portsmocks.NewMockBrokerLogicalConnection(t)
	72	+	var sent protocol.CommandRequest
	73	+	logical.EXPECT().SendClient(mock.Anything).RunAndReturn(func(message protocol.ClientMessage) error {
	74	+	sent = message.(protocol.CommandRequest)
	75	+	return nil
	76	+	})
	77	+	logical.EXPECT().ReceiveServer().RunAndReturn(func() (protocol.ServerMessage, error) {
	78	+	return protocol.CommandResult{RequestID: sent.RequestID, Outcome: protocol.CommandSucceeded, Output: testLocalCatalogue(t)}, nil
	79	+	})
	80	+	logical.EXPECT().Close().Return(nil)
	81	+	physical.EXPECT().OpenStream(mock.Anything, mock.Anything).Return(logical, nil)
	82	+	physical.EXPECT().Identity().Return(identity)
	83	+	physical.EXPECT().Incarnation().Return(ports.BrokerDaemonIncarnation{1})
	84	+	}
	85	+	}
	86	+	probe := &brokerRemoteProbe{
	87	+	epoch: 7, routes: routes, binder: portsmocks.NewMockBrokerIdentityBinder(t), connector: connector,
	88	+	hosts: &routeTestHosts{record: record, found: true},
	89	+	}
	90	+	probe.shared.share(sharedPhysicalsFunc(func(gotIdentity ports.BrokerDaemonIdentity, gotPolicy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
	91	+	if !tt.shared \|\| gotIdentity != identity \|\| gotPolicy != policy {
	92	+	return nil, false
	93	+	}
	94	+	return physical, true
	95	+	}))
	96	+
	97	+	observation, err := probe.Probe(context.Background(), registration)
	98	+	switch {
	99	+	case tt.wantErr != nil:
	100	+	require.ErrorIs(t, err, tt.wantErr)
	101	+	case tt.openErr != nil:
	102	+	require.ErrorIs(t, err, tt.openErr)
	103	+	default:
	104	+	require.NoError(t, err)
	105	+	require.Equal(t, identity, observation.Identity)
	106	+	require.True(t, observation.InventoryKnown)
	107	+	}
	108	+	})
	109	+	}
	110	+	}

M internal/usecase/broker/pool.go +28 -0


@@ -428,6 +428,34 @@ retire:
428	428		p.mu.Unlock()
429	429		}
430	430
	431	+	// SharedPhysical returns the published physical transport the pool holds for
	432	+	// one authenticated identity and exact policy, active or warm, without
	433	+	// reserving it. Observation borrows it to avoid a second bootstrap; the
	434	+	// borrower must not close it and must tolerate its retirement at any time.
	435	+	// Borrowing never refreshes warm order or the idle deadline.
	436	+	func (p *Pool) SharedPhysical(identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
	437	+	p.mu.Lock()
	438	+	defer p.mu.Unlock()
	439	+	e := p.entries[poolKey{identity: identity, policy: policy}]
	440	+	if p.closed \|\| e == nil \|\| e.retiring \|\| e.redirect != nil {
	441	+	return nil, false
	442	+	}
	443	+	select {
	444	+	case <-e.ready:
	445	+	default:
	446	+	return nil, false
	447	+	}
	448	+	if e.err != nil \|\| nilDependency(e.physical) {
	449	+	return nil, false
	450	+	}
	451	+	select {
	452	+	case <-e.physical.Done():
	453	+	return nil, false
	454	+	default:
	455	+	}
	456	+	return e.physical, true
	457	+	}
	458	+
431	459		// Close cancels pending operations, closes transports, and joins all owned
432	460		// workers. Bounded completion depends on the explicit cancellation/Close port
433	461		// contract; callers must not supply adapters that ignore it.


@@ -428,6 +428,34 @@ retire:
428	p.mu.Unlock()	428		p.mu.Unlock()
429	}	429		}
430		430
		431	+	// SharedPhysical returns the published physical transport the pool holds for
		432	+	// one authenticated identity and exact policy, active or warm, without
		433	+	// reserving it. Observation borrows it to avoid a second bootstrap; the
		434	+	// borrower must not close it and must tolerate its retirement at any time.
		435	+	// Borrowing never refreshes warm order or the idle deadline.
		436	+	func (p *Pool) SharedPhysical(identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
		437	+	p.mu.Lock()
		438	+	defer p.mu.Unlock()
		439	+	e := p.entries[poolKey{identity: identity, policy: policy}]
		440	+	if p.closed \|\| e == nil \|\| e.retiring \|\| e.redirect != nil {
		441	+	return nil, false
		442	+	}
		443	+	select {
		444	+	case <-e.ready:
		445	+	default:
		446	+	return nil, false
		447	+	}
		448	+	if e.err != nil \|\| nilDependency(e.physical) {
		449	+	return nil, false
		450	+	}
		451	+	select {
		452	+	case <-e.physical.Done():
		453	+	return nil, false
		454	+	default:
		455	+	}
		456	+	return e.physical, true
		457	+	}
		458	+
431	// Close cancels pending operations, closes transports, and joins all owned	459		// Close cancels pending operations, closes transports, and joins all owned
432	// workers. Bounded completion depends on the explicit cancellation/Close port	460		// workers. Bounded completion depends on the explicit cancellation/Close port
433	// contract; callers must not supply adapters that ignore it.	461		// contract; callers must not supply adapters that ignore it.