1package broker
2
3import (
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.
20type localProbeFunc func(context.Context) (ports.BrokerDaemonObservation, error)
21
22func (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.
32func 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}