1package daemon
2
3// Explicit KillResult daemon contract. handleKill answers every normally
4// decoded request with exactly one correlated KillResult before the control
5// connection closes; a KillDaemon result is delivered before teardown begins.
6// These tests are deterministic (no sleeps) and bounded by awaitFrame timeouts.
7
8import (
9 "errors"
10 "fmt"
11 "io"
12 "log/slog"
13 "strconv"
14 "sync"
15 "testing"
16
17 "github.com/stretchr/testify/mock"
18 "github.com/stretchr/testify/require"
19
20 "github.com/bnema/vev/internal/adapters/sessionwire"
21 "github.com/bnema/vev/internal/domain"
22 "github.com/bnema/vev/internal/protocol"
23 "github.com/bnema/vev/internal/protocol/wire"
24)
25
26func awaitKillResult(t *testing.T, sends chan wire.Envelope, requestID uint64) protocol.KillResult {
27 t.Helper()
28 frame := awaitFrame(t, sends, "KillResult")
29 result, ok := decodeServerMessage(t, frame).(protocol.KillResult)
30 require.True(t, ok, "expected a KillResult, got %T", decodeServerMessage(t, frame))
31 require.Equal(t, requestID, result.RequestID, "result must correlate to its request")
32 return result
33}
34
35// requireNoFurtherSends proves the handler produced exactly one result.
36func requireNoFurtherSends(t *testing.T, sends chan wire.Envelope) {
37 t.Helper()
38 select {
39 case extra := <-sends:
40 t.Fatalf("unexpected extra send %s", envelopeMessageName(t, extra.Payload))
41 default:
42 }
43}
44
45// TestHandleKillSessionSendsExactlyOneSucceededResult proves a direct kill of a
46// live session answers with one KillSucceeded result correlated to the request
47// and removes the session.
48func TestHandleKillSessionSendsExactlyOneSucceededResult(t *testing.T) {
49 d := newTestDaemon(t, nil, stubClock{})
50 addControlSession(d, "work", "t_work", "p_work")
51
52 tr, sends := newCapturingTransport(t)
53 d.handleKill(tr, protocol.Kill{RequestID: 7, Scope: protocol.KillSession, Name: "work"})
54
55 result := awaitKillResult(t, sends, 7)
56 require.Equal(t, protocol.KillResult{RequestID: 7, Outcome: protocol.KillSucceeded}, result)
57 require.Zero(t, sessionCount(d), "a succeeded kill must remove the session")
58 requireNoFurtherSends(t, sends)
59}
60
61// TestHandleKillSessionSendsDefiniteFailureResult proves a kill of a missing
62// session answers with one definite KillFailed result, not a close-as-success,
63// and leaves the registry untouched.
64func TestHandleKillSessionSendsDefiniteFailureResult(t *testing.T) {
65 d := newTestDaemon(t, nil, stubClock{})
66
67 tr, sends := newCapturingTransport(t)
68 d.handleKill(tr, protocol.Kill{RequestID: 9, Scope: protocol.KillSession, Name: "ghost"})
69
70 result := awaitKillResult(t, sends, 9)
71 require.Equal(t, protocol.KillFailed, result.Outcome)
72 require.Equal(t, protocol.ErrNoSuchSession, result.Code)
73 require.Equal(t, "no such session: ghost", result.Text)
74 requireNoFurtherSends(t, sends)
75}
76
77// TestHandleKillInvalidScopeSendsDefiniteFailureResult proves an invalid scope
78// still receives exactly one definite failure instead of a silent close.
79func TestHandleKillInvalidScopeSendsDefiniteFailureResult(t *testing.T) {
80 d := newTestDaemon(t, nil, stubClock{})
81
82 tr, sends := newCapturingTransport(t)
83 d.handleKill(tr, protocol.Kill{RequestID: 3, Scope: protocol.KillScope(9)})
84
85 result := awaitKillResult(t, sends, 3)
86 require.Equal(t, protocol.KillFailed, result.Outcome)
87 require.Equal(t, protocol.ErrInternal, result.Code)
88 require.Equal(t, "invalid kill scope", result.Text)
89}
90
91// TestHandleKillAllEmptyDaemonSendsOneSucceededResult proves a successful
92// kill-all answers with one KillSucceeded result and leaves the daemon serving.
93func TestHandleKillAllEmptyDaemonSendsOneSucceededResult(t *testing.T) {
94 d := newTestDaemon(t, nil, stubClock{})
95
96 tr, sends := newCapturingTransport(t)
97 d.handleKill(tr, protocol.Kill{RequestID: 11, Scope: protocol.KillAll})
98
99 result := awaitKillResult(t, sends, 11)
100 require.Equal(t, protocol.KillSucceeded, result.Outcome)
101 select {
102 case <-d.done:
103 t.Fatal("kill-all must not end the daemon")
104 default:
105 }
106 requireNoFurtherSends(t, sends)
107}
108
109// TestHandleKillAllReportsBoundedStructuredPartialFailures proves a kill-all
110// whose stopped records cannot be purged answers with one definite KillFailed
111// result carrying the total count plus at most purgeSummaryFailureLimit
112// structured failures, so a control response stays bounded.
113func TestHandleKillAllReportsBoundedStructuredPartialFailures(t *testing.T) {
114 const total = purgeSummaryFailureLimit + 3
115
116 fail := make(map[domain.IncarnationID]error, total)
117 repository := newSelectivePurgeRepository(fail)
118 d := newTestDaemon(t, nil, stubClock{})
119 WithSnapshotRepository(repository)(d)
120 store, _ := newMockStore(t)
121 WithStore(t, store)(d)
122
123 for i := 0; i < total; i++ {
124 incarnation := domain.IncarnationID{byte(i + 1)}
125 fail[incarnation] = errors.New("delete failed")
126 record := domain.CatalogueRecord{Name: fmt.Sprintf("s%02d", i), IncarnationID: incarnation, CreatedAt: int64(i + 1)}
127 require.NoError(t, d.catalogue.Create(record))
128 d.inactive[record.Name] = inactiveSessionFromRecord(record, protocol.SessionDown, nil)
129 }
130
131 tr, sends := newCapturingTransport(t)
132 d.handleKill(tr, protocol.Kill{RequestID: 13, Scope: protocol.KillAll})
133
134 result := awaitKillResult(t, sends, 13)
135 require.Equal(t, protocol.KillFailed, result.Outcome)
136 require.Equal(t, protocol.ErrInternal, result.Code)
137 require.Len(t, result.Failures, purgeSummaryFailureLimit, "wire detail must be bounded")
138 require.Contains(t, result.Text, fmt.Sprintf("%d record(s) could not be purged", total))
139 require.Contains(t, result.Text, "more omitted")
140 requireNoFurtherSends(t, sends)
141}
142
143// killOrderTransport records the order of the KillResult send and the control
144// close, and whether the daemon's shutdown signal was already closed when the
145// result was written.
146type killOrderTransport struct {
147 mu sync.Mutex
148 order []string
149 resultBeforeTeardown bool
150}
151
152func (r *killOrderTransport) record(name string) {
153 r.mu.Lock()
154 r.order = append(r.order, name)
155 r.mu.Unlock()
156}
157
158func (r *killOrderTransport) snapshot() ([]string, bool) {
159 r.mu.Lock()
160 defer r.mu.Unlock()
161 return append([]string(nil), r.order...), r.resultBeforeTeardown
162}
163
164func newKillOrderTransport(t *testing.T, teardown <-chan struct{}) (*mockServerConnection, *killOrderTransport) {
165 t.Helper()
166 recorder := &killOrderTransport{}
167 tr := newMockServerConnection(t)
168 tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f wire.Envelope) error {
169 name := envelopeMessageName(t, f.Payload)
170 if name == "KillResult" {
171 recorder.mu.Lock()
172 recorder.order = append(recorder.order, "result")
173 select {
174 case <-teardown:
175 default:
176 recorder.resultBeforeTeardown = true
177 }
178 recorder.mu.Unlock()
179 return nil
180 }
181 recorder.record(name)
182 return nil
183 }).Maybe()
184 tr.EXPECT().Close().RunAndReturn(func() error {
185 recorder.record("close")
186 return nil
187 }).Maybe()
188 return tr, recorder
189}
190
191// TestHandleKillDaemonSendsResultBeforeTeardownAndClose proves the explicit
192// daemon-stop result is delivered before shutdown begins and before the control
193// connection closes, so a client can observe success instead of inferring it
194// from a close.
195func TestHandleKillDaemonSendsResultBeforeTeardownAndClose(t *testing.T) {
196 d := newTestDaemon(t, nil, stubClock{})
197
198 tr, recorder := newKillOrderTransport(t, d.done)
199 d.handleKill(tr, protocol.Kill{RequestID: 42, Scope: protocol.KillDaemon})
200
201 order, beforeTeardown := recorder.snapshot()
202 require.Equal(t, []string{"result", "close"}, order)
203 require.True(t, beforeTeardown, "the daemon-stop result must be sent before teardown begins")
204 select {
205 case <-d.done:
206 default:
207 t.Fatal("KillDaemon must end the daemon")
208 }
209}
210
211// recordFailTransport records every sent envelope then fails, so a test can
212// inspect the exact result an undeliverable control send attempted.
213type recordFailTransport struct {
214 mu sync.Mutex
215 sent []wire.Envelope
216 err error
217}
218
219func (t *recordFailTransport) Send(envelope wire.Envelope) error {
220 t.mu.Lock()
221 t.sent = append(t.sent, envelope)
222 t.mu.Unlock()
223 return t.err
224}
225
226func (*recordFailTransport) Recv() (wire.Envelope, error) { return wire.Envelope{}, io.EOF }
227func (*recordFailTransport) Close() error { return nil }
228
229func (t *recordFailTransport) payloads() []wire.Envelope {
230 t.mu.Lock()
231 defer t.mu.Unlock()
232 return append([]wire.Envelope(nil), t.sent...)
233}
234
235// TestHandleKillAllSupersededByDaemonShutdownSendsOneUnknownOutcome proves a
236// KillAll that an explicit daemon shutdown superseded answers with exactly one
237// correlated KillOutcomeUnknown carrying ErrServerShutdown, records the outcome
238// and code in the send-failure log, and sends nothing further (no success and no
239// second result).
240func TestHandleKillAllSupersededByDaemonShutdownSendsOneUnknownOutcome(t *testing.T) {
241 d := newTestDaemon(t, nil, stubClock{})
242 var logged syncedLogBuffer
243 d.log = slog.New(slog.NewTextHandler(&logged, nil))
244
245 // An explicit stop already owns the lifecycle set, so the purge defers.
246 d.mu.Lock()
247 d.closing = true
248 d.mu.Unlock()
249
250 recorder := &recordFailTransport{err: errors.New("client gone")}
251 d.handleKill(&rawServerConnection{raw: recorder}, protocol.Kill{RequestID: 5, Scope: protocol.KillAll})
252
253 payloads := recorder.payloads()
254 require.Len(t, payloads, 1, "a superseded kill-all must attempt exactly one result")
255 message, err := sessionwire.DecodeServerEnvelope(payloads[0].Payload)
256 require.NoError(t, err)
257 result, ok := message.(protocol.KillResult)
258 require.True(t, ok, "expected KillResult, got %T", message)
259 require.Equal(t, protocol.KillResult{RequestID: 5, Outcome: protocol.KillOutcomeUnknown, Code: protocol.ErrServerShutdown, Text: "daemon is shutting down"}, result)
260
261 require.Contains(t, logged.String(), "kill result")
262 require.Contains(t, logged.String(), "client gone")
263 require.Contains(t, logged.String(), "outcome="+strconv.Itoa(int(protocol.KillOutcomeUnknown)), "the send-failure log must name the unobserved outcome")
264 require.Contains(t, logged.String(), "code="+strconv.Itoa(int(protocol.ErrServerShutdown)), "the send-failure log must name the unobserved code")
265}
266
267// TestKillAllUndeliverableResultIsLogged proves an explicit partial-failure
268// result that cannot be delivered is logged (never silently closing as
269// success) and still carries the bounded structured failures.
270func TestKillAllUndeliverableResultIsLogged(t *testing.T) {
271 repository := newSelectivePurgeRepository(map[domain.IncarnationID]error{{2}: errors.New("delete failed")})
272 d := newTestDaemon(t, nil, stubClock{})
273 var logged syncedLogBuffer
274 d.log = slog.New(slog.NewTextHandler(&logged, nil))
275 WithSnapshotRepository(repository)(d)
276 store, _ := newMockStore(t)
277 WithStore(t, store)(d)
278
279 record := domain.CatalogueRecord{Name: "stopped", IncarnationID: domain.IncarnationID{2}, CreatedAt: 7}
280 require.NoError(t, d.catalogue.Create(record))
281 d.inactive["stopped"] = inactiveSessionFromRecord(record, protocol.SessionDown, nil)
282
283 recorder := &recordFailTransport{err: errors.New("client gone")}
284 d.handleKill(&rawServerConnection{raw: recorder}, protocol.Kill{RequestID: 1, Scope: protocol.KillAll})
285
286 require.Contains(t, logged.String(), "kill result")
287 require.Contains(t, logged.String(), "client gone")
288
289 payloads := recorder.payloads()
290 require.Len(t, payloads, 1, "exactly one result is attempted")
291 message, err := sessionwire.DecodeServerEnvelope(payloads[0].Payload)
292 require.NoError(t, err)
293 result, ok := message.(protocol.KillResult)
294 require.True(t, ok, "expected KillResult, got %T", message)
295 require.Equal(t, protocol.KillFailed, result.Outcome)
296 require.Len(t, result.Failures, 1)
297 require.Equal(t, "stopped", result.Failures[0].Name)
298}