fix(broker): bound every observation attempt by a ports.Clock deadline

A probe that never returned kept its endpoint's single in-flight slot
forever (Checking=true), so no later observation was dispatched. Each
local and remote attempt now runs under RegistryConfig.ProbeTimeout
(default 20s) armed at dispatch; on expiry the probe context is
cancelled and the attempt reports a typed timeout and releases its slot
even if the probe ignores cancellation. Probe failures are logged at
debug.

commit: 3fbd5a1762944f150a08876819f30b1bac867aca
author: brice · 30d ago · 2026-09-23 03:23 UTC
parent: 9b767e7
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +223 -2

M internal/usecase/broker/registry.go +64 -1


@@ -37,8 +37,18 @@ const (
37	37		// before.
38	38		defaultDemandFreshForLocal = 1 * time.Second
39	39		defaultDemandFreshForRemote = 2 * time.Second
	40	+	// defaultProbeTimeout bounds one observation attempt end to end: dial,
	41	+	// SSH/QUIC bootstrap, daemonmux preamble, and the catalogue exchange. It
	42	+	// sits below the passive retry cap so a peer that never answers costs one
	43	+	// bounded attempt instead of pinning the endpoint's single in-flight slot.
	44	+	defaultProbeTimeout = 20 * time.Second
40	45		)
41	46
	47	+	// errProbeTimeout is the failure an attempt reports when its ports.Clock
	48	+	// deadline fires before the probe returns. It wraps context.DeadlineExceeded
	49	+	// so observationOutcome classifies it as a typed timeout.
	50	+	var errProbeTimeout = fmt.Errorf("broker: observation exceeded its deadline: %w", context.DeadlineExceeded)
	51	+
42	52		// Run lifecycle states. A registry runs exactly once: runIdle is the zero
43	53		// value, runActive marks the single admitted run, and runStopped permanently
44	54		// refuses further runs.

@@ -83,6 +93,11 @@ type RegistryConfig struct {
83	93		// defaultFreshFor (15s) exactly as before this field existed.
84	94		DemandFreshForLocal time.Duration
85	95		DemandFreshForRemote time.Duration
	96	+	// ProbeTimeout bounds every local and remote observation attempt on the
	97	+	// registry's ports.Clock. When it fires the attempt is released and
	98	+	// recorded as a timeout failure even if the probe never returns. A zero
	99	+	// value selects defaultProbeTimeout.
	100	+	ProbeTimeout time.Duration
86	101		}
87	102
88	103		// Registry owns configured hosts and their immutable observation projection.

@@ -165,6 +180,9 @@ type Registry struct {
165	180		// demand cadences; see RegistryConfig.DemandFreshForLocal/Remote.
166	181		demandFreshForLocal time.Duration
167	182		demandFreshForRemote time.Duration
	183	+	// probeTimeout is the resolved (non-zero) per-attempt deadline; see
	184	+	// RegistryConfig.ProbeTimeout.
	185	+	probeTimeout time.Duration
168	186		// demand counts live client subscriptions: Registry.SetDemand(true) is one
169	187		// subscription opening and SetDemand(false) is one closing. Guarded by mu.
170	188		// While demand > 0, apply and applyLocal schedule NextDue on the faster

@@ -207,6 +225,10 @@ func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore,
207	225		if demandFreshForRemote <= 0 {
208	226		demandFreshForRemote = defaultDemandFreshForRemote
209	227		}
	228	+	probeTimeout := cfg.ProbeTimeout
	229	+	if probeTimeout <= 0 {
	230	+	probeTimeout = defaultProbeTimeout
	231	+	}
210	232		r := &Registry{
211	233		epoch: epoch, probe: probe, clock: clock, log: log,
212	234		observationDisabled: cfg.ObservationDisabled,

@@ -218,6 +240,7 @@ func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore,
218	240		jitter: jittered,
219	241		demandFreshForLocal: demandFreshForLocal,
220	242		demandFreshForRemote: demandFreshForRemote,
	243	+	probeTimeout: probeTimeout,
221	244		}
222	245		if cfg.ObservationDisabled && cfg.Local != nil {
223	246		return nil, errors.New("broker: observation-disabled registry cannot observe the local daemon")

@@ -997,8 +1020,14 @@ func (r *Registry) dispatch(ctx context.Context, now time.Time, results chan<- p
997	1020		host.LastAttempt = now
998	1021		r.hosts[endpoint] = host
999	1022		registration := host.Registration
	1023	+	// The deadline is armed here, under r.mu and at dispatch time, so the
	1024	+	// attempt's bound starts with the attempt rather than whenever its
	1025	+	// goroutine is scheduled.
	1026	+	deadline := r.clock.NewTimer(r.probeTimeout)
1000	1027		go func(endpoint string, registration domain.RemoteRegistration, attempt uint64, probeCtx context.Context) {
1001		-	snapshot, err := r.probe.Probe(probeCtx, registration)
	1028	+	snapshot, err := observeBounded(probeCtx, deadline, func(ctx context.Context) (ports.BrokerDaemonObservation, error) {
	1029	+	return r.probe.Probe(ctx, registration)
	1030	+	})
1002	1031		result := probeResult{endpoint: endpoint, registration: registration, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}
1003	1032		select {
1004	1033		case results <- result:

@@ -1021,6 +1050,39 @@ func (r *Registry) dispatch(ctx context.Context, now time.Time, results chan<- p
1021	1050		r.mu.Unlock()
1022	1051		}
1023	1052
	1053	+	// observeBounded runs one observation under its attempt deadline. retire is the
	1054	+	// attempt's retirement context: a replaced, removed, or settled attempt returns
	1055	+	// at once and its completion is fenced by the caller. The probe itself runs on
	1056	+	// a context derived from retire that the deadline also cancels, so a
	1057	+	// ctx-honoring probe unwinds its dial, handshake, or exchange when either
	1058	+	// fires. The attempt never waits for the probe past its deadline: a probe that
	1059	+	// ignores cancellation is abandoned (its late result is discarded) and the
	1060	+	// attempt still reports errProbeTimeout, so the endpoint's single in-flight
	1061	+	// slot is always released.
	1062	+	func observeBounded(retire context.Context, deadline ports.Timer, observe func(context.Context) (ports.BrokerDaemonObservation, error)) (ports.BrokerDaemonObservation, error) {
	1063	+	defer deadline.Stop()
	1064	+	probeCtx, cancel := context.WithCancelCause(retire)
	1065	+	defer cancel(nil)
	1066	+	type outcome struct {
	1067	+	snapshot ports.BrokerDaemonObservation
	1068	+	err error
	1069	+	}
	1070	+	done := make(chan outcome, 1)
	1071	+	go func() {
	1072	+	snapshot, err := observe(probeCtx)
	1073	+	done <- outcome{snapshot: snapshot, err: err}
	1074	+	}()
	1075	+	select {
	1076	+	case result := <-done:
	1077	+	return result.snapshot, result.err
	1078	+	case <-deadline.C():
	1079	+	cancel(errProbeTimeout)
	1080	+	return ports.BrokerDaemonObservation{}, errProbeTimeout
	1081	+	case <-retire.Done():
	1082	+	return ports.BrokerDaemonObservation{}, retire.Err()
	1083	+	}
	1084	+	}
	1085	+
1024	1086		func (r *Registry) apply(result probeResult) {
1025	1087		r.mu.Lock()
1026	1088		defer r.mu.Unlock()

@@ -1085,6 +1147,7 @@ func (r *Registry) apply(result probeResult) {
1085	1147		current.ConsecutiveFailures++
1086	1148		current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
1087	1149		current.NextDue = result.at.Add(r.jitter(r.retryDelay(current.ConsecutiveFailures), result.endpoint, result.attempt))
	1150	+	r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err)
1088	1151		r.hosts[result.endpoint] = current
1089	1152		r.publishLocked(true)
1090	1153		}


@@ -37,8 +37,18 @@ const (
37		// before.	37		// before.
38		defaultDemandFreshForLocal = 1 * time.Second	38		defaultDemandFreshForLocal = 1 * time.Second
39		defaultDemandFreshForRemote = 2 * time.Second	39		defaultDemandFreshForRemote = 2 * time.Second
			40	+	// defaultProbeTimeout bounds one observation attempt end to end: dial,
			41	+	// SSH/QUIC bootstrap, daemonmux preamble, and the catalogue exchange. It
			42	+	// sits below the passive retry cap so a peer that never answers costs one
			43	+	// bounded attempt instead of pinning the endpoint's single in-flight slot.
			44	+	defaultProbeTimeout = 20 * time.Second
40		)	45		)
41			46
			47	+	// errProbeTimeout is the failure an attempt reports when its ports.Clock
			48	+	// deadline fires before the probe returns. It wraps context.DeadlineExceeded
			49	+	// so observationOutcome classifies it as a typed timeout.
			50	+	var errProbeTimeout = fmt.Errorf("broker: observation exceeded its deadline: %w", context.DeadlineExceeded)
			51	+
42		// Run lifecycle states. A registry runs exactly once: runIdle is the zero	52		// Run lifecycle states. A registry runs exactly once: runIdle is the zero
43		// value, runActive marks the single admitted run, and runStopped permanently	53		// value, runActive marks the single admitted run, and runStopped permanently
44		// refuses further runs.	54		// refuses further runs.

@@ -83,6 +93,11 @@ type RegistryConfig struct {
83		// defaultFreshFor (15s) exactly as before this field existed.	93		// defaultFreshFor (15s) exactly as before this field existed.
84		DemandFreshForLocal time.Duration	94		DemandFreshForLocal time.Duration
85		DemandFreshForRemote time.Duration	95		DemandFreshForRemote time.Duration
			96	+	// ProbeTimeout bounds every local and remote observation attempt on the
			97	+	// registry's ports.Clock. When it fires the attempt is released and
			98	+	// recorded as a timeout failure even if the probe never returns. A zero
			99	+	// value selects defaultProbeTimeout.
			100	+	ProbeTimeout time.Duration
86		}	101		}
87			102
88		// Registry owns configured hosts and their immutable observation projection.	103		// Registry owns configured hosts and their immutable observation projection.

@@ -165,6 +180,9 @@ type Registry struct {
165		// demand cadences; see RegistryConfig.DemandFreshForLocal/Remote.	180		// demand cadences; see RegistryConfig.DemandFreshForLocal/Remote.
166		demandFreshForLocal time.Duration	181		demandFreshForLocal time.Duration
167		demandFreshForRemote time.Duration	182		demandFreshForRemote time.Duration
			183	+	// probeTimeout is the resolved (non-zero) per-attempt deadline; see
			184	+	// RegistryConfig.ProbeTimeout.
			185	+	probeTimeout time.Duration
168		// demand counts live client subscriptions: Registry.SetDemand(true) is one	186		// demand counts live client subscriptions: Registry.SetDemand(true) is one
169		// subscription opening and SetDemand(false) is one closing. Guarded by mu.	187		// subscription opening and SetDemand(false) is one closing. Guarded by mu.
170		// While demand > 0, apply and applyLocal schedule NextDue on the faster	188		// While demand > 0, apply and applyLocal schedule NextDue on the faster

@@ -207,6 +225,10 @@ func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore,
207		if demandFreshForRemote <= 0 {	225		if demandFreshForRemote <= 0 {
208		demandFreshForRemote = defaultDemandFreshForRemote	226		demandFreshForRemote = defaultDemandFreshForRemote
209		}	227		}
			228	+	probeTimeout := cfg.ProbeTimeout
			229	+	if probeTimeout <= 0 {
			230	+	probeTimeout = defaultProbeTimeout
			231	+	}
210		r := &Registry{	232		r := &Registry{
211		epoch: epoch, probe: probe, clock: clock, log: log,	233		epoch: epoch, probe: probe, clock: clock, log: log,
212		observationDisabled: cfg.ObservationDisabled,	234		observationDisabled: cfg.ObservationDisabled,

@@ -218,6 +240,7 @@ func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore,
218		jitter: jittered,	240		jitter: jittered,
219		demandFreshForLocal: demandFreshForLocal,	241		demandFreshForLocal: demandFreshForLocal,
220		demandFreshForRemote: demandFreshForRemote,	242		demandFreshForRemote: demandFreshForRemote,
			243	+	probeTimeout: probeTimeout,
221		}	244		}
222		if cfg.ObservationDisabled && cfg.Local != nil {	245		if cfg.ObservationDisabled && cfg.Local != nil {
223		return nil, errors.New("broker: observation-disabled registry cannot observe the local daemon")	246		return nil, errors.New("broker: observation-disabled registry cannot observe the local daemon")

@@ -997,8 +1020,14 @@ func (r *Registry) dispatch(ctx context.Context, now time.Time, results chan<- p
997		host.LastAttempt = now	1020		host.LastAttempt = now
998		r.hosts[endpoint] = host	1021		r.hosts[endpoint] = host
999		registration := host.Registration	1022		registration := host.Registration
			1023	+	// The deadline is armed here, under r.mu and at dispatch time, so the
			1024	+	// attempt's bound starts with the attempt rather than whenever its
			1025	+	// goroutine is scheduled.
			1026	+	deadline := r.clock.NewTimer(r.probeTimeout)
1000		go func(endpoint string, registration domain.RemoteRegistration, attempt uint64, probeCtx context.Context) {	1027		go func(endpoint string, registration domain.RemoteRegistration, attempt uint64, probeCtx context.Context) {
1001	-	snapshot, err := r.probe.Probe(probeCtx, registration)	1028	+	snapshot, err := observeBounded(probeCtx, deadline, func(ctx context.Context) (ports.BrokerDaemonObservation, error) {
			1029	+	return r.probe.Probe(ctx, registration)
			1030	+	})
1002		result := probeResult{endpoint: endpoint, registration: registration, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}	1031		result := probeResult{endpoint: endpoint, registration: registration, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}
1003		select {	1032		select {
1004		case results <- result:	1033		case results <- result:

@@ -1021,6 +1050,39 @@ func (r *Registry) dispatch(ctx context.Context, now time.Time, results chan<- p
1021		r.mu.Unlock()	1050		r.mu.Unlock()
1022		}	1051		}
1023			1052
			1053	+	// observeBounded runs one observation under its attempt deadline. retire is the
			1054	+	// attempt's retirement context: a replaced, removed, or settled attempt returns
			1055	+	// at once and its completion is fenced by the caller. The probe itself runs on
			1056	+	// a context derived from retire that the deadline also cancels, so a
			1057	+	// ctx-honoring probe unwinds its dial, handshake, or exchange when either
			1058	+	// fires. The attempt never waits for the probe past its deadline: a probe that
			1059	+	// ignores cancellation is abandoned (its late result is discarded) and the
			1060	+	// attempt still reports errProbeTimeout, so the endpoint's single in-flight
			1061	+	// slot is always released.
			1062	+	func observeBounded(retire context.Context, deadline ports.Timer, observe func(context.Context) (ports.BrokerDaemonObservation, error)) (ports.BrokerDaemonObservation, error) {
			1063	+	defer deadline.Stop()
			1064	+	probeCtx, cancel := context.WithCancelCause(retire)
			1065	+	defer cancel(nil)
			1066	+	type outcome struct {
			1067	+	snapshot ports.BrokerDaemonObservation
			1068	+	err error
			1069	+	}
			1070	+	done := make(chan outcome, 1)
			1071	+	go func() {
			1072	+	snapshot, err := observe(probeCtx)
			1073	+	done <- outcome{snapshot: snapshot, err: err}
			1074	+	}()
			1075	+	select {
			1076	+	case result := <-done:
			1077	+	return result.snapshot, result.err
			1078	+	case <-deadline.C():
			1079	+	cancel(errProbeTimeout)
			1080	+	return ports.BrokerDaemonObservation{}, errProbeTimeout
			1081	+	case <-retire.Done():
			1082	+	return ports.BrokerDaemonObservation{}, retire.Err()
			1083	+	}
			1084	+	}
			1085	+
1024		func (r *Registry) apply(result probeResult) {	1086		func (r *Registry) apply(result probeResult) {
1025		r.mu.Lock()	1087		r.mu.Lock()
1026		defer r.mu.Unlock()	1088		defer r.mu.Unlock()

@@ -1085,6 +1147,7 @@ func (r *Registry) apply(result probeResult) {
1085		current.ConsecutiveFailures++	1147		current.ConsecutiveFailures++
1086		current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}	1148		current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
1087		current.NextDue = result.at.Add(r.jitter(r.retryDelay(current.ConsecutiveFailures), result.endpoint, result.attempt))	1149		current.NextDue = result.at.Add(r.jitter(r.retryDelay(current.ConsecutiveFailures), result.endpoint, result.attempt))
			1150	+	r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err)
1088		r.hosts[result.endpoint] = current	1151		r.hosts[result.endpoint] = current
1089		r.publishLocked(true)	1152		r.publishLocked(true)
1090		}	1153		}

M internal/usecase/broker/registry_local.go +2 -1


@@ -97,8 +97,9 @@ func (r *Registry) dispatchLocalLocked(ctx context.Context, now time.Time, resul
97	97		r.localAttempt = &probeAttempt{token: attempt, cancel: cancel}
98	98		r.localHost.Checking = true
99	99		r.localHost.LastAttempt = now
	100	+	deadline := r.clock.NewTimer(r.probeTimeout)
100	101		go func(attempt uint64, probeCtx context.Context) {
101		-	snapshot, err := r.local.Probe.ProbeLocal(probeCtx)
	102	+	snapshot, err := observeBounded(probeCtx, deadline, r.local.Probe.ProbeLocal)
102	103		result := probeResult{local: true, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}
103	104		select {
104	105		case results <- result:

 	r.localAttempt = &probeAttempt{token: attempt, cancel: cancel}
 	r.localHost.Checking = true
 	r.localHost.LastAttempt = now
+	deadline := r.clock.NewTimer(r.probeTimeout)
 	go func(attempt uint64, probeCtx context.Context) {
-		snapshot, err := r.local.Probe.ProbeLocal(probeCtx)
+		snapshot, err := observeBounded(probeCtx, deadline, r.local.Probe.ProbeLocal)
 		result := probeResult{local: true, attempt: attempt, snapshot: snapshot, err: err, at: r.clock.Now()}
 		select {
 		case results <- result:

A internal/usecase/broker/registry_probe_deadline_test.go +157 -0

@@ -0,0 +1,157 @@
	1	+	package broker
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync/atomic"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/bnema/vev/internal/domain"
	11	+	"github.com/bnema/vev/internal/ports"
	12	+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
	13	+	"github.com/bnema/vev/internal/protocol"
	14	+	"github.com/stretchr/testify/mock"
	15	+	"github.com/stretchr/testify/require"
	16	+	)
	17	+
	18	+	// localProbeFunc adapts a function to LocalProbe. LocalProbe is a use-case-local
	19	+	// seam outside internal/ports, so no generated portsmocks mock exists for it.
	20	+	type localProbeFunc func(context.Context) (ports.BrokerDaemonObservation, error)
	21	+
	22	+	func (f localProbeFunc) ProbeLocal(ctx context.Context) (ports.BrokerDaemonObservation, error) {
	23	+	return f(ctx)
	24	+	}
	25	+
	26	+	// TestRegistryProbeDeadlineReleasesAttempt proves every observation attempt,
	27	+	// local and remote, is bounded by the registry's ports.Clock deadline: a probe
	28	+	// that never answers (whether or not it honors cancellation) is recorded as a
	29	+	// typed timeout, its in-flight slot is released, and the next attempt is
	30	+	// dispatched on the retry cadence. A probe that answers in time never reports a
	31	+	// timeout once the deadline would have fired.
	32	+	func TestRegistryProbeDeadlineReleasesAttempt(t *testing.T) {
	33	+	const probeTimeout = 3 * time.Second
	34	+	tests := []struct {
	35	+	name string
	36	+	// local selects the broker's own daemon instead of one remote host.
	37	+	local bool
	38	+	// honorsCancel: the hung probe returns once its context is cancelled;
	39	+	// otherwise it ignores cancellation until the test ends.
	40	+	honorsCancel bool
	41	+	// answers: the first attempt succeeds immediately instead of hanging.
	42	+	answers bool
	43	+	}{
	44	+	{name: "remote probe honoring cancellation times out", honorsCancel: true},
	45	+	{name: "remote probe ignoring cancellation still releases", honorsCancel: false},
	46	+	{name: "local probe honoring cancellation times out", local: true, honorsCancel: true},
	47	+	{name: "local probe ignoring cancellation still releases", local: true, honorsCancel: false},
	48	+	{name: "remote probe answering in time is not timed out", answers: true},
	49	+	}
	50	+	for _, tt := range tests {
	51	+	t.Run(tt.name, func(t *testing.T) {
	52	+	clock := newManualClock(time.Unix(100, 0))
	53	+	reg := registration(t, "user@host:22", 1)
	54	+	release := make(chan struct{})
	55	+	t.Cleanup(func() { close(release) })
	56	+	// calls carries each attempt's context to the test goroutine; the
	57	+	// probe worker only sends, it never asserts.
	58	+	calls := make(chan context.Context, 8)
	59	+	reachable := ports.BrokerDaemonObservation{
	60	+	Identity: "daemon", Incarnation: ports.BrokerDaemonIncarnation{7}, ProtocolVersion: protocol.Version,
	61	+	Availability: domain.RemoteAvailabilityReachable, InventoryKnown: true,
	62	+	}
	63	+	var attempts atomic.Int32
	64	+	observe := func(ctx context.Context) (ports.BrokerDaemonObservation, error) {
	65	+	first := attempts.Add(1) == 1
	66	+	calls <- ctx
	67	+	if tt.answers \|\| !first {
	68	+	return reachable, nil
	69	+	}
	70	+	if tt.honorsCancel {
	71	+	<-ctx.Done()
	72	+	return ports.BrokerDaemonObservation{}, context.Cause(ctx)
	73	+	}
	74	+	<-release
	75	+	return ports.BrokerDaemonObservation{}, errors.New("late completion")
	76	+	}
	77	+
	78	+	remote := portsmocks.NewMockBrokerHostProbe(t)
	79	+	var localProbe *LocalObservation
	80	+	if tt.local {
	81	+	localProbe = &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: localProbeFunc(observe)}
	82	+	} else {
	83	+	remote.EXPECT().Probe(mock.Anything, reg).RunAndReturn(func(ctx context.Context, _ domain.RemoteRegistration) (ports.BrokerDaemonObservation, error) {
	84	+	return observe(ctx)
	85	+	})
	86	+	}
	87	+	registry, err := NewRegistryWithConfig(1, newTestStore(), remote, clock, nil, RegistryConfig{Local: localProbe, ProbeTimeout: probeTimeout})
	88	+	require.NoError(t, err)
	89	+	registry.jitter = identityJitter
	90	+	if !tt.local {
	91	+	require.NoError(t, registry.setHosts(hostRecords(reg)))
	92	+	}
	93	+	startRegistry(t, registry)
	94	+
	95	+	host := func(predicate func(ports.BrokerDaemonObservation) bool) ports.BrokerDaemonObservation {
	96	+	if tt.local {
	97	+	return waitLocal(t, registry, predicate)
	98	+	}
	99	+	return waitHost(t, registry, reg.Endpoint, predicate)
	100	+	}
	101	+	receive := func() context.Context {
	102	+	t.Helper()
	103	+	select {
	104	+	case ctx := <-calls:
	105	+	return ctx
	106	+	case <-time.After(time.Second):
	107	+	t.Fatal("timed out waiting for a probe attempt")
	108	+	return nil
	109	+	}
	110	+	}
	111	+
	112	+	firstCtx := receive()
	113	+	if tt.answers {
	114	+	host(func(o ports.BrokerDaemonObservation) bool {
	115	+	return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
	116	+	})
	117	+	// The deadline of a completed attempt is stopped: moving past it
	118	+	// never turns the observation into a timeout.
	119	+	clock.Advance(probeTimeout)
	120	+	observed := host(func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
	121	+	require.Equal(t, domain.RemoteFailureNone, observed.LastFailure.Kind)
	122	+	require.Zero(t, observed.ConsecutiveFailures)
	123	+	return
	124	+	}
	125	+
	126	+	host(func(o ports.BrokerDaemonObservation) bool { return o.Checking })
	127	+	clock.Advance(probeTimeout)
	128	+	failed := host(func(o ports.BrokerDaemonObservation) bool {
	129	+	return !o.Checking && o.ConsecutiveFailures == 1
	130	+	})
	131	+	require.Equal(t, domain.RemoteFailureTimeout, failed.LastFailure.Kind)
	132	+	require.ErrorIs(t, failed.LastFailure.Err, context.DeadlineExceeded)
	133	+	// The hung attempt's own context was cancelled by the deadline, so a
	134	+	// ctx-honoring dial or exchange unwinds instead of lingering.
	135	+	require.ErrorIs(t, context.Cause(firstCtx), errProbeTimeout)
	136	+
	137	+	// The slot was released: the retry cadence dispatches a fresh
	138	+	// attempt, which succeeds. Advance repeatedly because the run loop
	139	+	// re-arms its timer asynchronously after publishing the failure.
	140	+	var second context.Context
	141	+	require.Eventually(t, func() bool {
	142	+	clock.Advance(defaultRetryBase)
	143	+	select {
	144	+	case second = <-calls:
	145	+	return true
	146	+	default:
	147	+	return false
	148	+	}
	149	+	}, 2*time.Second, time.Millisecond)
	150	+	require.NotNil(t, second)
	151	+	recovered := host(func(o ports.BrokerDaemonObservation) bool {
	152	+	return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
	153	+	})
	154	+	require.Zero(t, recovered.ConsecutiveFailures)
	155	+	})
	156	+	}
	157	+	}

@@ -0,0 +1,157 @@
	1	+	package broker
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync/atomic"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/bnema/vev/internal/domain"
	11	+	"github.com/bnema/vev/internal/ports"
	12	+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
	13	+	"github.com/bnema/vev/internal/protocol"
	14	+	"github.com/stretchr/testify/mock"
	15	+	"github.com/stretchr/testify/require"
	16	+	)
	17	+
	18	+	// localProbeFunc adapts a function to LocalProbe. LocalProbe is a use-case-local
	19	+	// seam outside internal/ports, so no generated portsmocks mock exists for it.
	20	+	type localProbeFunc func(context.Context) (ports.BrokerDaemonObservation, error)
	21	+
	22	+	func (f localProbeFunc) ProbeLocal(ctx context.Context) (ports.BrokerDaemonObservation, error) {
	23	+	return f(ctx)
	24	+	}
	25	+
	26	+	// TestRegistryProbeDeadlineReleasesAttempt proves every observation attempt,
	27	+	// local and remote, is bounded by the registry's ports.Clock deadline: a probe
	28	+	// that never answers (whether or not it honors cancellation) is recorded as a
	29	+	// typed timeout, its in-flight slot is released, and the next attempt is
	30	+	// dispatched on the retry cadence. A probe that answers in time never reports a
	31	+	// timeout once the deadline would have fired.
	32	+	func TestRegistryProbeDeadlineReleasesAttempt(t *testing.T) {
	33	+	const probeTimeout = 3 * time.Second
	34	+	tests := []struct {
	35	+	name string
	36	+	// local selects the broker's own daemon instead of one remote host.
	37	+	local bool
	38	+	// honorsCancel: the hung probe returns once its context is cancelled;
	39	+	// otherwise it ignores cancellation until the test ends.
	40	+	honorsCancel bool
	41	+	// answers: the first attempt succeeds immediately instead of hanging.
	42	+	answers bool
	43	+	}{
	44	+	{name: "remote probe honoring cancellation times out", honorsCancel: true},
	45	+	{name: "remote probe ignoring cancellation still releases", honorsCancel: false},
	46	+	{name: "local probe honoring cancellation times out", local: true, honorsCancel: true},
	47	+	{name: "local probe ignoring cancellation still releases", local: true, honorsCancel: false},
	48	+	{name: "remote probe answering in time is not timed out", answers: true},
	49	+	}
	50	+	for _, tt := range tests {
	51	+	t.Run(tt.name, func(t *testing.T) {
	52	+	clock := newManualClock(time.Unix(100, 0))
	53	+	reg := registration(t, "user@host:22", 1)
	54	+	release := make(chan struct{})
	55	+	t.Cleanup(func() { close(release) })
	56	+	// calls carries each attempt's context to the test goroutine; the
	57	+	// probe worker only sends, it never asserts.
	58	+	calls := make(chan context.Context, 8)
	59	+	reachable := ports.BrokerDaemonObservation{
	60	+	Identity: "daemon", Incarnation: ports.BrokerDaemonIncarnation{7}, ProtocolVersion: protocol.Version,
	61	+	Availability: domain.RemoteAvailabilityReachable, InventoryKnown: true,
	62	+	}
	63	+	var attempts atomic.Int32
	64	+	observe := func(ctx context.Context) (ports.BrokerDaemonObservation, error) {
	65	+	first := attempts.Add(1) == 1
	66	+	calls <- ctx
	67	+	if tt.answers \|\| !first {
	68	+	return reachable, nil
	69	+	}
	70	+	if tt.honorsCancel {
	71	+	<-ctx.Done()
	72	+	return ports.BrokerDaemonObservation{}, context.Cause(ctx)
	73	+	}
	74	+	<-release
	75	+	return ports.BrokerDaemonObservation{}, errors.New("late completion")
	76	+	}
	77	+
	78	+	remote := portsmocks.NewMockBrokerHostProbe(t)
	79	+	var localProbe *LocalObservation
	80	+	if tt.local {
	81	+	localProbe = &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: localProbeFunc(observe)}
	82	+	} else {
	83	+	remote.EXPECT().Probe(mock.Anything, reg).RunAndReturn(func(ctx context.Context, _ domain.RemoteRegistration) (ports.BrokerDaemonObservation, error) {
	84	+	return observe(ctx)
	85	+	})
	86	+	}
	87	+	registry, err := NewRegistryWithConfig(1, newTestStore(), remote, clock, nil, RegistryConfig{Local: localProbe, ProbeTimeout: probeTimeout})
	88	+	require.NoError(t, err)
	89	+	registry.jitter = identityJitter
	90	+	if !tt.local {
	91	+	require.NoError(t, registry.setHosts(hostRecords(reg)))
	92	+	}
	93	+	startRegistry(t, registry)
	94	+
	95	+	host := func(predicate func(ports.BrokerDaemonObservation) bool) ports.BrokerDaemonObservation {
	96	+	if tt.local {
	97	+	return waitLocal(t, registry, predicate)
	98	+	}
	99	+	return waitHost(t, registry, reg.Endpoint, predicate)
	100	+	}
	101	+	receive := func() context.Context {
	102	+	t.Helper()
	103	+	select {
	104	+	case ctx := <-calls:
	105	+	return ctx
	106	+	case <-time.After(time.Second):
	107	+	t.Fatal("timed out waiting for a probe attempt")
	108	+	return nil
	109	+	}
	110	+	}
	111	+
	112	+	firstCtx := receive()
	113	+	if tt.answers {
	114	+	host(func(o ports.BrokerDaemonObservation) bool {
	115	+	return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
	116	+	})
	117	+	// The deadline of a completed attempt is stopped: moving past it
	118	+	// never turns the observation into a timeout.
	119	+	clock.Advance(probeTimeout)
	120	+	observed := host(func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
	121	+	require.Equal(t, domain.RemoteFailureNone, observed.LastFailure.Kind)
	122	+	require.Zero(t, observed.ConsecutiveFailures)
	123	+	return
	124	+	}
	125	+
	126	+	host(func(o ports.BrokerDaemonObservation) bool { return o.Checking })
	127	+	clock.Advance(probeTimeout)
	128	+	failed := host(func(o ports.BrokerDaemonObservation) bool {
	129	+	return !o.Checking && o.ConsecutiveFailures == 1
	130	+	})
	131	+	require.Equal(t, domain.RemoteFailureTimeout, failed.LastFailure.Kind)
	132	+	require.ErrorIs(t, failed.LastFailure.Err, context.DeadlineExceeded)
	133	+	// The hung attempt's own context was cancelled by the deadline, so a
	134	+	// ctx-honoring dial or exchange unwinds instead of lingering.
	135	+	require.ErrorIs(t, context.Cause(firstCtx), errProbeTimeout)
	136	+
	137	+	// The slot was released: the retry cadence dispatches a fresh
	138	+	// attempt, which succeeds. Advance repeatedly because the run loop
	139	+	// re-arms its timer asynchronously after publishing the failure.
	140	+	var second context.Context
	141	+	require.Eventually(t, func() bool {
	142	+	clock.Advance(defaultRetryBase)
	143	+	select {
	144	+	case second = <-calls:
	145	+	return true
	146	+	default:
	147	+	return false
	148	+	}
	149	+	}, 2*time.Second, time.Millisecond)
	150	+	require.NotNil(t, second)
	151	+	recovered := host(func(o ports.BrokerDaemonObservation) bool {
	152	+	return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
	153	+	})
	154	+	require.Zero(t, recovered.ConsecutiveFailures)
	155	+	})
	156	+	}
	157	+	}