1package apps_test
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "sync/atomic"
8 "testing"
9 "time"
10
11 "github.com/bnema/zerowrap"
12 "github.com/stretchr/testify/assert"
13 "github.com/stretchr/testify/mock"
14 "github.com/stretchr/testify/require"
15
16 "github.com/bnema/gordon/internal/domain"
17 "github.com/bnema/gordon/internal/usecase/apps"
18 "github.com/bnema/gordon/internal/usecase/deployment"
19)
20
21type deployCall struct {
22 op *domain.AppOperation
23 err error
24}
25
26// callDeploy runs Deploy off the test goroutine so a blocking execution can
27// never hang the test: only a prompt return is awaited.
28func callDeploy(t *testing.T, svc *apps.AppServiceImpl, ctx context.Context, key string) <-chan deployCall {
29 t.Helper()
30 out := make(chan deployCall, 1)
31 go func() {
32 op, err := svc.Deploy(ctx, "blog", "rev-1", "web", false, key)
33 out <- deployCall{op: op, err: err}
34 }()
35 return out
36}
37
38func awaitDeploy(t *testing.T, ch <-chan deployCall) deployCall {
39 t.Helper()
40 select {
41 case res := <-ch:
42 return res
43 case <-time.After(3 * time.Second):
44 t.Fatal("Deploy did not return promptly")
45 return deployCall{}
46 }
47}
48
49// gate returns a release channel and an idempotent closer, so a blocking test
50// execution is always unblocked before Shutdown waits on it.
51func gate() (chan struct{}, func()) {
52 ch := make(chan struct{})
53 var once sync.Once
54 return ch, func() { once.Do(func() { close(ch) }) }
55}
56
57func runningDeployOp() domain.AppOperation {
58 return domain.AppOperation{Op: "op-1", Kind: "deploy", App: "blog", InputRevision: "rev-1", StartedAt: time.Now().UTC()}
59}
60
61func ownedStart(op domain.AppOperation) *deployment.StartDeployResult {
62 return &deployment.StartDeployResult{
63 Claim: deployment.DeployClaim{App: op.App, Op: op.Op, Service: "web", Revision: "rev-1", Journal: op},
64 Owned: true,
65 }
66}
67
68func replayStart(op domain.AppOperation) *deployment.StartDeployResult {
69 return &deployment.StartDeployResult{
70 Claim: deployment.DeployClaim{App: op.App, Op: op.Op, Service: "web", Revision: "rev-1", Journal: op},
71 Owned: false,
72 }
73}
74
75// TestAppServiceImpl_Deploy_ReturnsRunningJournalAfterPersistence proves the
76// request path is two-phase: StartDeploy has already journaled the operation
77// when the background execution starts, and Deploy returns that non-terminal
78// journal without waiting for the replacement.
79func TestAppServiceImpl_Deploy_ReturnsRunningJournalAfterPersistence(t *testing.T) {
80 ctx := context.Background()
81 store := newMockAppState(t)
82 deploy := newMockDeployEngine(t)
83 svc := apps.NewAppServiceImpl(store, deploy, newMockSecretWriter(t), zerowrap.Default())
84 t.Cleanup(func() { require.NoError(t, svc.Shutdown(context.Background())) })
85
86 op := runningDeployOp()
87 store.EXPECT().LoadOperation(mock.Anything, "blog", "op-1").Return(op, nil)
88
89 startInputs := make(chan deployment.DeployInput, 1)
90 deploy.startDeployFn = func(_ context.Context, input deployment.DeployInput) (*deployment.StartDeployResult, error) {
91 startInputs <- input
92 return ownedStart(op), nil
93 }
94
95 persistedBeforeExecution := make(chan error, 1)
96 release, releaseExecution := gate()
97 t.Cleanup(releaseExecution)
98 deploy.executeDeployFn = func(execCtx context.Context, claim deployment.DeployClaim) (*deployment.DeployResult, error) {
99 // The claimed journal must be durable and queryable before any effect.
100 stored, err := store.LoadOperation(execCtx, claim.App, claim.Op)
101 if err == nil && (stored.Op != claim.Op || stored.Terminal()) {
102 err = errors.New("claimed journal is not the persisted running operation")
103 }
104 persistedBeforeExecution <- err
105 <-release
106 return &deployment.DeployResult{Op: claim.Op, App: claim.App}, nil
107 }
108
109 res := awaitDeploy(t, callDeploy(t, svc, ctx, "key-1"))
110 require.NoError(t, res.err)
111 require.NotNil(t, res.op)
112 assert.Equal(t, "op-1", res.op.Op)
113 assert.False(t, res.op.Terminal(), "Deploy returns the still-running journal")
114 assert.Equal(t, deployment.DeployInput{App: "blog", Revision: "rev-1", Service: "web", Op: "key-1"}, <-startInputs)
115
116 select {
117 case err := <-persistedBeforeExecution:
118 require.NoError(t, err, "execution must start only after the journal is persisted")
119 case <-time.After(3 * time.Second):
120 t.Fatal("background execution never started")
121 }
122 releaseExecution()
123}
124
125// TestAppServiceImpl_Deploy_RequestCancellationDoesNotCancelExecution proves
126// the execution runs on the daemon context, never on the request context.
127func TestAppServiceImpl_Deploy_RequestCancellationDoesNotCancelExecution(t *testing.T) {
128 requestCtx, cancelRequest := context.WithCancel(context.Background())
129 defer cancelRequest()
130 store := newMockAppState(t)
131 deploy := newMockDeployEngine(t)
132 svc := apps.NewAppServiceImpl(store, deploy, newMockSecretWriter(t), zerowrap.Default())
133 t.Cleanup(func() { require.NoError(t, svc.Shutdown(context.Background())) })
134
135 op := runningDeployOp()
136 store.EXPECT().LoadOperation(mock.Anything, "blog", "op-1").Return(op, nil)
137 deploy.startDeployFn = func(context.Context, deployment.DeployInput) (*deployment.StartDeployResult, error) {
138 return ownedStart(op), nil
139 }
140
141 execCtxCh := make(chan context.Context, 1)
142 release, releaseExecution := gate()
143 t.Cleanup(releaseExecution)
144 deploy.executeDeployFn = func(execCtx context.Context, _ deployment.DeployClaim) (*deployment.DeployResult, error) {
145 execCtxCh <- execCtx
146 <-release
147 return nil, nil
148 }
149
150 res := awaitDeploy(t, callDeploy(t, svc, requestCtx, "key-1"))
151 require.NoError(t, res.err)
152 require.NotNil(t, res.op)
153
154 execCtx := <-execCtxCh
155 cancelRequest()
156 select {
157 case <-requestCtx.Done():
158 default:
159 t.Fatal("request context was not cancelled")
160 }
161 require.NotEqual(t, requestCtx, execCtx, "execution must not run on the request context")
162 select {
163 case <-execCtx.Done():
164 t.Fatal("request cancellation cancelled the in-flight execution")
165 case <-time.After(50 * time.Millisecond):
166 }
167 releaseExecution()
168}
169
170// TestAppServiceImpl_Deploy_DuplicateKeyNeverSpawnsSecondExecution proves the
171// claimed key is the only execution: a replay returns the stored journal, an
172// in-flight replay reports a conflict, and ExecuteDeploy runs exactly once.
173func TestAppServiceImpl_Deploy_DuplicateKeyNeverSpawnsSecondExecution(t *testing.T) {
174 ctx := context.Background()
175 store := newMockAppState(t)
176 deploy := newMockDeployEngine(t)
177 svc := apps.NewAppServiceImpl(store, deploy, newMockSecretWriter(t), zerowrap.Default())
178 t.Cleanup(func() { require.NoError(t, svc.Shutdown(context.Background())) })
179
180 op := runningDeployOp()
181 store.EXPECT().LoadOperation(mock.Anything, "blog", "op-1").Return(op, nil)
182
183 var startCalls atomic.Int32
184 deploy.startDeployFn = func(context.Context, deployment.DeployInput) (*deployment.StartDeployResult, error) {
185 if startCalls.Add(1) == 1 {
186 return ownedStart(op), nil
187 }
188 return replayStart(op), nil
189 }
190 var execCalls atomic.Int32
191 execStarted := make(chan struct{})
192 release, releaseExecution := gate()
193 t.Cleanup(releaseExecution)
194 deploy.executeDeployFn = func(context.Context, deployment.DeployClaim) (*deployment.DeployResult, error) {
195 execCalls.Add(1)
196 close(execStarted)
197 <-release
198 return &deployment.DeployResult{Op: "op-1"}, nil
199 }
200
201 first := awaitDeploy(t, callDeploy(t, svc, ctx, "key-1"))
202 require.NoError(t, first.err)
203 require.NotNil(t, first.op)
204 select {
205 case <-execStarted:
206 case <-time.After(3 * time.Second):
207 t.Fatal("first execution never started")
208 }
209
210 second := awaitDeploy(t, callDeploy(t, svc, ctx, "key-1"))
211 require.ErrorIs(t, second.err, domain.ErrAppStateConflict, "an in-flight replay is never a second execution")
212 require.NotNil(t, second.op)
213 assert.Equal(t, "op-1", second.op.Op)
214 assert.Equal(t, int32(1), execCalls.Load(), "a duplicate key never spawns a second execution")
215 releaseExecution()
216}
217
218// TestAppServiceImpl_Deploy_TerminalFailureRemainsQueryable proves a failed
219// execution is recorded once, stays readable by key, and its replay reports
220// the stored failure without executing again.
221func TestAppServiceImpl_Deploy_TerminalFailureRemainsQueryable(t *testing.T) {
222 ctx := context.Background()
223 store := newMockAppState(t)
224 deploy := newMockDeployEngine(t)
225 svc := apps.NewAppServiceImpl(store, deploy, newMockSecretWriter(t), zerowrap.Default())
226
227 failed := runningDeployOp()
228 failed.Outcome = domain.AppOutcomeFailed
229 store.EXPECT().LoadOperation(mock.Anything, "blog", "op-1").Return(failed, nil)
230
231 var startCalls atomic.Int32
232 deploy.startDeployFn = func(context.Context, deployment.DeployInput) (*deployment.StartDeployResult, error) {
233 if startCalls.Add(1) == 1 {
234 return ownedStart(runningDeployOp()), nil
235 }
236 return replayStart(failed), nil
237 }
238 var execCalls atomic.Int32
239 executed := make(chan struct{}, 1)
240 deploy.executeDeployFn = func(context.Context, deployment.DeployClaim) (*deployment.DeployResult, error) {
241 execCalls.Add(1)
242 executed <- struct{}{}
243 return nil, errors.New("replacement failed")
244 }
245
246 res := awaitDeploy(t, callDeploy(t, svc, ctx, "key-1"))
247 require.NoError(t, res.err, "a failed execution is reported through the journal, not the start call")
248 require.NotNil(t, res.op)
249 assert.True(t, res.op.Terminal())
250 // Join the background execution before asserting the journal, so the
251 // count is deterministic rather than a race with the goroutine.
252 <-executed
253 assert.Equal(t, int32(1), execCalls.Load())
254
255 // A terminal replay stays queryable and never executes again, even
256 // though it is answered as a conflict.
257 retry := awaitDeploy(t, callDeploy(t, svc, ctx, "key-1"))
258 require.ErrorIs(t, retry.err, domain.ErrAppStateConflict)
259 require.NotNil(t, retry.op)
260 assert.Equal(t, domain.AppOutcomeFailed, retry.op.Outcome)
261 assert.Equal(t, int32(1), execCalls.Load(), "a terminal replay never executes again")
262
263 require.NoError(t, svc.Shutdown(context.Background()))
264
265 got, err := svc.OperationByKey(ctx, "blog", "op-1")
266 require.NoError(t, err)
267 assert.Equal(t, domain.AppOutcomeFailed, got.Outcome, "terminal failure remains queryable by key")
268}
269
270// TestAppServiceImpl_Deploy_ShutdownLeavesRecoverableNonTerminalJournal proves
271// graceful shutdown cancels the execution and leaves the journal persisted
272// non-terminal for reconciliation, instead of losing or finalizing it.
273func TestAppServiceImpl_Deploy_ShutdownLeavesRecoverableNonTerminalJournal(t *testing.T) {
274 ctx := context.Background()
275 store := newMockAppState(t)
276 deploy := newMockDeployEngine(t)
277 svc := apps.NewAppServiceImpl(store, deploy, newMockSecretWriter(t), zerowrap.Default()).
278 WithDaemonContext(context.Background())
279
280 op := runningDeployOp()
281 store.EXPECT().LoadOperation(mock.Anything, "blog", "op-1").Return(op, nil)
282 deploy.startDeployFn = func(context.Context, deployment.DeployInput) (*deployment.StartDeployResult, error) {
283 return ownedStart(op), nil
284 }
285
286 execCtxCh := make(chan context.Context, 1)
287 deploy.executeDeployFn = func(execCtx context.Context, _ deployment.DeployClaim) (*deployment.DeployResult, error) {
288 execCtxCh <- execCtx
289 <-execCtx.Done()
290 return nil, execCtx.Err()
291 }
292
293 res := awaitDeploy(t, callDeploy(t, svc, ctx, "key-1"))
294 require.NoError(t, res.err)
295 require.NotNil(t, res.op)
296
297 execCtx := <-execCtxCh
298 require.NoError(t, svc.Shutdown(context.Background()), "Shutdown waits for the execution to unwind")
299 require.ErrorIs(t, execCtx.Err(), context.Canceled, "shutdown cancels the in-flight execution")
300
301 got, err := svc.OperationByKey(ctx, "blog", "op-1")
302 require.NoError(t, err)
303 assert.False(t, got.Terminal(), "shutdown must leave the non-terminal journal for reconciliation")
304 assert.Equal(t, "op-1", got.Op)
305}
306
307// TestAppServiceImpl_Deploy_RefusesAfterShutdownBeforeClaim proves a deploy
308// that arrives after shutdown began is refused with a state conflict before
309// the engine can claim it, so neither a durable 202-running journal nor a
310// workload effect can be produced. The engine mock has no expectations, so
311// any StartDeploy or ExecuteDeploy call fails the test.
312func TestAppServiceImpl_Deploy_RefusesAfterShutdownBeforeClaim(t *testing.T) {
313 ctx := context.Background()
314 store := newMockAppState(t)
315 deploy := newMockDeployEngine(t)
316 svc := apps.NewAppServiceImpl(store, deploy, newMockSecretWriter(t), zerowrap.Default())
317
318 require.NoError(t, svc.Shutdown(context.Background()))
319
320 res := awaitDeploy(t, callDeploy(t, svc, ctx, "key-1"))
321 require.ErrorIs(t, res.err, domain.ErrAppStateConflict, "a deploy after shutdown is refused with a state conflict")
322 require.Nil(t, res.op, "a refused deploy never produces a journal to answer 202")
323}
324
325// TestAppServiceImpl_Deploy_ShutdownRacingClaimSettlesWithoutExecuting proves
326// the claim/schedule hand-off is closed against shutdown: once Shutdown starts
327// after StartDeploy claimed the operation but before it was scheduled, the
328// owned claim is settled without starting, Shutdown waits for that settle, and
329// the request answers a conflict instead of a false 202-running orphan.
330func TestAppServiceImpl_Deploy_ShutdownRacingClaimSettlesWithoutExecuting(t *testing.T) {
331 ctx := context.Background()
332 store := newMockAppState(t)
333 deploy := newMockDeployEngine(t)
334 svc := apps.NewAppServiceImpl(store, deploy, newMockSecretWriter(t), zerowrap.Default())
335
336 op := runningDeployOp()
337 store.EXPECT().LoadOperation(mock.Anything, "blog", "op-1").Return(op, nil)
338
339 claimed := make(chan struct{})
340 releaseClaim := make(chan struct{})
341 deploy.startDeployFn = func(context.Context, deployment.DeployInput) (*deployment.StartDeployResult, error) {
342 close(claimed)
343 <-releaseClaim
344 return ownedStart(op), nil
345 }
346 var execCalls atomic.Int32
347 deploy.executeDeployFn = func(context.Context, deployment.DeployClaim) (*deployment.DeployResult, error) {
348 execCalls.Add(1)
349 return nil, nil
350 }
351 abandoned := make(chan deployment.DeployClaim, 1)
352 deploy.abandonDeployFn = func(_ context.Context, claim deployment.DeployClaim) error {
353 abandoned <- claim
354 return nil
355 }
356
357 resCh := callDeploy(t, svc, ctx, "key-1")
358 <-claimed
359
360 shutdownDone := make(chan error, 1)
361 go func() { shutdownDone <- svc.Shutdown(context.Background()) }()
362
363 // Shutdown must wait for the hand-off rather than return while the claim
364 // is still being made.
365 select {
366 case <-shutdownDone:
367 t.Fatal("Shutdown returned before the in-flight claim hand-off settled")
368 case <-time.After(50 * time.Millisecond):
369 }
370
371 // The claim completes now; shutdown is already set, so it must be settled
372 // rather than executed.
373 close(releaseClaim)
374
375 select {
376 case claim := <-abandoned:
377 assert.Equal(t, "op-1", claim.Op)
378 case <-time.After(3 * time.Second):
379 t.Fatal("the raced claim was not settled")
380 }
381 require.NoError(t, <-shutdownDone, "Shutdown waits for the settle and returns cleanly")
382
383 res := awaitDeploy(t, resCh)
384 require.ErrorIs(t, res.err, domain.ErrAppStateConflict, "a settled claim answers a conflict, never 202-running")
385 require.NotNil(t, res.op)
386 assert.Equal(t, int32(0), execCalls.Load(), "a claim settled during shutdown never executes")
387}