467a7dd feat(broker): retain first probe transport in warm pool · vev

M internal/app/broker_remote_probe.go +17 -2


@@ -41,6 +41,7 @@ type brokerRemoteProbe struct {
41	41		// sharedPhysicals lends the pooled transport of an attached or warm daemon.
42	42		type sharedPhysicals interface {
43	43		SharedPhysical(ports.BrokerDaemonIdentity, ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool)
	44	+	AdoptPhysical(ports.BrokerPhysicalConnection, ports.BrokerPolicy) bool
44	45		}
45	46
46	47		// pooledPhysicals lets an observation probe borrow the broker pool's physical

@@ -53,6 +54,11 @@ type pooledPhysicals struct{ pool atomic.Value }
53	54
54	55		func (s *pooledPhysicals) share(pool sharedPhysicals) { s.pool.Store(pool) }
55	56
	57	+	func (s *pooledPhysicals) adopt(physical ports.BrokerPhysicalConnection, policy ports.BrokerPolicy) bool {
	58	+	pool, _ := s.pool.Load().(sharedPhysicals)
	59	+	return pool != nil && pool.AdoptPhysical(physical, policy)
	60	+	}
	61	+
56	62		// observe reports handled=false when the probe must dial its own transport.
57	63		func (s *pooledPhysicals) observe(ctx context.Context, identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy, request ports.BrokerOpenStreamRequest) (ports.BrokerDaemonObservation, bool, error) {
58	64		pool, _ := s.pool.Load().(sharedPhysicals)

@@ -101,7 +107,12 @@ func (p *brokerRemoteProbe) Probe(ctx context.Context, registration domain.Remot
101	107		if err != nil {
102	108		return ports.BrokerDaemonObservation{}, err
103	109		}
104		-	defer physical.Close()
	110	+	adopted := false
	111	+	defer func() {
	112	+	if !adopted {
	113	+	_ = physical.Close()
	114	+	}
	115	+	}()
105	116		if (endpoint.ExpectedIdentity.Bound && physical.Identity() != endpoint.ExpectedIdentity.Identity) \|\| physical.Policy() != endpoint.Policy {
106	117		return ports.BrokerDaemonObservation{}, errors.New("vev: remote probe authenticated binding mismatch")
107	118		}

@@ -109,7 +120,11 @@ func (p *brokerRemoteProbe) Probe(ctx context.Context, registration domain.Remot
109	120		if err != nil \|\| identity != physical.Identity() {
110	121		return ports.BrokerDaemonObservation{}, errors.New("vev: remote probe could not commit authenticated binding")
111	122		}
112		-	return observeDaemonCatalogue(ctx, physical, request)
	123	+	observation, err := observeDaemonCatalogue(ctx, physical, request)
	124	+	if err == nil && ctx.Err() == nil {
	125	+	adopted = p.shared.adopt(physical, endpoint.Policy)
	126	+	}
	127	+	return observation, err
113	128		}
114	129
115	130		// observeDaemonCatalogue observes one already authenticated daemon over its

 // sharedPhysicals lends the pooled transport of an attached or warm daemon.
 type sharedPhysicals interface {
 	SharedPhysical(ports.BrokerDaemonIdentity, ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool)
+	AdoptPhysical(ports.BrokerPhysicalConnection, ports.BrokerPolicy) bool
 }
 // pooledPhysicals lets an observation probe borrow the broker pool's physical
 func (s *pooledPhysicals) share(pool sharedPhysicals) { s.pool.Store(pool) }
+func (s *pooledPhysicals) adopt(physical ports.BrokerPhysicalConnection, policy ports.BrokerPolicy) bool {
+	pool, _ := s.pool.Load().(sharedPhysicals)
+	return pool != nil && pool.AdoptPhysical(physical, policy)
+}
 // observe reports handled=false when the probe must dial its own transport.
 func (s *pooledPhysicals) observe(ctx context.Context, identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy, request ports.BrokerOpenStreamRequest) (ports.BrokerDaemonObservation, bool, error) {
 	pool, _ := s.pool.Load().(sharedPhysicals)
 	if err != nil {
 		return ports.BrokerDaemonObservation{}, err
 	}
-	defer physical.Close()
+	adopted := false
+	defer func() {
+		if !adopted {
+			_ = physical.Close()
+		}
+	}()
 	if (endpoint.ExpectedIdentity.Bound && physical.Identity() != endpoint.ExpectedIdentity.Identity) || physical.Policy() != endpoint.Policy {
 		return ports.BrokerDaemonObservation{}, errors.New("vev: remote probe authenticated binding mismatch")
 	}
 	if err != nil || identity != physical.Identity() {
 		return ports.BrokerDaemonObservation{}, errors.New("vev: remote probe could not commit authenticated binding")
 	}
-	return observeDaemonCatalogue(ctx, physical, request)
+	observation, err := observeDaemonCatalogue(ctx, physical, request)
+	if err == nil && ctx.Err() == nil {
+		adopted = p.shared.adopt(physical, endpoint.Policy)
+	}
+	return observation, err
 }
 // observeDaemonCatalogue observes one already authenticated daemon over its

M internal/app/broker_remote_probe_shared_test.go +4 -0


@@ -20,6 +20,10 @@ func (f sharedPhysicalsFunc) SharedPhysical(identity ports.BrokerDaemonIdentity,
20	20		return f(identity, policy)
21	21		}
22	22
	23	+	func (sharedPhysicalsFunc) AdoptPhysical(ports.BrokerPhysicalConnection, ports.BrokerPolicy) bool {
	24	+	return false
	25	+	}
	26	+
23	27		// TestBrokerRemoteProbeBorrowsWarmTransport proves observation reuses the
24	28		// broker pool's attached or warm transport instead of bootstrapping SSH/QUIC
25	29		// again, and redials only when the borrowed transport has retired.


@@ -20,6 +20,10 @@ func (f sharedPhysicalsFunc) SharedPhysical(identity ports.BrokerDaemonIdentity,
20	return f(identity, policy)	20		return f(identity, policy)
21	}	21		}
22		22
		23	+	func (sharedPhysicalsFunc) AdoptPhysical(ports.BrokerPhysicalConnection, ports.BrokerPolicy) bool {
		24	+	return false
		25	+	}
		26	+
23	// TestBrokerRemoteProbeBorrowsWarmTransport proves observation reuses the	27		// TestBrokerRemoteProbeBorrowsWarmTransport proves observation reuses the
24	// broker pool's attached or warm transport instead of bootstrapping SSH/QUIC	28		// broker pool's attached or warm transport instead of bootstrapping SSH/QUIC
25	// again, and redials only when the borrowed transport has retired.	29		// again, and redials only when the borrowed transport has retired.

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


@@ -489,6 +489,89 @@ func (p Pool) idleLocked(e poolEntry) []context.CancelFunc {
489	489		return evicted
490	490		}
491	491
	492	+	// AdoptPhysical publishes a probe-authenticated transport as a warm entry.
	493	+	// The caller retains ownership when adoption fails and must close it. A
	494	+	// concurrently opened pooled transport wins: probes must not replace it.
	495	+	func (p *Pool) AdoptPhysical(physical ports.BrokerPhysicalConnection, policy ports.BrokerPolicy) bool {
	496	+	if nilDependency(physical) \|\| physical.Identity().Validate() != nil \|\|
	497	+	physical.Incarnation().Validate() != nil \|\| !physical.Policy().Compatible(policy) \|\| p.limits.Warm == 0 {
	498	+	return false
	499	+	}
	500	+	select {
	501	+	case <-physical.Done():
	502	+	return false
	503	+	default:
	504	+	}
	505	+	key := poolKey{identity: physical.Identity(), policy: policy}
	506	+	p.mu.Lock()
	507	+	if p.closed \|\| len(p.entries)+len(p.pending) >= p.limits.Physical \|\| p.entries[key] != nil {
	508	+	p.mu.Unlock()
	509	+	return false
	510	+	}
	511	+	ctx, cancel := context.WithCancel(p.ctx)
	512	+	e := &poolEntry{key: key, ctx: ctx, cancel: cancel, ready: make(chan struct{}), wake: make(chan struct{}, 1), physical: physical}
	513	+	close(e.ready)
	514	+	p.entries[key] = e
	515	+	evicted := p.idleLocked(e)
	516	+	p.wg.Add(1)
	517	+	p.mu.Unlock()
	518	+	for _, evict := range evicted {
	519	+	evict()
	520	+	}
	521	+	go p.watchAdopted(e)
	522	+	return true
	523	+	}
	524	+
	525	+	func (p Pool) watchAdopted(e poolEntry) {
	526	+	defer p.wg.Done()
	527	+	physical := e.physical
	528	+	var timer ports.Timer
	529	+	if p.limits.Idle > 0 {
	530	+	timer = p.clock.NewTimer(p.limits.Idle)
	531	+	defer stopPoolTimer(timer)
	532	+	}
	533	+	for {
	534	+	p.mu.Lock()
	535	+	idle := e.refs == 0
	536	+	p.mu.Unlock()
	537	+	var expiry <-chan time.Time
	538	+	if timer != nil {
	539	+	stopPoolTimer(timer)
	540	+	if idle {
	541	+	timer.Reset(p.limits.Idle)
	542	+	expiry = timer.C()
	543	+	}
	544	+	}
	545	+	select {
	546	+	case <-e.ctx.Done():
	547	+	goto retire
	548	+	case <-physical.Done():
	549	+	goto retire
	550	+	case <-e.wake:
	551	+	continue
	552	+	case <-expiry:
	553	+	p.mu.Lock()
	554	+	if e.refs == 0 {
	555	+	e.retiring = true
	556	+	p.mu.Unlock()
	557	+	goto retire
	558	+	}
	559	+	p.mu.Unlock()
	560	+	}
	561	+	}
	562	+	retire:
	563	+	p.mu.Lock()
	564	+	e.retiring = true
	565	+	p.mu.Unlock()
	566	+	e.cancel()
	567	+	_ = physical.Close()
	568	+	p.mu.Lock()
	569	+	if p.entries[e.key] == e {
	570	+	delete(p.entries, e.key)
	571	+	}
	572	+	p.mu.Unlock()
	573	+	}
	574	+
492	575		// SharedPhysical returns the published physical transport the pool holds for
493	576		// one authenticated identity and exact policy, active or warm, without
494	577		// reserving it. Observation borrows it to avoid a second bootstrap; the


@@ -489,6 +489,89 @@ func (p Pool) idleLocked(e poolEntry) []context.CancelFunc {
489	return evicted	489		return evicted
490	}	490		}
491		491
		492	+	// AdoptPhysical publishes a probe-authenticated transport as a warm entry.
		493	+	// The caller retains ownership when adoption fails and must close it. A
		494	+	// concurrently opened pooled transport wins: probes must not replace it.
		495	+	func (p *Pool) AdoptPhysical(physical ports.BrokerPhysicalConnection, policy ports.BrokerPolicy) bool {
		496	+	if nilDependency(physical) \|\| physical.Identity().Validate() != nil \|\|
		497	+	physical.Incarnation().Validate() != nil \|\| !physical.Policy().Compatible(policy) \|\| p.limits.Warm == 0 {
		498	+	return false
		499	+	}
		500	+	select {
		501	+	case <-physical.Done():
		502	+	return false
		503	+	default:
		504	+	}
		505	+	key := poolKey{identity: physical.Identity(), policy: policy}
		506	+	p.mu.Lock()
		507	+	if p.closed \|\| len(p.entries)+len(p.pending) >= p.limits.Physical \|\| p.entries[key] != nil {
		508	+	p.mu.Unlock()
		509	+	return false
		510	+	}
		511	+	ctx, cancel := context.WithCancel(p.ctx)
		512	+	e := &poolEntry{key: key, ctx: ctx, cancel: cancel, ready: make(chan struct{}), wake: make(chan struct{}, 1), physical: physical}
		513	+	close(e.ready)
		514	+	p.entries[key] = e
		515	+	evicted := p.idleLocked(e)
		516	+	p.wg.Add(1)
		517	+	p.mu.Unlock()
		518	+	for _, evict := range evicted {
		519	+	evict()
		520	+	}
		521	+	go p.watchAdopted(e)
		522	+	return true
		523	+	}
		524	+
		525	+	func (p Pool) watchAdopted(e poolEntry) {
		526	+	defer p.wg.Done()
		527	+	physical := e.physical
		528	+	var timer ports.Timer
		529	+	if p.limits.Idle > 0 {
		530	+	timer = p.clock.NewTimer(p.limits.Idle)
		531	+	defer stopPoolTimer(timer)
		532	+	}
		533	+	for {
		534	+	p.mu.Lock()
		535	+	idle := e.refs == 0
		536	+	p.mu.Unlock()
		537	+	var expiry <-chan time.Time
		538	+	if timer != nil {
		539	+	stopPoolTimer(timer)
		540	+	if idle {
		541	+	timer.Reset(p.limits.Idle)
		542	+	expiry = timer.C()
		543	+	}
		544	+	}
		545	+	select {
		546	+	case <-e.ctx.Done():
		547	+	goto retire
		548	+	case <-physical.Done():
		549	+	goto retire
		550	+	case <-e.wake:
		551	+	continue
		552	+	case <-expiry:
		553	+	p.mu.Lock()
		554	+	if e.refs == 0 {
		555	+	e.retiring = true
		556	+	p.mu.Unlock()
		557	+	goto retire
		558	+	}
		559	+	p.mu.Unlock()
		560	+	}
		561	+	}
		562	+	retire:
		563	+	p.mu.Lock()
		564	+	e.retiring = true
		565	+	p.mu.Unlock()
		566	+	e.cancel()
		567	+	_ = physical.Close()
		568	+	p.mu.Lock()
		569	+	if p.entries[e.key] == e {
		570	+	delete(p.entries, e.key)
		571	+	}
		572	+	p.mu.Unlock()
		573	+	}
		574	+
492	// SharedPhysical returns the published physical transport the pool holds for	575		// SharedPhysical returns the published physical transport the pool holds for
493	// one authenticated identity and exact policy, active or warm, without	576		// one authenticated identity and exact policy, active or warm, without
494	// reserving it. Observation borrows it to avoid a second bootstrap; the	577		// reserving it. Observation borrows it to avoid a second bootstrap; the

M internal/usecase/broker/pool_warm_test.go +29 -0


@@ -194,6 +194,35 @@ func TestPoolWarmReuse(t *testing.T) {
194	194		}
195	195		}
196	196
	197	+	func TestPoolAdoptProbePhysical(t *testing.T) {
	198	+	for _, tt := range []struct {
	199	+	name string
	200	+	warm int
	201	+	want bool
	202	+	}{
	203	+	{name: "warm probe is reused", warm: 2, want: true},
	204	+	{name: "warm disabled refuses probe", warm: 0},
	205	+	} {
	206	+	t.Run(tt.name, func(t *testing.T) {
	207	+	h := newWarmHarness(t, tt.warm, 0)
	208	+	physical := &fakePhysical{endpoint: ports.BrokerDialTarget{Policy: poolPolicy(), ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "alpha", Bound: true}}, done: make(chan struct{})}
	209	+	require.Equal(t, tt.want, h.pool.AdoptPhysical(physical, poolPolicy()))
	210	+	if !tt.want {
	211	+	require.NoError(t, physical.Close())
	212	+	return
	213	+	}
	214	+	shared, ok := h.pool.SharedPhysical("alpha", poolPolicy())
	215	+	require.True(t, ok)
	216	+	require.Same(t, physical, shared)
	217	+	h.visit(t, "alpha")
	218	+	require.Empty(t, h.physicals("alpha"), "attach must not redial after a first probe")
	219	+	require.False(t, h.pool.AdoptPhysical(&fakePhysical{endpoint: physical.endpoint, done: make(chan struct{})}, poolPolicy()), "duplicate probe cannot replace the attachment")
	220	+	require.NoError(t, h.pool.Close())
	221	+	require.True(t, closed(physical))
	222	+	})
	223	+	}
	224	+	}
	225	+
197	226		func TestPoolCloseRetiresWarmTransports(t *testing.T) {
198	227		h := newWarmHarness(t, 2, 0)
199	228		h.visit(t, "alpha")


@@ -194,6 +194,35 @@ func TestPoolWarmReuse(t *testing.T) {
194	}	194		}
195	}	195		}
196		196
		197	+	func TestPoolAdoptProbePhysical(t *testing.T) {
		198	+	for _, tt := range []struct {
		199	+	name string
		200	+	warm int
		201	+	want bool
		202	+	}{
		203	+	{name: "warm probe is reused", warm: 2, want: true},
		204	+	{name: "warm disabled refuses probe", warm: 0},
		205	+	} {
		206	+	t.Run(tt.name, func(t *testing.T) {
		207	+	h := newWarmHarness(t, tt.warm, 0)
		208	+	physical := &fakePhysical{endpoint: ports.BrokerDialTarget{Policy: poolPolicy(), ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "alpha", Bound: true}}, done: make(chan struct{})}
		209	+	require.Equal(t, tt.want, h.pool.AdoptPhysical(physical, poolPolicy()))
		210	+	if !tt.want {
		211	+	require.NoError(t, physical.Close())
		212	+	return
		213	+	}
		214	+	shared, ok := h.pool.SharedPhysical("alpha", poolPolicy())
		215	+	require.True(t, ok)
		216	+	require.Same(t, physical, shared)
		217	+	h.visit(t, "alpha")
		218	+	require.Empty(t, h.physicals("alpha"), "attach must not redial after a first probe")
		219	+	require.False(t, h.pool.AdoptPhysical(&fakePhysical{endpoint: physical.endpoint, done: make(chan struct{})}, poolPolicy()), "duplicate probe cannot replace the attachment")
		220	+	require.NoError(t, h.pool.Close())
		221	+	require.True(t, closed(physical))
		222	+	})
		223	+	}
		224	+	}
		225	+
197	func TestPoolCloseRetiresWarmTransports(t *testing.T) {	226		func TestPoolCloseRetiresWarmTransports(t *testing.T) {
198	h := newWarmHarness(t, 2, 0)	227		h := newWarmHarness(t, 2, 0)
199	h.visit(t, "alpha")	228		h.visit(t, "alpha")