1package deployment
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "maps"
8 "sort"
9 "strings"
10 "time"
11
12 "github.com/bnema/zerowrap"
13
14 "github.com/bnema/gordon/internal/domain"
15)
16
17// LifecycleResult carries the terminal outcome of a lifecycle verb.
18type LifecycleResult struct {
19 Op string
20 App string
21 Verb string
22 Outcome string
23 Services map[string]ServiceResult
24 // Warnings are operation-level leftovers that belong to no single
25 // service, such as a private network that could not be verified or
26 // removed.
27 Warnings []CleanupWarning
28}
29
30// Stop persists the durable stopped intent first, then stops and removes
31// exact active containers by ID. Volumes and secrets are retained.
32func (s *Service) Stop(ctx context.Context, app, opID string) (*LifecycleResult, error) {
33 release, err := s.acquireAppContext(ctx, app)
34 if err != nil {
35 return nil, err
36 }
37 defer release()
38 return s.stopLocked(ctx, app, opID)
39}
40
41func (s *Service) stopLocked(ctx context.Context, app, opID string) (*LifecycleResult, error) {
42 ctx = lifecycleCtx(ctx, "Stop", app)
43 log := zerowrap.FromCtx(ctx)
44 if err := s.deps.State.Recover(ctx); err != nil {
45 return nil, fmt.Errorf("deployment: recover before stop: %w", err)
46 }
47 if err := s.reconcileInterruptedDeploy(ctx, app); err != nil {
48 return nil, err
49 }
50 op := domain.AppOperation{
51 Kind: "stop", App: app,
52 StartedAt: time.Now().UTC(),
53 Request: domain.AppOperationRequestFor("stop", app, "", ""),
54 }
55 op, owned, err := s.claimOperation(ctx, opID, op, []domain.AppOperationStep{{ID: "intent.stopped", State: domain.AppStepPending}})
56 if err != nil {
57 return nil, err
58 }
59 if !owned {
60 return replayedLifecycleResult(op), replayError(op)
61 }
62 if err := s.deps.State.SaveIntent(ctx, domain.AppStopIntent{
63 App: app, Stopped: true, UpdatedBy: op.Op, UpdatedAt: time.Now().UTC(),
64 }); err != nil {
65 intentErr := fmt.Errorf("deployment: persist stopped intent: %w", err)
66 s.failOperation(ctx, &op, intentErr, "error")
67 return nil, intentErr
68 }
69 op.Steps[0].State = domain.AppStepSucceeded
70 active, _, err := s.deps.State.LoadActive(ctx, app)
71 if err != nil {
72 activeErr := fmt.Errorf("deployment: load active: %w", err)
73 s.failOperation(ctx, &op, activeErr, "error")
74 return nil, activeErr
75 }
76 result := &LifecycleResult{Op: op.Op, App: app, Verb: "stop", Services: map[string]ServiceResult{}}
77 var failures []string
78 for _, name := range sortedServiceNames(active) {
79 container := active.Services[name].Container
80 step, warnings, err := s.stopService(ctx, app, name, active.Services[name])
81 op.Steps = append(op.Steps, step)
82 if err != nil {
83 result.Services[name] = ServiceResult{Result: "failed", Before: container, Error: err.Error(), CleanupWarnings: warnings}
84 failures = append(failures, name)
85 continue
86 }
87 result.Services[name] = ServiceResult{Result: "deployed", Before: container, After: "", CleanupWarnings: warnings}
88 }
89 // A superseded predecessor an overlap replacement left running is not in
90 // ACTIVE, so the loop above never saw it: a stopped app must not keep it
91 // running. One that cannot be retired keeps its marker and fails the stop.
92 pendingFailed, pendingErr := s.retirePendingForStop(ctx, app, result)
93 failures = append(failures, pendingFailed...)
94 op.Outcome = ComputeOutcome(result.Services)
95 if pendingErr != nil {
96 // An app-level cleanup failure belongs to no service, so the
97 // per-service results alone would journal a success and a replay of
98 // this key would read it. It is a failed step and a failed outcome.
99 op.Steps = append(op.Steps, domain.AppOperationStep{
100 ID: "pending.retire", State: domain.AppStepFailed, Error: pendingErr.Error(),
101 })
102 op.Outcome = domain.AppOutcomeFailed
103 }
104 op.Warnings = journalWarnings(append(collectCleanupWarnings(result.Services), result.Warnings...))
105 if err := s.deps.State.SaveOperation(ctx, op); err != nil {
106 log.Warn().Err(err).Msg("deployment: failed to record stop outcome")
107 }
108 if err := s.refreshTraffic(ctx, app); err != nil {
109 s.failTrafficPublication(ctx, &op, err)
110 return result, err
111 }
112 if len(failures) > 0 {
113 return result, fmt.Errorf("deployment: stop failed for %s: %w", strings.Join(failures, ", "), domain.ErrAppStateConflict)
114 }
115 return result, nil
116}
117
118// stopService withdraws one service, then retires its exact container
119// with the effective grace. A withdrawal or runtime failure is reported,
120// never treated as a successful stop. The confirmed disappearance
121// releases the container's backend claims and its recovery inhibition:
122// a stopped app is never revived by recovery.
123func (s *Service) stopService(ctx context.Context, app, name string, eff domain.AppEffectiveService) (domain.AppOperationStep, []CleanupWarning, error) {
124 container := eff.Container
125 step := domain.AppOperationStep{ID: "service." + name + ".stop", State: domain.AppStepPending, Before: container}
126 fail := func(err error) (domain.AppOperationStep, []CleanupWarning, error) {
127 step.State = domain.AppStepFailed
128 step.Error = err.Error()
129 return step, nil, err
130 }
131 if err := s.withdrawForRecovery(ctx, app, name); err != nil {
132 return fail(err)
133 }
134 if container != "" {
135 retired := s.retireContainer(ctx, app, retireOptions{
136 Service: name, Grace: serviceStopGrace(eff), ClearInhibition: true,
137 }, container)
138 if !retired.Gone {
139 return fail(fmt.Errorf("deployment: stop %s/%s container %s: %s", app, name, container, cleanupDetail(retired)))
140 }
141 step.After = container
142 step.State = domain.AppStepSucceeded
143 return step, retired.Warnings, nil
144 }
145 step.State = domain.AppStepSucceeded
146 return step, nil, nil
147}
148
149// Start clears the stopped intent and ensures running from active records.
150// It starts missing/stopped instances without duplicating running ones
151// and never activates pending desired revisions.
152func (s *Service) Start(ctx context.Context, app, opID string) (*LifecycleResult, error) {
153 release, err := s.acquireAppContext(ctx, app)
154 if err != nil {
155 return nil, err
156 }
157 defer release()
158 return s.startLocked(ctx, app, opID)
159}
160
161func (s *Service) startLocked(ctx context.Context, app, opID string) (*LifecycleResult, error) {
162 ctx = lifecycleCtx(ctx, "Start", app)
163 log := zerowrap.FromCtx(ctx)
164 if err := s.deps.State.Recover(ctx); err != nil {
165 return nil, fmt.Errorf("deployment: recover before start: %w", err)
166 }
167 if err := s.reconcileInterruptedDeploy(ctx, app); err != nil {
168 return nil, err
169 }
170 op := domain.AppOperation{
171 Kind: "start", App: app,
172 StartedAt: time.Now().UTC(),
173 Request: domain.AppOperationRequestFor("start", app, "", ""),
174 }
175 op, owned, err := s.claimOperation(ctx, opID, op, []domain.AppOperationStep{{ID: "intent.running", State: domain.AppStepPending}})
176 if err != nil {
177 return nil, err
178 }
179 if !owned {
180 return replayedLifecycleResult(op), replayError(op)
181 }
182 if err := s.deps.State.SaveIntent(ctx, domain.AppStopIntent{
183 App: app, Stopped: false, UpdatedBy: op.Op, UpdatedAt: time.Now().UTC(),
184 }); err != nil {
185 intentErr := fmt.Errorf("deployment: clear stopped intent: %w", err)
186 s.failOperation(ctx, &op, intentErr, "error")
187 return nil, intentErr
188 }
189 op.Steps[0].State = domain.AppStepSucceeded
190 active, _, err := s.deps.State.LoadActive(ctx, app)
191 if err != nil {
192 activeErr := fmt.Errorf("deployment: load active: %w", err)
193 s.failOperation(ctx, &op, activeErr, "error")
194 return nil, activeErr
195 }
196 result := &LifecycleResult{Op: op.Op, App: app, Verb: "start", Services: map[string]ServiceResult{}}
197 for _, name := range sortedServiceNames(active) {
198 eff := active.Services[name]
199 // The step is journaled before the runtime work, so a candidate
200 // created by a redeploy is durably recorded as soon as it exists.
201 op.Steps = append(op.Steps, domain.AppOperationStep{
202 ID: "service." + name + ".start", State: domain.AppStepPending, Before: eff.Container, Service: name,
203 })
204 s.ensureServiceRunning(ctx, app, opID, name, eff, &op, len(op.Steps)-1, result)
205 }
206 op.Outcome = ComputeOutcome(result.Services)
207 op.Warnings = journalWarnings(collectCleanupWarnings(result.Services))
208 if err := s.deps.State.SaveOperation(ctx, op); err != nil {
209 log.Warn().Err(err).Msg("deployment: failed to record start outcome")
210 }
211 if err := s.refreshTraffic(ctx, app); err != nil {
212 s.failTrafficPublication(ctx, &op, err)
213 return result, err
214 }
215 // Routing accepted the verified generations: record their executions.
216 for name, svcResult := range result.Services {
217 s.recordVerifiedExecution(app, name, svcResult)
218 }
219 return result, nil
220}
221
222// Restart restarts from pinned digests without re-resolution. Empty
223// service means all services in sorted order. Every service restarts in
224// place: traffic is withdrawn, the same pinned container is restarted, its
225// readiness probe is checked, and traffic is republished. No second
226// container is created and the pinned digest never changes.
227func (s *Service) Restart(ctx context.Context, app, service, opID string) (*LifecycleResult, error) {
228 release, err := s.acquireAppContext(ctx, app)
229 if err != nil {
230 return nil, err
231 }
232 defer release()
233 return s.restartLocked(ctx, app, service, opID)
234}
235
236func (s *Service) restartLocked(ctx context.Context, app, service, opID string) (*LifecycleResult, error) {
237 ctx = lifecycleCtx(ctx, "Restart", app)
238 if err := s.deps.State.Recover(ctx); err != nil {
239 return nil, fmt.Errorf("deployment: recover before restart: %w", err)
240 }
241 if err := s.reconcileInterruptedDeploy(ctx, app); err != nil {
242 return nil, err
243 }
244 active, ok, err := s.deps.State.LoadActive(ctx, app)
245 if err != nil {
246 return nil, fmt.Errorf("deployment: load active: %w", err)
247 }
248 if !ok {
249 return nil, fmt.Errorf("deployment: app %q was never deployed: %w", app, domain.ErrAppStateConflict)
250 }
251 names := sortedServiceNames(active)
252 if service != "" {
253 if _, ok := active.Services[service]; !ok {
254 return nil, fmt.Errorf("deployment: service %q not active: %w", service, domain.ErrAppStateConflict)
255 }
256 names = []string{service}
257 }
258 op := domain.AppOperation{
259 Kind: "restart", App: app,
260 StartedAt: time.Now().UTC(),
261 Request: domain.AppOperationRequestFor("restart", app, "", service),
262 }
263 op, owned, err := s.claimOperation(ctx, opID, op, nil)
264 if err != nil {
265 return nil, err
266 }
267 if !owned {
268 return replayedLifecycleResult(op), replayError(op)
269 }
270 result := &LifecycleResult{Op: op.Op, App: app, Verb: "restart", Services: map[string]ServiceResult{}}
271 var failures []string
272 for _, name := range names {
273 eff := active.Services[name]
274 // The step is journaled before the runtime work, so a candidate
275 // created by a rebuild of a missing generation is durably recorded.
276 op.Steps = append(op.Steps, domain.AppOperationStep{
277 ID: "service." + name + ".restart", State: domain.AppStepPending, Before: eff.Container, Service: name,
278 })
279 svcResult := s.restartOneService(ctx, app, op.Op, name, eff, &op, len(op.Steps)-1)
280 result.Services[name] = svcResult
281 if svcResult.Result != "deployed" {
282 failures = append(failures, name)
283 }
284 }
285 op.Outcome = ComputeOutcome(result.Services)
286 op.Warnings = journalWarnings(collectCleanupWarnings(result.Services))
287 if err := s.deps.State.SaveOperation(ctx, op); err != nil {
288 return result, fmt.Errorf("deployment: persist restart journal: %w", err)
289 }
290 if err := s.refreshTraffic(ctx, app); err != nil {
291 s.failTrafficPublication(ctx, &op, err)
292 return result, err
293 }
294 // Routing accepted the restarted generations: their executions are
295 // verified. The app lock is held, so no periodic pass ran in between.
296 for name, svcResult := range result.Services {
297 s.recordVerifiedExecution(app, name, svcResult)
298 }
299 if len(failures) > 0 {
300 return result, fmt.Errorf("deployment: restart failed for %s: %w", strings.Join(failures, ", "), domain.ErrAppStateConflict)
301 }
302 return result, nil
303}
304
305// restartOneService withdraws one service, restarts its exact container,
306// re-inspects its binds, and verifies readiness before it may be published
307// again. Every failure leaves the service withdrawn with its recorded binds
308// cleared, so a failed restart is never served. A generation whose recorded
309// container no longer exists is rebuilt from the pinned ACTIVE digest, so a
310// failed replacement cannot leave restart as a permanent dead end.
311func (s *Service) restartOneService(ctx context.Context, app, opID, name string, eff domain.AppEffectiveService, op *domain.AppOperation, index int) ServiceResult {
312 step := &op.Steps[index]
313 result := ServiceResult{Result: "failed", Before: eff.Container, After: eff.Container}
314 fail := func(msg string) ServiceResult {
315 step.State = domain.AppStepFailed
316 step.Error = msg
317 result.Result = "failed"
318 result.Error = msg
319 return result
320 }
321 if eff.Container == "" {
322 result.After = ""
323 return fail("no active container")
324 }
325 // Fail closed before any runtime mutation if a bind's policy was
326 // revoked or became invalid since ACTIVE was published.
327 if _, err := s.resolveServiceBinds(app, eff.Spec); err != nil {
328 return fail(err.Error())
329 }
330 // A device grant revoked since ACTIVE was published must fail
331 // before any runtime mutation, under the same rule as binds. An
332 // in-place restart never rewrites device configuration.
333 if _, err := s.resolveServiceDevices(app, eff.Spec); err != nil {
334 return fail(err.Error())
335 }
336 if err := s.verifyReusableCapabilities(ctx, app, eff); err != nil {
337 return fail(err.Error())
338 }
339 if err := s.refuseInhibitedRestart(ctx, app, name, eff.Container); err != nil {
340 _ = s.withdrawForRecovery(ctx, app, name)
341 return fail(err.Error())
342 }
343 // Withdraw before restarting: a restarting or unverified generation
344 // must not keep receiving traffic.
345 if err := s.withdrawForRecovery(ctx, app, name); err != nil {
346 return fail(err.Error())
347 }
348 if err := s.deps.Runtime.RestartContainer(ctx, eff.Container, serviceStopGrace(eff)); err != nil {
349 if errors.Is(err, domain.ErrContainerNotFound) {
350 // The recorded generation is gone: rebuild and publish it from the
351 // pinned ACTIVE digest instead of leaving the service withdrawn
352 // until a deploy.
353 svcResult, rebuildErr := s.redeployPinned(ctx, app, opID, name, eff, s.journalCandidate(op, index))
354 if rebuildErr != nil {
355 svcResult.Error = rebuildErr.Error()
356 step.State = domain.AppStepFailed
357 step.Error = rebuildErr.Error()
358 step.Diagnostics = svcResult.Diagnostics
359 // The recorded container is proven gone: the inhibition the
360 // rebuild wrote for it protects nothing, and keeping it would
361 // refuse every later start, recovery pass, and restart.
362 if clearErr := s.clearRecoveryInhibition(ctx, app, name, eff.Container); clearErr != nil {
363 svcResult.CleanupWarnings = append(svcResult.CleanupWarnings, CleanupWarning{
364 Service: name, Leftover: eff.Container, Detail: "clear recovery inhibition: " + clearErr.Error(),
365 })
366 }
367 return svcResult
368 }
369 step.State = domain.AppStepSucceeded
370 step.After = svcResult.After
371 return svcResult
372 }
373 s.clearServiceBinds(ctx, app, name)
374 return fail(err.Error())
375 }
376 // The restarted execution is observed before readiness runs.
377 startedAt := s.executionStart(ctx, eff.Container)
378 // Ephemeral loopback binds are not guaranteed stable across a runtime
379 // restart: re-inspect before probing so the proxy never dials a stale
380 // bind.
381 binds, udpBinds, err := s.refreshBackendBinds(ctx, app, name, eff, s.deps.Traffic != nil)
382 if err != nil {
383 s.clearServiceBinds(ctx, app, name)
384 return fail(err.Error())
385 }
386 if err := s.waitServiceReady(ctx, app, eff.Container, deploymentReadiness(eff.Spec), binds); err != nil {
387 s.clearServiceBinds(ctx, app, name)
388 return fail(err.Error())
389 }
390 step.State = domain.AppStepSucceeded
391 step.After = eff.Container
392 return ServiceResult{Result: "deployed", Before: eff.Container, After: eff.Container, BackendBinds: binds, UDPBackendBinds: udpBinds, VerifiedStartedAt: startedAt}
393}
394
395// Remove withdraws workloads by exact container ID; volumes, secrets, and
396// ownership records are retained under the old UUID. The public name is
397// freed for reuse; a new app never implicitly adopts retained resources.
398// Stopped intent and a durable recovery inhibition per removed container
399// are persisted BEFORE any runtime effect, so native restart policy can
400// never revive a removed generation into service.
401func (s *Service) Remove(ctx context.Context, app, opID string) (*LifecycleResult, error) {
402 release, err := s.acquireAppContext(ctx, app)
403 if err != nil {
404 return nil, err
405 }
406 defer release()
407 return s.removeLocked(ctx, app, opID)
408}
409
410func (s *Service) removeLocked(ctx context.Context, app, opID string) (*LifecycleResult, error) {
411 ctx = lifecycleCtx(ctx, "Remove", app)
412 log := zerowrap.FromCtx(ctx)
413 if err := s.deps.State.Recover(ctx); err != nil {
414 return nil, fmt.Errorf("deployment: recover before remove: %w", err)
415 }
416 if err := s.reconcileInterruptedDeploy(ctx, app); err != nil {
417 return nil, err
418 }
419 op := domain.AppOperation{
420 Kind: "remove", App: app,
421 StartedAt: time.Now().UTC(),
422 Request: domain.AppOperationRequestFor("remove", app, "", ""),
423 }
424 op, owned, err := s.claimOperation(ctx, opID, op, nil)
425 if err != nil {
426 return nil, err
427 }
428 if !owned {
429 return replayedLifecycleResult(op), replayError(op)
430 }
431 active, _, err := s.deps.State.LoadActive(ctx, app)
432 if err != nil {
433 activeErr := fmt.Errorf("deployment: load active: %w", err)
434 s.failOperation(ctx, &op, activeErr, "error")
435 return nil, activeErr
436 }
437 result := &LifecycleResult{Op: op.Op, App: app, Verb: "remove", Services: map[string]ServiceResult{}}
438 // Durable stopped intent and per-container inhibition precede every
439 // runtime effect: a crash between here and the container stop must
440 // never leave a generation that recovery could restart.
441 if err := s.deps.State.SaveIntent(ctx, domain.AppStopIntent{
442 App: app, Stopped: true, UpdatedBy: op.Op, UpdatedAt: time.Now().UTC(),
443 }); err != nil {
444 intentErr := fmt.Errorf("deployment: persist stopped intent before remove: %w", err)
445 s.failOperation(ctx, &op, intentErr, "error")
446 return nil, intentErr
447 }
448 steps, err := s.removeServiceContainers(ctx, app, op.Op, active, result)
449 if err != nil {
450 op.Steps = steps
451 s.failOperation(ctx, &op, err, "error")
452 return nil, err
453 }
454 // Every exact container is confirmed gone, so the app's private
455 // networks are reclaimed immediately, while the ownership record is
456 // still live and its UUID can be verified against the runtime names
457 // and labels. Shared networks stay, and retained volumes, secrets, and
458 // images are never touched.
459 ownership, err := s.deps.State.LoadOwnership(ctx, app)
460 if err != nil {
461 retireErr := fmt.Errorf("deployment: load ownership before reclamation: %w", err)
462 s.failOperation(ctx, &op, retireErr, "error")
463 return nil, retireErr
464 }
465 networkWarnings, err := s.reclaimPrivateNetworks(ctx, app, ownership)
466 if err != nil {
467 s.failOperation(ctx, &op, err, "error")
468 return nil, err
469 }
470 result.Warnings = append(result.Warnings, networkWarnings...)
471 op.Steps = steps
472 // End the incarnation atomically: the ownership record is archived
473 // with its resources retained under the old UUID, the app UUID is
474 // reset so a name reuse allocates a new incarnation, and desired,
475 // active, intent, and staged state are cleared. Without this a
476 // reapply would reuse the UUID and inherit the old secrets and
477 // volumes.
478 if err := s.deps.State.RetireApp(ctx, app); err != nil {
479 retireErr := fmt.Errorf("deployment: retire incarnation: %w", err)
480 s.failOperation(ctx, &op, retireErr, "error")
481 return nil, retireErr
482 }
483 op.Outcome = domain.AppOutcomeSuccess
484 op.Warnings = journalWarnings(append(collectCleanupWarnings(result.Services), result.Warnings...))
485 if err := s.deps.State.SaveOperation(ctx, op); err != nil {
486 log.Warn().Err(err).Msg("deployment: failed to record remove outcome")
487 }
488 if err := s.refreshTraffic(ctx, app); err != nil {
489 s.failTrafficPublication(ctx, &op, err)
490 return result, err
491 }
492 return result, nil
493}
494
495// replayedLifecycleResult projects a replayed journal into a lifecycle
496// result without touching any workload.
497func replayedLifecycleResult(op domain.AppOperation) *LifecycleResult {
498 return &LifecycleResult{Op: op.Op, App: op.App, Verb: op.Kind, Outcome: op.Outcome}
499}
500
501// refuseInhibitedRestart blocks a restart of a generation whose recovery
502// is durably inhibited: a replacement may already have written to the
503// volume this generation still owns, so reviving it could corrupt newer
504// data. Deploying a new revision supersedes the marker safely.
505func (s *Service) refuseInhibitedRestart(ctx context.Context, app, name, containerID string) error {
506 inhibited, err := s.recoveryInhibited(ctx, app, name, containerID)
507 if err != nil {
508 return err
509 }
510 if inhibited {
511 return fmt.Errorf("deployment: restart %q/%q refused: recovery inhibited for container %s", app, name, containerID)
512 }
513 return nil
514}
515
516// removeServiceContainers persists one inhibition per removed container
517// before stopping and removing it by exact ID. Volumes are never deleted.
518// A container that could not be confirmed gone fails the remove: ACTIVE,
519// stopped intent, and the inhibition stay in place so the periodic pass
520// keeps converging the survivor instead of losing track of it.
521func (s *Service) removeServiceContainers(ctx context.Context, app, opID string, active domain.AppActive, result *LifecycleResult) ([]domain.AppOperationStep, error) {
522 var steps []domain.AppOperationStep
523 for _, name := range sortedServiceNames(active) {
524 container := active.Services[name].Container
525 step := domain.AppOperationStep{ID: "service." + name + ".remove", State: domain.AppStepPending, Before: container}
526 if container != "" {
527 if err := s.inhibitRecovery(ctx, app, name, container, "removed", opID); err != nil {
528 return nil, err
529 }
530 retired := s.retireContainer(ctx, app, retireOptions{
531 Service: name, Grace: serviceStopGrace(active.Services[name]), ClearInhibition: true,
532 }, container)
533 if !retired.Gone {
534 return nil, fmt.Errorf("deployment: remove %q/%q: container %s not confirmed gone: %s", app, name, container, cleanupDetail(retired))
535 }
536 result.Services[name] = ServiceResult{Result: domain.AppServiceRemoved, Before: container, After: "", CleanupWarnings: retired.Warnings}
537 } else {
538 result.Services[name] = ServiceResult{Result: domain.AppServiceRemoved, Before: container, After: ""}
539 }
540 step.State = domain.AppStepSucceeded
541 steps = append(steps, step)
542 }
543 if err := s.retireAllPending(ctx, app); err != nil {
544 return nil, err
545 }
546 return steps, nil
547}
548
549// retireAllPending retires every superseded container an overlap replacement
550// left running beside its successor. Ending the incarnation deletes the
551// retirement markers with the rest of the app state, so a container that is
552// not confirmed gone here would be orphaned and untracked: a failure keeps the
553// markers and fails the caller. Every marker is attempted, and a marker whose
554// claims could not be released is a failure too, so it is retried instead of
555// leaking its claims.
556func (s *Service) retireAllPending(ctx context.Context, app string) error {
557 pending, err := s.pendingRetirements(ctx, app)
558 if err != nil {
559 return err
560 }
561 var failures []error
562 for _, marker := range pending {
563 retired := s.retireContainer(ctx, app, retireOptions{
564 Service: marker.Service, Grace: marker.StopGrace, ClearInhibition: true, RetryOnLeak: true,
565 }, marker.ContainerID)
566 if !retired.Settled {
567 failures = append(failures, fmt.Errorf("deployment: remove %q/%q: superseded container %s not retired: %s", app, marker.Service, marker.ContainerID, cleanupDetail(retired)))
568 }
569 }
570 return errors.Join(failures...)
571}
572
573// pendingRetirements loads the retirement-pending markers of one app.
574func (s *Service) pendingRetirements(ctx context.Context, app string) ([]domain.AppRecoveryInhibition, error) {
575 inhibitions, err := s.deps.State.LoadRecoveryInhibitions(ctx, app)
576 if err != nil {
577 return nil, fmt.Errorf("deployment: load recovery inhibitions: %w", err)
578 }
579 _, pending := splitRecoveryInhibitions(app, inhibitions)
580 return pending, nil
581}
582
583// retirePendingForStop retires the pending superseded containers of a stopped
584// app, recording a leftover or a failure on the service they belong to. It
585// returns the services whose predecessor is not confirmed retired, and an
586// app-level error when the markers themselves could not be loaded: nothing
587// can then be said about any predecessor, which the caller must journal as a
588// failed operation.
589func (s *Service) retirePendingForStop(ctx context.Context, app string, result *LifecycleResult) ([]string, error) {
590 pending, err := s.pendingRetirements(ctx, app)
591 if err != nil {
592 result.Warnings = append(result.Warnings, CleanupWarning{Detail: "pending retirements unknown: " + err.Error()})
593 return []string{"pending retirements"}, err
594 }
595 var failed []string
596 for _, marker := range pending {
597 retired := s.retireContainer(ctx, app, retireOptions{
598 Service: marker.Service, Grace: marker.StopGrace, ClearInhibition: true, RetryOnLeak: true,
599 }, marker.ContainerID)
600 svc := result.Services[marker.Service]
601 svc.CleanupWarnings = append(svc.CleanupWarnings, retired.Warnings...)
602 if !retired.Settled {
603 svc.Result = "failed"
604 svc.Error = "superseded container " + marker.ContainerID + " not retired: " + cleanupDetail(retired)
605 failed = append(failed, marker.Service)
606 }
607 result.Services[marker.Service] = svc
608 }
609 return failed, nil
610}
611
612// ReconcileBoot reconciles apps intended to run after a daemon boot:
613// every non-stopped app with ACTIVE services is verified through Start,
614// including all-running apps: ephemeral loopback binds may shift across
615// any runtime restart while the daemon is away, so recorded binds are
616// re-inspected (never trusted) before the proxy can dial them.
617// Recovery first; no duplication of running instances; no activation of
618// pending desired revisions. Explicitly stopped apps stay stopped.
619// A failing app never blocks the verification of the following apps;
620// failures are aggregated and every unverified bind is withdrawn from
621// ACTIVE (see failServiceStep), so the post-boot host index rebuild
622// projects fail-closed for all of them.
623func (s *Service) ReconcileBoot(ctx context.Context) error {
624 ctx = lifecycleCtx(ctx, "ReconcileBoot", "boot")
625 if err := s.deps.State.Recover(ctx); err != nil {
626 return fmt.Errorf("deployment: recover before boot reconcile: %w", err)
627 }
628 apps, err := s.deps.State.ListApps(ctx)
629 if err != nil {
630 return fmt.Errorf("deployment: list apps: %w", err)
631 }
632 var failures []error
633 for _, app := range apps {
634 if err := s.reconcileBootApp(ctx, app); err != nil {
635 failures = append(failures, err)
636 }
637 }
638 return errors.Join(failures...)
639}
640
641// reconcileBootApp verifies one app at boot under the app lock. Explicitly
642// stopped apps and apps without ACTIVE services are skipped without error.
643// Stopped intent is converged: a container revived by native restart
644// policy while the daemon was away is stopped again, never restarted.
645func (s *Service) reconcileBootApp(ctx context.Context, app string) error {
646 release, err := s.acquireAppContext(ctx, app)
647 if err != nil {
648 return err
649 }
650 defer release()
651 intent, err := s.deps.State.LoadIntent(ctx, app)
652 if err != nil {
653 return fmt.Errorf("deployment: load intent: %w", err)
654 }
655 active, ok, err := s.deps.State.LoadActive(ctx, app)
656 if err != nil {
657 return fmt.Errorf("deployment: load active: %w", err)
658 }
659 if !ok || len(active.Services) == 0 {
660 // An interrupted operation is reconciled even when ACTIVE carries no
661 // service: a first deployment can be interrupted between creating its
662 // candidate and publishing it. No start follows here, so this is the
663 // only reconciliation of the boot pass.
664 return s.reconcileInterruptedDeploy(ctx, app)
665 }
666 if intent.Stopped {
667 // A stopped app does not start, so its unfinished operation is
668 // reconciled here instead.
669 if err := s.reconcileInterruptedDeploy(ctx, app); err != nil {
670 return err
671 }
672 for _, name := range sortedServiceNames(active) {
673 if err := s.convergeStoppedService(ctx, app, name, active.Services[name]); err != nil {
674 return fmt.Errorf("deployment: boot stopped convergence %q/%q: %w", app, name, err)
675 }
676 }
677 return nil
678 }
679 // startLocked reconciles the app's unfinished operation before it starts
680 // anything, so the boot pass reconciles exactly once.
681 result, err := s.startLocked(ctx, app, "")
682 if err != nil {
683 return fmt.Errorf("deployment: boot start %q: %w", app, err)
684 }
685 for name, svc := range result.Services {
686 if svc.Result == "failed" {
687 return fmt.Errorf("deployment: boot verify %q/%q: %s", app, name, svc.Error)
688 }
689 }
690 return nil
691}
692
693// ensureServiceRunning starts one service when needed. Running
694// containers are bind-verified before success: ephemeral loopback binds
695// may shift across any runtime restart (stop/start, daemon absence,
696// host reboot), so the recorded binds are re-inspected and persisted
697// before the proxy can dial them. Verification failure fails the step;
698// a stale recorded bind is never served. A recorded container that no
699// longer exists is rebuilt from the revision pinned in ACTIVE.
700func (s *Service) ensureServiceRunning(ctx context.Context, app, opID, name string, eff domain.AppEffectiveService, op *domain.AppOperation, index int, result *LifecycleResult) {
701 step := &op.Steps[index]
702 if eff.Container != "" {
703 // Fail closed before any runtime mutation if a bind's policy was
704 // revoked or became invalid since ACTIVE was published.
705 if _, err := s.resolveServiceBinds(app, eff.Spec); err != nil {
706 s.failServiceStep(ctx, app, name, eff, step, result, err.Error())
707 return
708 }
709 // Revoked device grants fail before any runtime mutation.
710 if _, err := s.resolveServiceDevices(app, eff.Spec); err != nil {
711 s.failServiceStep(ctx, app, name, eff, step, result, err.Error())
712 return
713 }
714 if err := s.verifyReusableCapabilities(ctx, app, eff); err != nil {
715 s.failServiceStep(ctx, app, name, eff, step, result, err.Error())
716 return
717 }
718 // Boot recovery refuses a generation whose recovery is durably
719 // inhibited: a replacement may already have written to its
720 // volume, so reviving this ID could corrupt the newer data.
721 inhibited, err := s.recoveryInhibited(ctx, app, name, eff.Container)
722 if err != nil {
723 s.failServiceStep(ctx, app, name, eff, step, result, err.Error())
724 return
725 }
726 if inhibited {
727 s.failServiceStep(ctx, app, name, eff, step, result,
728 fmt.Sprintf("recovery inhibited for container %s", eff.Container))
729 return
730 }
731 if ok, err := s.deps.Runtime.IsContainerRunning(ctx, eff.Container); err == nil && ok {
732 s.verifyRunningService(ctx, app, name, eff, step, result)
733 return
734 }
735 // Restart by exact container ID when the runtime still has it.
736 if err := s.deps.Runtime.StartContainer(ctx, eff.Container); err == nil {
737 s.verifyRunningService(ctx, app, name, eff, step, result)
738 return
739 }
740 }
741 // Otherwise rebuild from the revision pinned in ACTIVE.
742 svcResult, err := s.redeployPinned(ctx, app, opID, name, eff, s.journalCandidate(op, index))
743 if err != nil {
744 step.State = domain.AppStepFailed
745 step.Error = err.Error()
746 step.Diagnostics = svcResult.Diagnostics
747 result.Services[name] = svcResult
748 return
749 }
750 step.State = domain.AppStepSucceeded
751 step.After = svcResult.After
752 result.Services[name] = svcResult
753}
754
755// redeployPinned rebuilds one service from the revision pinned in ACTIVE and
756// publishes it. It is the shared recovery path for a recorded container that
757// no longer exists: boot/start recovery, and a restart of a missing
758// generation. The returned result is always populated, so a caller can
759// journal the attempt even when err is non-nil.
760func (s *Service) redeployPinned(ctx context.Context, app, opID, name string, eff domain.AppEffectiveService, journal candidateJournal) (ServiceResult, error) {
761 rev, err := s.deps.State.LoadRevision(ctx, app, eff.EffectiveRevision)
762 if err != nil {
763 return failedServiceResult(eff.Container, err), err
764 }
765 runtimeImage, err := s.preflightImage(ctx, eff.Spec.Image, eff.Digest)
766 if err != nil {
767 return failedServiceResult(eff.Container, err), err
768 }
769 if _, err := s.resolveServiceBinds(app, eff.Spec); err != nil {
770 return failedServiceResult(eff.Container, err), err
771 }
772 // Revoked device grants fail before any runtime mutation.
773 if _, err := s.resolveServiceDevices(app, eff.Spec); err != nil {
774 return failedServiceResult(eff.Container, err), err
775 }
776 if _, err := s.resolveServiceCapabilities(app, eff.Spec); err != nil {
777 return failedServiceResult(eff.Container, err), err
778 }
779 pinned := pinnedService{
780 name: name, spec: eff.Spec, digest: eff.Digest, runtimeImage: runtimeImage, appEnv: maps.Clone(rev.Spec.Env),
781 appNetworks: append([]domain.AppSharedNetwork(nil), rev.Spec.Networks...),
782 sharedNetworks: domain.AppServiceSharedNetworks(rev.Spec, name),
783 }
784 svcResult := s.deployService(ctx, app, rev.Revision, pinned, opID, eff.Container, serviceStopGrace(eff), journal)
785 if svcResult.Result == "failed" {
786 return svcResult, errors.New(svcResult.Error)
787 }
788 // The step may only be recorded as succeeded once the effective state is
789 // published: a journaled success is never ahead of what the proxy reaches.
790 if err := s.publishService(ctx, app, rev.Revision, pinned, svcResult, opID); err != nil {
791 svcResult.Result = "failed"
792 svcResult.Error = err.Error()
793 return svcResult, err
794 }
795 // The execution is deliberately not recorded here: the graph is not
796 // applied yet. Recording it before routing was accepted would let the next
797 // periodic pass see a verified, unrouted generation and skip the
798 // publication. Each caller records after its own refreshTraffic succeeded.
799 return svcResult, nil
800}
801
802// failedServiceResult builds the terminal result of a service step that
803// failed before any container existed.
804func failedServiceResult(before string, err error) ServiceResult {
805 return ServiceResult{Result: "failed", Before: before, Error: err.Error()}
806}
807
808// verifyRunningService withdraws a running or freshly started generation,
809// re-inspects its loopback binds, and verifies readiness before the service
810// may be published again. Any failure leaves the service withdrawn with its
811// recorded binds cleared, so a not-ready backend is never projected.
812func (s *Service) verifyRunningService(ctx context.Context, app, name string, eff domain.AppEffectiveService, step *domain.AppOperationStep, result *LifecycleResult) {
813 if err := s.withdrawForRecovery(ctx, app, name); err != nil {
814 s.failServiceStep(ctx, app, name, eff, step, result, err.Error())
815 return
816 }
817 startedAt := s.executionStart(ctx, eff.Container)
818 binds, udpBinds, err := s.refreshBackendBinds(ctx, app, name, eff, s.deps.Traffic != nil)
819 if err != nil {
820 s.failServiceStep(ctx, app, name, eff, step, result, err.Error())
821 return
822 }
823 if err := s.waitServiceReady(ctx, app, eff.Container, eff.Spec, binds); err != nil {
824 s.failServiceStep(ctx, app, name, eff, step, result, err.Error())
825 return
826 }
827 // The service is running and its re-inspected loopback binds are
828 // persisted, so the step may succeed: the graph apply for this app
829 // follows once, after the loop.
830 step.State = domain.AppStepSucceeded
831 step.After = eff.Container
832 result.Services[name] = ServiceResult{Result: "deployed", Before: eff.Container, After: eff.Container, BackendBinds: binds, UDPBackendBinds: udpBinds, VerifiedStartedAt: startedAt}
833}
834
835// refreshBackendBinds re-inspects the loopback publishes of a restarted
836// container and persists them to the current active record when needed.
837// Ephemeral binds may shift across runtime restarts; the proxy must
838// never dial the stale recorded bind. Uses the non-destructive
839// inspectBackendBinds: an existing ACTIVE container is never stopped
840// or removed by re-inspection.
841func (s *Service) refreshBackendBinds(ctx context.Context, app, name string, eff domain.AppEffectiveService, forcePersist bool) (map[int]int, map[int]int, error) {
842 ports := backendPorts(eff.Spec)
843 if len(ports) == 0 {
844 return nil, nil, nil
845 }
846 binds, udpBinds, err := s.inspectBackendBinds(ctx, app, name, eff.Container, ports)
847 if err != nil {
848 return nil, nil, err
849 }
850 if !forcePersist && equalBinds(binds, eff.BackendBinds) && equalBinds(udpBinds, eff.UDPBackendBinds) {
851 return binds, udpBinds, nil
852 }
853 active, ok, err := s.deps.State.LoadActive(ctx, app)
854 if err != nil {
855 return nil, nil, fmt.Errorf("deployment: reload active for bind refresh: %w", err)
856 }
857 if !ok {
858 return nil, nil, fmt.Errorf("deployment: app %q has no active state: %w", app, domain.ErrAppStateConflict)
859 }
860 if err := s.persistBackendBinds(ctx, active, name, eff, binds, udpBinds, forcePersist); err != nil {
861 return nil, nil, err
862 }
863 return binds, udpBinds, nil
864}
865
866func (s *Service) persistBackendBinds(ctx context.Context, active domain.AppActive, name string, eff domain.AppEffectiveService, binds, udpBinds map[int]int, force bool) error {
867 current, exists := active.Services[name]
868 if exists && current.Container != eff.Container {
869 return fmt.Errorf("deployment: active generation changed during bind refresh: %w", domain.ErrAppStateConflict)
870 }
871 if !force && exists && equalBinds(binds, current.BackendBinds) && equalBinds(udpBinds, current.UDPBackendBinds) {
872 return nil
873 }
874 if !exists {
875 current = eff
876 }
877 current.BackendBinds = binds
878 current.UDPBackendBinds = udpBinds
879 active.Services[name] = current
880 if err := s.deps.State.SaveActive(ctx, active); err != nil {
881 return fmt.Errorf("deployment: persist refreshed binds: %w", err)
882 }
883 return nil
884}
885
886// failServiceStep records a failed service step and withdraws its
887// recorded backend binds from ACTIVE: a bind that failed verification
888// (or could not be refreshed) must never be served again — the
889// projected backend would otherwise point at a dead or recycled
890// loopback port. Invalidation errors are logged, never fatal: the
891// step already failed, and the next rebuild projects fail-closed.
892func (s *Service) failServiceStep(ctx context.Context, app, name string, eff domain.AppEffectiveService, step *domain.AppOperationStep, result *LifecycleResult, errMsg string) {
893 step.State = domain.AppStepFailed
894 step.Error = errMsg
895 result.Services[name] = ServiceResult{Result: "failed", Before: eff.Container, Error: errMsg}
896 s.clearServiceBinds(ctx, app, name)
897}
898
899// clearServiceBinds withdraws a service's recorded backend binds from
900// ACTIVE through the canonical traffic boundary: a bind that failed
901// verification, refresh, or readiness must never be served, because the
902// projected backend would point at a dead, unready, or recycled loopback
903// port. No graph is applied here: the caller already owns publication
904// (or none is due). Errors are logged, never fatal: the step already
905// failed and the next rebuild projects fail-closed.
906func (s *Service) clearServiceBinds(ctx context.Context, app, name string) {
907 if s.deps.Traffic == nil {
908 return
909 }
910 if err := s.deps.Traffic.WithdrawServiceState(ctx, app, name); err != nil {
911 log := zerowrap.FromCtx(ctx)
912 log.Warn().Err(err).Str("app", app).Str("service", name).Msg("deployment: failed to withdraw unverified binds")
913 }
914}
915
916func equalBinds(a, b map[int]int) bool {
917 if len(a) != len(b) {
918 return false
919 }
920 for port, host := range a {
921 if b[port] != host {
922 return false
923 }
924 }
925 return true
926}
927func sortedServiceNames(active domain.AppActive) []string {
928 names := make([]string, 0, len(active.Services))
929 for name := range active.Services {
930 names = append(names, name)
931 }
932 sort.Strings(names)
933 return names
934}
935
936// lifecycleCtx tags lifecycle use-case logs.
937func lifecycleCtx(ctx context.Context, useCase, app string) context.Context {
938 return zerowrap.CtxWithFields(ctx, map[string]any{
939 zerowrap.FieldLayer: "usecase",
940 zerowrap.FieldUseCase: useCase,
941 "app": app,
942 })
943}