1package deployment
2
3import (
4 "context"
5 "fmt"
6 "maps"
7 "reflect"
8 "sort"
9 "time"
10
11 "github.com/bnema/zerowrap"
12
13 "github.com/bnema/gordon/internal/domain"
14)
15
16// failLogTailLines bounds the log tail attached to service failures.
17const failLogTailLines = 50
18
19// failLogTailTimeout bounds the diagnostics read of a failing container.
20const failLogTailTimeout = 10 * time.Second
21
22// Deploy executes a preflighted revision: fail-fast across services in
23// sorted name order. A service without volumes, binds, devices, or UDP interfaces is
24// replaced with a short overlap: its replacement starts and becomes ready
25// while the superseded container keeps serving, traffic switches to it, and
26// only then is the superseded container stopped, so the service is not
27// interrupted. Every other service is replaced sequentially — withdraw
28// traffic, retire the superseded container, start the replacement and wait
29// for its readiness probe — so a deployment may briefly interrupt it and two
30// Gordon-managed generations of a single-writer service never run at the same
31// time. Volumes are never deleted: no RemoveVolume call, no volume-deletion
32// flags on container removal.
33//
34// It is the synchronous composition of the two engine phases: StartDeploy
35// claims the key and journals the operation, then ExecuteDeploy runs the
36// effects for the owner only.
37func (s *Service) Deploy(ctx context.Context, input DeployInput) (*DeployResult, error) {
38 started, err := s.StartDeploy(ctx, input)
39 if err != nil {
40 return nil, err
41 }
42 if !started.Owned {
43 // The key already answered this request: its stored journal is the
44 // result and no workload is touched again.
45 return journaledOrNil(input, &started.Claim.Journal, started.ReplayError())
46 }
47 return s.ExecuteDeploy(ctx, started.Claim)
48}
49
50// DeployClaim is the durable identity of one started deploy operation. It is
51// the hand-off between the two phases: StartDeploy returns it and
52// ExecuteDeploy consumes it. App and Op are the durable identity, so a
53// caller that reconstructs the claim from them alone (for example after a
54// restart, with no in-process state) leaves Journal and Resolved empty and
55// ExecuteDeploy loads and resolves both from the store.
56type DeployClaim struct {
57 App string
58 Op string
59 Service string
60 Revision string
61 // Resolved is the revision StartDeploy resolved for this claim. Empty
62 // when the claim was reconstructed from identity alone.
63 Resolved domain.AppDesiredRevision
64 // Journal is the claimed operation held in process. Its Op is empty when
65 // the claim was reconstructed from identity alone.
66 Journal domain.AppOperation
67}
68
69// StartDeployResult reports whether this caller owns the claimed operation.
70// Owned is true only for the single caller that claimed Op. Every idempotent
71// replay (the same key already answered) reports Owned false, and the stored
72// journal in Claim.Journal is the result. Only an owner may pass the claim to
73// ExecuteDeploy.
74type StartDeployResult struct {
75 Claim DeployClaim
76 Owned bool
77}
78
79// ReplayError is the explicit disposition of a claim this caller does not
80// own: nil for a terminal success, and a conflict for an in-flight,
81// interrupted, or failed operation, so a replay is never mistaken for a
82// second execution. It is nil for an owned operation.
83func (r StartDeployResult) ReplayError() error {
84 if r.Owned {
85 return nil
86 }
87 return replayError(r.Claim.Journal)
88}
89
90// StartDeploy performs the first phase of a deploy. Under the app lock it
91// converges any interrupted predecessor, resolves the request to a revision
92// and runs the targeted-deploy convergence checks, atomically claims the
93// request key, and persists the non-terminal journal. It performs no image
94// pull, pinning, or workload mutation, so a crash right after it leaves a
95// durable claim that reconciliation can converge.
96//
97// The returned claim distinguishes the newly owned operation (Owned true)
98// from an idempotent replay (Owned false): the same key is never handed to
99// two owners, and a replay never executes effects again.
100func (s *Service) StartDeploy(ctx context.Context, input DeployInput) (*StartDeployResult, error) {
101 // Planning and claiming acquire no runtime resource, so StartDeploy takes
102 // only the per-app coordinator. ExecuteDeploy takes the GC shared lease
103 // across resource acquisition and publication, where it is required.
104 release, err := s.coord.acquire(ctx, input.App)
105 if err != nil {
106 return nil, err
107 }
108 defer release()
109 return s.startDeployLocked(ctx, input)
110}
111
112func (s *Service) startDeployLocked(ctx context.Context, input DeployInput) (*StartDeployResult, error) {
113 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
114 zerowrap.FieldLayer: "usecase",
115 zerowrap.FieldUseCase: "StartDeploy",
116 "app": input.App,
117 })
118 log := zerowrap.FromCtx(ctx)
119
120 if err := s.deps.State.Recover(ctx); err != nil {
121 return nil, fmt.Errorf("deployment: recover before deploy: %w", err)
122 }
123 // A mutation of this app never starts on top of an interrupted one: its
124 // never-published candidates must be gone before this mutation creates a
125 // generation of the same service. ACTIVE is materialized first, so the
126 // published generation cannot be mistaken for a leftover.
127 if err := s.reconcileInterruptedDeploy(ctx, input.App); err != nil {
128 return nil, err
129 }
130 rev, op, owned, err := s.claimDeploymentLocked(ctx, input)
131 if err != nil {
132 return nil, err
133 }
134 if owned {
135 // The claimed operation is now live in this process. It is
136 // registered before the caller releases the coordinator, so a
137 // foreground mutation that takes the lock in the gap before
138 // ExecuteDeploy cannot reconcile the claim away.
139 s.markOperationLive(input.App, op.Op)
140 } else {
141 log.Info().Str("op", op.Op).Msg("deployment: replayed an already claimed operation")
142 }
143 return &StartDeployResult{
144 Claim: DeployClaim{
145 App: input.App, Op: op.Op, Service: input.Service, Revision: input.Revision,
146 Resolved: rev, Journal: op,
147 },
148 Owned: owned,
149 }, nil
150}
151
152// ExecuteDeploy performs the second phase of a deploy: it reacquires the app
153// lock, loads and verifies the claimed journal by app/op identity, runs the
154// image pull/pin and the per-service replacement, and records the terminal
155// outcome. A claim that is already terminal, or that does not match its
156// identity, is never executed: it replays its stored outcome instead. An
157// operation that stops mid-execution (shutdown or cancellation) leaves its
158// non-terminal journal behind for reconciliation, so it can never run twice.
159//
160// The operation stays live until this returns: the store record, not the
161// in-process claim, is what executes, and the live marker is cleared only
162// after the outcome was recorded or a recoverable non-terminal claim was
163// left behind, while the app lock is still held.
164func (s *Service) ExecuteDeploy(ctx context.Context, claim DeployClaim) (*DeployResult, error) {
165 release, err := s.acquireAppContext(ctx, claim.App)
166 if err != nil {
167 // The claim is left as persisted: recoverable, but no longer owned
168 // here, so a later reconciliation can converge it.
169 s.clearOperationLive(claim.App, claim.Op)
170 return nil, err
171 }
172 defer release()
173 // Registered after release, so it runs first: the marker is dropped
174 // before the app lock is released, and never while a live goroutine may
175 // still persist state.
176 defer s.clearOperationLive(claim.App, claim.Op)
177 return s.executeLocked(ctx, claim)
178}
179
180// AbandonDeploy settles a claim this process owns but will never execute, such
181// as when daemon shutdown races the hand-off between StartDeploy and
182// ExecuteDeploy. It drops the in-process live marker and converges the still
183// non-terminal journal as an interrupted operation, so no durable claim stays
184// marked live forever and a later reconciliation reads a terminal outcome
185// instead of a permanently in-flight key. It takes the app lock but performs
186// no workload effect.
187func (s *Service) AbandonDeploy(ctx context.Context, claim DeployClaim) error {
188 if claim.App == "" || claim.Op == "" {
189 return fmt.Errorf("deployment: abandon requires an app and operation identity: %w", domain.ErrAppStateConflict)
190 }
191 // The marker is dropped before any fallible step: even if convergence
192 // fails, the claim is no longer owned here, so a later reconciliation
193 // (foreground or boot) can converge it instead of skipping it as live.
194 s.clearOperationLive(claim.App, claim.Op)
195 release, err := s.coord.acquire(ctx, claim.App)
196 if err != nil {
197 return err
198 }
199 defer release()
200 return s.reconcileInterruptedDeploy(ctx, claim.App)
201}
202
203// executeLocked runs the claimed, non-terminal operation. The caller holds
204// the app lock. Every failure that happens before a terminal outcome is
205// recorded (load active, revision, service step, traffic) leaves the journal
206// with its step state, never a silent success.
207func (s *Service) executeLocked(ctx context.Context, claim DeployClaim) (*DeployResult, error) {
208 input := DeployInput{App: claim.App, Revision: claim.Revision, Service: claim.Service}
209 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
210 zerowrap.FieldLayer: "usecase",
211 zerowrap.FieldUseCase: "ExecuteDeploy",
212 "app": claim.App,
213 "op": claim.Op,
214 })
215 log := zerowrap.FromCtx(ctx)
216
217 op, err := s.loadClaim(ctx, claim)
218 if err != nil {
219 return nil, err
220 }
221 if op.Terminal() {
222 // The claim was finalized between the phases: never execute a settled
223 // key, replay its stored outcome instead.
224 return journaledDeployResult(input, &op), replayError(op)
225 }
226 // The claimed revision is authoritative for the whole execution: it was
227 // resolved when the key was claimed and is what the journal records, so a
228 // desired-state change between the two phases must never redirect this
229 // operation to a different revision than the one it pinned.
230 rev, err := s.claimRevision(ctx, claim, op)
231 if err != nil {
232 s.failOperation(ctx, &op, err, "error")
233 return journaledDeployResult(input, &op), err
234 }
235 pinned, err := s.pinPreflightLocked(ctx, input.App, input.Service, rev, &op)
236 if err != nil {
237 return journaledDeployResult(input, &op), err
238 }
239 sort.Slice(pinned, func(i, j int) bool { return pinned[i].name < pinned[j].name })
240
241 active, _, err := s.deps.State.LoadActive(ctx, input.App)
242 if err != nil {
243 loadErr := fmt.Errorf("deployment: load active: %w", err)
244 s.failOperation(ctx, &op, loadErr, "error")
245 return journaledDeployResult(input, &op), loadErr
246 }
247
248 result := &DeployResult{
249 Op: op.Op,
250 App: input.App,
251 Revision: rev.Revision,
252 Services: map[string]ServiceResult{},
253 }
254 // A targeted deploy is intentionally a partial plan: services omitted
255 // from pinned remain active. Full deploys reconcile actual removals.
256 if input.Service == "" {
257 if err := s.reconcileRemovalsForDeploy(ctx, input.App, active, pinned, &op, result, log); err != nil {
258 return result, err
259 }
260 }
261 for i, p := range pinned {
262 if err := s.runServiceStep(ctx, input.App, rev.Revision, i, p, &op, active, result, log); err != nil {
263 return result, err
264 }
265 }
266 op.Outcome = ComputeOutcome(result.Services)
267 op.Warnings = journalWarnings(collectCleanupWarnings(result.Services))
268 result.CleanupWarnings = collectCleanupWarnings(result.Services)
269 if saveErr := s.deps.State.SaveOperation(ctx, op); saveErr != nil {
270 log.Warn().Err(saveErr).Msg("deployment: failed to record deploy outcome")
271 }
272 return result, nil
273}
274
275// loadClaim verifies the claimed journal by app/op identity. The store is
276// authoritative: the operation is always reloaded by app/op ID, so a stale
277// in-process journal can never be executed (for example when another actor
278// finalized the key between the claim and execution). A journal that does
279// not match the identity, or is not a deploy, is refused.
280func (s *Service) loadClaim(ctx context.Context, claim DeployClaim) (domain.AppOperation, error) {
281 if claim.App == "" || claim.Op == "" {
282 return domain.AppOperation{}, fmt.Errorf("deployment: execute requires an app and operation identity: %w", domain.ErrAppStateConflict)
283 }
284 op, err := s.deps.State.LoadOperation(ctx, claim.App, claim.Op)
285 if err != nil {
286 return domain.AppOperation{}, fmt.Errorf("deployment: load claimed operation %q: %w", claim.Op, err)
287 }
288 if op.App != claim.App || op.Op != claim.Op {
289 return domain.AppOperation{}, fmt.Errorf("deployment: claim does not identify app %q operation %q: %w", claim.App, claim.Op, domain.ErrAppStateConflict)
290 }
291 if op.Kind != "deploy" {
292 return domain.AppOperation{}, fmt.Errorf("deployment: operation %q is %q, not a deploy: %w", claim.Op, op.Kind, domain.ErrAppStateConflict)
293 }
294 return op, nil
295}
296
297// claimRevision returns the revision StartDeploy captured. A claim
298// reconstructed from identity alone resolves the revision its journal
299// recorded, so a resumed execution pins the revision it was started for.
300func (s *Service) claimRevision(ctx context.Context, claim DeployClaim, op domain.AppOperation) (domain.AppDesiredRevision, error) {
301 if claim.Resolved.Revision != "" {
302 return claim.Resolved, nil
303 }
304 return s.resolveRevision(ctx, DeployInput{App: claim.App, Revision: op.InputRevision, Service: claim.Service})
305}
306
307// runServiceStep executes one pinned service in the journal: replace it,
308// publish its ACTIVE entry, then publish traffic for the new container. The
309// step is recorded as succeeded only after the runtime effect, the ACTIVE
310// publication, and the traffic apply all succeeded, so a journaled success
311// is never ahead of what is routed.
312//
313// An overlap-eligible service (see overlapEligible) is replaced without
314// withdrawing the superseded container: it keeps serving and is retired only
315// after the new generation is published and routed. Before the switch the
316// superseded container gets a durable retirement-pending marker carrying its
317// stop grace. That marker, not the operation journal, owns the cleanup: it
318// survives newer operations, and the periodic reconciliation retires the
319// container once its successor is verified and routed.
320//
321// If the publication or the traffic apply of an overlap step fails, the
322// superseded container is deliberately left running: the proxy may still
323// route to it, and ACTIVE may already name the new container, so stopping it
324// could create an outage the failed deploy cannot repair. The step is
325// journaled as failed with the new container as After and a cleanup warning
326// names the superseded one. If ACTIVE names the new container, the periodic
327// reconciliation verifies and republishes it and then retires the superseded
328// one through its marker; if ACTIVE never changed, it still names the
329// superseded container, which keeps serving, and the marker is dropped.
330func (s *Service) runServiceStep(
331 ctx context.Context,
332 app, revision string,
333 index int,
334 p pinnedService,
335 op *domain.AppOperation,
336 active domain.AppActive,
337 result *DeployResult,
338 log zerowrap.Logger,
339) error {
340 before := ""
341 if eff, ok := active.Services[p.name]; ok {
342 before = eff.Container
343 }
344 stepID := "service." + p.name + ".replace"
345 // The step is journaled before the replacement runs and carries the
346 // candidate's container ID as soon as it exists, so an interrupted deploy
347 // leaves a trace boot recovery can clean up instead of an untracked
348 // generation.
349 step := &op.Steps[index+1]
350 *step = domain.AppOperationStep{
351 ID: stepID, Service: p.name,
352 Digest: p.digest, Image: p.runtimeImage,
353 Before: before, State: domain.AppStepPending,
354 }
355 var svcResult ServiceResult
356 overlap := false
357 // The grace of the generation being superseded, read before anything
358 // publishes a record of its replacement.
359 beforeGrace := activeStopGrace(active, p.name)
360 if current, ok := s.unchangedService(ctx, app, p, active); ok {
361 // Same image, spec, environment, and secret values already running:
362 // keep the container.
363 svcResult = ServiceResult{
364 Result: ServiceResultUnchanged, EffectiveRevision: revision,
365 Before: before, After: before,
366 BackendBinds: current.BackendBinds, UDPBackendBinds: current.UDPBackendBinds,
367 RestartUnsafe: singleWriterRequired(p.spec),
368 }
369 step.Detail = domain.AppServiceUnchanged + ": already running " + p.spec.Image + " (" + p.digest + ")"
370 } else if before != "" && canOverlap(p.spec, active.Services[p.name].Spec) {
371 overlap = true
372 svcResult = s.deployOverlap(ctx, app, revision, p, op.Op, before, beforeGrace, s.journalCandidate(op, index+1))
373 } else {
374 svcResult = s.deployService(ctx, app, revision, p, op.Op, before, beforeGrace, s.journalCandidate(op, index+1))
375 }
376 result.Services[p.name] = svcResult
377 // A recorded candidate is authoritative while the step is unresolved: a
378 // replacement that failed after creating its container must stay traceable
379 // for reconciliation.
380 if svcResult.After != "" {
381 step.After = svcResult.After
382 }
383 fail := func(err error) error {
384 step.State = domain.AppStepFailed
385 step.Error = err.Error()
386 step.Diagnostics = svcResult.Diagnostics
387 op.Warnings = journalWarnings(collectCleanupWarnings(result.Services))
388 op.Outcome = ComputeOutcome(result.Services)
389 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
390 log.Warn().Err(saveErr).Msg("deployment: failed to record service failure")
391 }
392 return err
393 }
394 if svcResult.Result == "failed" {
395 // Fail fast: later services stay unchanged.
396 return fail(fmt.Errorf("deployment: service %q failed at %s: %s: %w",
397 p.name, stepID, svcResult.Error, domain.ErrAppStateConflict))
398 }
399 svcResult, err := s.publishAndRoute(ctx, app, revision, p, svcResult, op.Op)
400 if err != nil {
401 if overlap {
402 svcResult.CleanupWarnings = append(svcResult.CleanupWarnings, CleanupWarning{
403 Service: p.name, Leftover: before,
404 Detail: "superseded container kept running: the new generation was not confirmed published and routed",
405 })
406 }
407 result.Services[p.name] = svcResult
408 return fail(err)
409 }
410 // The generation is verified and routed: the periodic pass must not
411 // mistake its first observation for an unseen execution and withdraw it.
412 s.recordVerifiedExecution(app, p.name, svcResult)
413 if overlap {
414 svcResult = s.retireSuperseded(ctx, app, p.name, before, beforeGrace, svcResult)
415 }
416 result.Services[p.name] = svcResult
417 step.State = domain.AppStepSucceeded
418 step.Error = ""
419 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
420 log.Warn().Err(saveErr).Msg("deployment: failed to checkpoint service progress")
421 }
422 return nil
423}
424
425// publishAndRoute publishes the per-service effective state, so the new
426// binds become the service's recorded backend, then rebuilds the traffic
427// graph so the proxy is repointed at them. ACTIVE already names the new
428// container when the graph is applied, but the operation is a failure until
429// routing accepted it, so a rejected graph is returned as a failed result and
430// never journaled as success.
431func (s *Service) publishAndRoute(ctx context.Context, app, revision string, p pinnedService, svcResult ServiceResult, opID string) (ServiceResult, error) {
432 if err := s.publishService(ctx, app, revision, p, svcResult, opID); err != nil {
433 svcResult.Result = "failed"
434 svcResult.Error = err.Error()
435 return svcResult, err
436 }
437 if err := s.refreshTraffic(ctx, app); err != nil {
438 svcResult.Result = "failed"
439 svcResult.Error = err.Error()
440 return svcResult, err
441 }
442 return svcResult, nil
443}
444
445// retireSuperseded stops the container an overlap replacement superseded,
446// once the new generation is published and routed. It uses the stop grace of
447// the generation being stopped: the runtime sends SIGTERM, then kills after
448// the grace. Backend claims are released and the retirement-pending marker is
449// cleared by retireContainer once the container is confirmed gone. A
450// retirement that fails is not a deploy failure — the new generation already
451// serves — but a cleanup warning on the service; the marker stays, so the
452// periodic reconciliation retries it.
453func (s *Service) retireSuperseded(ctx context.Context, app, service, before string, grace time.Duration, result ServiceResult) ServiceResult {
454 retired := s.retireContainer(ctx, app, retireOptions{Service: service, Grace: grace, ClearInhibition: true, RetryOnLeak: true}, before)
455 result.CleanupWarnings = append(result.CleanupWarnings, retired.Warnings...)
456 if !retired.Gone && len(retired.Warnings) == 0 {
457 result.CleanupWarnings = append(result.CleanupWarnings, CleanupWarning{
458 Service: service, Leftover: before, Detail: "superseded container not confirmed gone",
459 })
460 }
461 return result
462}
463
464// unchangedService reports whether the service already runs exactly what p
465// would create, in a healthy published state: the same digest, service spec,
466// app environment, and shared networks, in a running container that is not
467// withdrawn or recovery-inhibited and whose recorded binds cover every
468// backend port, and whose environment already holds the current secret
469// values. Services with host binds or devices are never skipped: their
470// resolved sources depend on host policy that ACTIVE does not record.
471func (s *Service) unchangedService(ctx context.Context, app string, p pinnedService, active domain.AppActive) (domain.AppEffectiveService, bool) {
472 eff, ok := active.Services[p.name]
473 if !ok || eff.Container == "" || eff.Digest == "" || eff.Digest != p.digest {
474 return eff, false
475 }
476 reason := s.unchangedRefusal(ctx, app, p, eff)
477 if reason != "" {
478 log := zerowrap.FromCtx(ctx)
479 log.Debug().Str("app", app).Str("service", p.name).Str("reason", reason).
480 Msg("deployment: same digest, replacing service")
481 return eff, false
482 }
483 return eff, true
484}
485
486// unchangedRefusal returns why a service running the same digest must still
487// be replaced, or "" when it can be kept.
488func (s *Service) unchangedRefusal(ctx context.Context, app string, p pinnedService, eff domain.AppEffectiveService) string {
489 switch {
490 case len(p.spec.Binds) > 0 || len(p.spec.Devices) > 0:
491 return "host binds or devices"
492 case len(p.spec.Capabilities) > 0:
493 return "administrative capabilities"
494 case !domain.SameAppServiceRuntime(p.spec, eff.Spec):
495 // Backup declarations never reach the container, so a backup-only
496 // change keeps it; publishService still records the new spec in ACTIVE.
497 return "spec changed"
498 case !bindsCover(eff, backendPorts(p.spec)):
499 return "backend binds missing"
500 case s.publication.inhibited(app, p.name):
501 return "withdrawal pending"
502 }
503 previous, err := s.deps.State.LoadRevision(ctx, app, eff.EffectiveRevision)
504 if err != nil {
505 return "load effective revision: " + err.Error()
506 }
507 if !maps.Equal(previous.Spec.Env, p.appEnv) {
508 return "app env changed"
509 }
510 if !reflect.DeepEqual(domain.AppServiceSharedNetworks(previous.Spec, p.name), p.sharedNetworks) {
511 return "shared networks changed"
512 }
513 if inhibited, err := s.recoveryInhibited(ctx, app, p.name, eff.Container); err != nil || inhibited {
514 return "recovery inhibited"
515 }
516 container, err := s.deps.Runtime.InspectContainer(ctx, eff.Container)
517 if err != nil || container.Status != string(domain.ContainerStatusRunning) {
518 return "container not running"
519 }
520 if !s.envCurrent(ctx, app, p, container) {
521 return "environment or secrets changed"
522 }
523 return ""
524}
525
526// envCurrent reports whether the running container was created with the
527// environment p would get now, including current secret values. The
528// runtime env also holds image-declared keys, so every desired entry must
529// be present; removed keys are caught by the spec and app env comparisons.
530// File-delivered secrets are compared against the container's host files.
531// Values are compared in memory only and never logged.
532func (s *Service) envCurrent(ctx context.Context, app string, p pinnedService, container *domain.Container) bool {
533 desired, files, err := s.serviceEnv(ctx, app, p)
534 if err != nil {
535 return false
536 }
537 have := make(map[string]struct{}, len(container.Env))
538 for _, entry := range container.Env {
539 have[entry] = struct{}{}
540 }
541 for _, entry := range desired {
542 if _, ok := have[entry]; !ok {
543 return false
544 }
545 }
546 return s.secretFilesCurrent(ctx, app, container.Name, files)
547}
548
549// secretFilesCurrent reports whether every host secret file of the
550// container holds the current value. Values are compared in memory only.
551func (s *Service) secretFilesCurrent(ctx context.Context, app, container string, files map[string]string) bool {
552 if len(files) == 0 {
553 return true
554 }
555 if s.deps.SecretFiles == nil {
556 return false
557 }
558 for key, want := range files {
559 got, err := s.deps.SecretFiles.Read(ctx, app, container, key)
560 if err != nil || got != want {
561 return false
562 }
563 }
564 return true
565}
566
567// bindsCover reports whether the recorded loopback binds publish every
568// backend port the spec requires, so a kept container stays routable.
569func bindsCover(eff domain.AppEffectiveService, ports []domain.ContainerBackendPort) bool {
570 for _, port := range ports {
571 binds := eff.BackendBinds
572 if port.Protocol == domain.NetworkProtocolUDP {
573 binds = eff.UDPBackendBinds
574 }
575 if binds[port.ContainerPort] == 0 {
576 return false
577 }
578 }
579 return true
580}
581
582// failOperation records a terminal failure on an already claimed
583// journal, so a keyed repeat replays a terminal outcome instead of
584// reading a permanently in-flight claim. stepID names the failing phase
585// (a service step, a traffic publication, or a bare error). Journal-write
586// failures are logged, never returned: the caller already holds a more
587// specific error, and the next mutation's recovery pass converges state.
588func (s *Service) failOperation(ctx context.Context, op *domain.AppOperation, err error, stepID string) {
589 op.Outcome = domain.AppOutcomeFailed
590 op.Steps = append(op.Steps, domain.AppOperationStep{
591 ID: stepID, State: domain.AppStepFailed, Error: err.Error(),
592 })
593 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
594 log := zerowrap.FromCtx(ctx)
595 log.Warn().Err(saveErr).Str("op", op.Op).Msg("deployment: failed to record operation failure")
596 }
597}
598
599// failTrafficPublication records that the final graph apply was rejected.
600// The workload steps already ran and stay accurate, but the operation is a
601// failure: routing did not accept the published state, so a later replay
602// of the same key must never read success.
603func (s *Service) failTrafficPublication(ctx context.Context, op *domain.AppOperation, err error) {
604 s.failOperation(ctx, op, err, "traffic.publish")
605}
606
607// journaledOrNil reports a preflight that already answered the request
608// key: the stored journal when there is one, the bare error otherwise.
609func journaledOrNil(input DeployInput, op *domain.AppOperation, err error) (*DeployResult, error) {
610 if op == nil {
611 return nil, err
612 }
613 return journaledDeployResult(input, op), err
614}
615
616func journaledDeployResult(input DeployInput, op *domain.AppOperation) *DeployResult {
617 return &DeployResult{
618 Op: op.Op,
619 App: input.App,
620 Revision: op.InputRevision,
621 Outcome: op.Outcome,
622 }
623}
624
625// candidateJournal durably records a freshly created replacement container
626// before it can be started or join a network. An interrupted deployment then
627// leaves a trace boot recovery removes, instead of an untracked running
628// generation that could overlap the one recovery rebuilds.
629//
630// The window between the runtime creating a container and this write is the
631// irreducible one: a container ID cannot be recorded before it exists.
632// sync/atomic is not needed here: the journal write is synchronous.
633type candidateJournal func(ctx context.Context, containerID string) error
634
635// journalCandidate records one created candidate in the operation journal.
636// The step is addressed by index and read at write time, so a later append to
637// the operation's steps can never leave the recorder writing to a stale
638// element. A nil operation (an internal path with no journal) records nothing.
639func (s *Service) journalCandidate(op *domain.AppOperation, index int) candidateJournal {
640 if op == nil || index < 0 || index >= len(op.Steps) {
641 return nil
642 }
643 return func(ctx context.Context, containerID string) error {
644 op.Steps[index].After = containerID
645 if err := s.deps.State.SaveOperation(ctx, *op); err != nil {
646 return fmt.Errorf("deployment: record candidate %s: %w", containerID, err)
647 }
648 return nil
649 }
650}
651
652// deployService replaces one service sequentially and returns its terminal
653// result. before is the ACTIVE container being superseded (empty on a first
654// deployment); beforeGrace is its effective stop grace. It serves every
655// service that is not overlap-eligible, a first deployment, and the
656// rebuild of a missing generation by start and restart; deployOverlap is the
657// overlapping alternative used by a deploy.
658//
659// The order is fixed: withdraw from traffic and confirm it, durably inhibit
660// a single-writer generation, retire the superseded container and confirm
661// it is gone, then create and start the replacement and wait for its
662// readiness probe. Nothing is created before the previous generation is
663// confirmed gone, so two generations never overlap; a withdrawal or
664// retirement that cannot be confirmed aborts without creating a candidate.
665func (s *Service) deployService(ctx context.Context, app, revision string, p pinnedService, opID, before string, beforeGrace time.Duration, journal candidateJournal) ServiceResult {
666 // Revalidate the current authorization and source before withdrawing or
667 // retiring the serving generation. A reload between preflight and this
668 // service step must fail without mutating the existing workload.
669 if failed, ok := s.revalidateService(app, revision, p, before); !ok {
670 return failed
671 }
672 // Withdraw the service from traffic first and confirm it: a withdrawal
673 // that cannot be applied must block every container mutation, so no
674 // unverified generation stays routable. A first deployment has no
675 // published generation to withdraw.
676 if before != "" {
677 if err := s.withdrawForRecovery(ctx, app, p.name); err != nil {
678 return s.failResult(revision, before, "", err)
679 }
680 }
681 // A volume- or bind-owning replacement may write while the old
682 // generation still exists and could be revived by native restart policy.
683 // Inhibit that generation durably BEFORE the write can happen, so boot or
684 // periodic recovery can never restart the old writer on top of the new
685 // one. Cleared once the safe generation is published (publishService) or
686 // the operator removes the app.
687 if before != "" && singleWriterRequired(p.spec) {
688 if err := s.inhibitRecovery(ctx, app, p.name, before, domain.AppInhibitReplacementPending, opID); err != nil {
689 return s.failResult(revision, before, "", err)
690 }
691 }
692 if before != "" {
693 // The superseded generation must be confirmed gone before a new
694 // generation starts: a failed retirement aborts the replacement
695 // instead of allowing overlapping generations, and the generation
696 // keeps its inhibition and its claims.
697 retired := s.retireContainer(ctx, app, retireOptions{
698 Service: p.name, Grace: beforeGrace,
699 }, before)
700 if !retired.Gone {
701 return s.failResult(revision, before, "", fmt.Errorf("deployment: retire superseded container %s: %s", before, cleanupDetail(retired)))
702 }
703 }
704 return s.startCandidate(ctx, app, revision, p, opID, before, journal)
705}
706
707// deployOverlap creates the replacement of an overlap-eligible service while
708// the superseded container keeps running and stays routed: no inhibition and
709// no retirement. Container names carry the operation, so the two generations
710// never collide, and their loopback binds are ephemeral.
711//
712// The only withdrawal it can perform is the retry of one already pending for
713// this service: an earlier withdrawal that could not be proven leaves the
714// service marked, and a marked service would be withdrawn again by the
715// traffic refresh that follows the switch, erasing the binds the switch just
716// published. The retry runs before the candidate exists, so a failure aborts
717// the deploy untouched.
718//
719// It returns the ready candidate with its binds, after durably recording the
720// retirement-pending marker of the superseded container; on any failure the
721// candidate is removed and the superseded container is left exactly as it was.
722// Retiring the superseded container is the caller's job, once the candidate is
723// published and routed (retireSuperseded).
724func (s *Service) deployOverlap(ctx context.Context, app, revision string, p pinnedService, opID, before string, beforeGrace time.Duration, journal candidateJournal) ServiceResult {
725 if failed, ok := s.revalidateService(app, revision, p, before); !ok {
726 return failed
727 }
728 if s.publication.inhibited(app, p.name) {
729 if err := s.withdrawForRecovery(ctx, app, p.name); err != nil {
730 return s.failResult(revision, before, "", err)
731 }
732 }
733 result := s.startCandidate(ctx, app, revision, p, opID, before, journal)
734 if result.Result != "deployed" {
735 return result
736 }
737 // The marker is durable before anything is published: whatever happens
738 // next, the superseded container is retired by the periodic
739 // reconciliation instead of being forgotten, and recovery never revives it.
740 if err := s.inhibitRetirement(ctx, app, p.name, before, opID, beforeGrace); err != nil {
741 result.Result = "failed"
742 result.Error = err.Error()
743 result.CleanupWarnings = append(result.CleanupWarnings, s.retireCandidate(ctx, app, p.name, result.After)...)
744 return result
745 }
746 result.SupersededRunning = true
747 return result
748}
749
750// revalidateService re-resolves the binds, devices, and capabilities of a service against
751// the current policy before any workload mutation. It returns the failed
752// result and false when either was revoked since preflight.
753func (s *Service) revalidateService(app, revision string, p pinnedService, before string) (ServiceResult, bool) {
754 if _, err := s.resolveServiceBinds(app, p.spec); err != nil {
755 return s.failResult(revision, before, "", err), false
756 }
757 if _, err := s.resolveServiceDevices(app, p.spec); err != nil {
758 return s.failResult(revision, before, "", err), false
759 }
760 if _, err := s.resolveServiceCapabilities(app, p.spec); err != nil {
761 return s.failResult(revision, before, "", err), false
762 }
763 return ServiceResult{}, true
764}
765
766// startCandidate creates and starts the replacement generation and waits for
767// its readiness probe. A candidate that fails readiness is removed after its
768// redacted diagnostics were captured; whatever generation it supersedes is
769// not touched here.
770func (s *Service) startCandidate(ctx context.Context, app, revision string, p pinnedService, opID, before string, journal candidateJournal) ServiceResult {
771 candidate, err := s.createAndStart(ctx, app, revision, p, opID, journal)
772 if err != nil {
773 result := s.failResult(revision, before, "", err)
774 if candidate.Container != nil {
775 // The candidate exists: keep its ID in the terminal result so the
776 // journal and the operator can still trace it.
777 result.After = candidate.Container.ID
778 }
779 return result
780 }
781 created := candidate.Container
782 if err := s.waitServiceReady(ctx, app, created.ID, deploymentReadiness(p.spec), candidate.TCPBinds); err != nil {
783 // Capture redacted diagnostics while the failed replacement still
784 // exists, then remove only that candidate. A superseded container
785 // that was already retired is never recreated. Cleanup is best
786 // effort: a canceled context can leave the candidate behind, which is
787 // then reported as a leftover warning instead of a false success.
788 tail := s.redactDiagnostics(ctx, app, p, s.logTail(ctx, created.ID))
789 cleanup := s.retireCandidate(ctx, app, p.name, created.ID)
790 return ServiceResult{
791 Result: "failed",
792 EffectiveRevision: revision,
793 Before: before,
794 After: created.ID,
795 RestartUnsafe: singleWriterRequired(p.spec),
796 Error: err.Error(),
797 Diagnostics: tail,
798 CleanupWarnings: cleanup,
799 }
800 }
801 return ServiceResult{
802 Result: "deployed",
803 EffectiveRevision: revision,
804 Before: before,
805 After: created.ID,
806 RestartUnsafe: singleWriterRequired(p.spec),
807 BackendBinds: candidate.TCPBinds,
808 UDPBackendBinds: candidate.UDPBinds,
809 VerifiedStartedAt: candidate.StartedAt,
810 }
811}
812
813// singleWriterRequired reports services that must never have two generations
814// running concurrently. Persistent volumes and any bind (especially a
815// writable one) may be written by both, so they are treated identically.
816// Devices and UDP interfaces are also excluded from the overlap (see
817// overlapEligible).
818func singleWriterRequired(spec domain.AppService) bool {
819 return len(spec.Volumes) > 0 || len(spec.Binds) > 0
820}
821
822// canOverlap reports whether a replacement may run beside its predecessor: both
823// the new spec and the predecessor's effective spec (as recorded in ACTIVE)
824// must be overlap-eligible. A predecessor that owned a volume, bind, device,
825// or UDP interface is a single writer or has pinned UDP sessions whatever the
826// replacement declares, so removing the resource from the spec does not make
827// the transition safe.
828func canOverlap(next, previous domain.AppService) bool {
829 return overlapEligible(next) && overlapEligible(previous)
830}
831
832// overlapEligible reports whether a service may be replaced with a short
833// overlap of the old and the new generation: it has no volume, bind, device,
834// or UDP interface. Volumes, binds, and devices are stateful or exclusive
835// resources two generations must not share. UDP sessions are pinned by the
836// relay to the backend they started on and outlive the switch for the
837// configured drain timeout, so they would keep reaching a backend that is
838// about to be stopped and whose loopback port may be recycled. TCP is fine:
839// established relays do not redial. Every other service keeps the sequential
840// replacement, so two generations of such a service never run together.
841func overlapEligible(spec domain.AppService) bool {
842 return len(spec.Volumes) == 0 && len(spec.Binds) == 0 && len(spec.Devices) == 0 && len(spec.UDP) == 0
843}
844
845// refreshTraffic rebuilds the proxy host index and applies the full
846// HTTP/L4 graph after activation. A failure is fatal to the caller:
847// ACTIVE is published but the new service is not routable, so the
848// operation must not report success. Any service of this app whose
849// fail-closed withdrawal was never applied is withdrawn again first, so
850// a later publication never runs on top of unproven forwarding.
851func (s *Service) refreshTraffic(ctx context.Context, app string) error {
852 if s.deps.Traffic == nil {
853 return nil
854 }
855 for _, service := range s.publication.pending(app) {
856 if err := s.withdrawForRecovery(ctx, app, service); err != nil {
857 return err
858 }
859 }
860 if err := s.deps.Traffic.RebuildTraffic(ctx); err != nil {
861 return fmt.Errorf("deployment: rebuild traffic for %q: %w", app, err)
862 }
863 return nil
864}
865
866// publishService writes the per-service effective record once the
867// replacement passed readiness.
868func (s *Service) publishService(ctx context.Context, app, revision string, p pinnedService, result ServiceResult, opID string) error {
869 if app == "" {
870 return fmt.Errorf("deployment: publish active: empty app: %w", domain.ErrAppStateConflict)
871 }
872 active, ok, err := s.deps.State.LoadActive(ctx, app)
873 if err != nil {
874 return fmt.Errorf("deployment: load active: %w", err)
875 }
876 if !ok {
877 active = domain.AppActive{App: app, Services: map[string]domain.AppEffectiveService{}}
878 }
879 if active.Services == nil {
880 active.Services = map[string]domain.AppEffectiveService{}
881 }
882 activatedBy, activatedAt := opID, time.Now().UTC()
883 if previous, ok := active.Services[p.name]; ok && result.Result == ServiceResultUnchanged {
884 // A kept container keeps the activation of the operation that created it.
885 activatedBy, activatedAt = previous.ActivatedBy, previous.ActivatedAt
886 }
887 active.Services[p.name] = domain.AppEffectiveService{
888 EffectiveRevision: revision,
889 ActivatedBy: activatedBy,
890 ActivatedAt: activatedAt,
891 Image: p.spec.Image,
892 Digest: p.digest,
893 Container: result.After,
894 Spec: p.spec,
895 BackendBinds: result.BackendBinds,
896 UDPBackendBinds: result.UDPBackendBinds,
897 }
898 active.ConvergedRevision = revision
899 converged := true
900 for _, svc := range active.Services {
901 if svc.EffectiveRevision != revision {
902 converged = false
903 break
904 }
905 }
906 active.Converged = converged
907 if converged {
908 active.Networks = append([]domain.AppSharedNetwork(nil), p.appNetworks...)
909 }
910 if err := s.deps.State.SaveActive(ctx, active); err != nil {
911 return fmt.Errorf("deployment: publish active: %w", err)
912 }
913 if err := s.recordOwnership(ctx, app, p); err != nil {
914 return err
915 }
916 return s.clearSupersededMarker(ctx, app, p.name, result)
917}
918
919// clearSupersededMarker clears the recovery inhibition of the generation a
920// sequential replacement superseded: that clearing is what makes the new
921// generation authoritative, and a failed publication leaves the marker in
922// place. An overlap replacement is the exception: its superseded container
923// still runs, and its retirement-pending marker is what gets it retired.
924func (s *Service) clearSupersededMarker(ctx context.Context, app, service string, result ServiceResult) error {
925 if result.Before == "" || result.Before == result.After || result.SupersededRunning {
926 return nil
927 }
928 return s.clearRecoveryInhibition(ctx, app, service, result.Before)
929}
930
931// recordOwnership stamps volume/secret/network ownership after publication.
932func (s *Service) recordOwnership(ctx context.Context, app string, p pinnedService) error {
933 ownership, err := s.deps.State.LoadOwnership(ctx, app)
934 if err != nil {
935 return err
936 }
937 if ownership.App == "" {
938 ownership.App = app
939 }
940 // ownership.ID arrives from LoadOwnership (store-assigned UUID);
941 // empty only for legacy records predating UUIDs.
942 volumes := append([]domain.AppOwnedVolume(nil), ownership.Volumes...)
943 seen := map[string]int{}
944 for i, vol := range volumes {
945 seen[vol.Name] = i
946 }
947 for _, vol := range p.spec.Volumes {
948 entry := domain.AppOwnedVolume{
949 Name: vol.Name,
950 Service: p.spec.Name,
951 RuntimeName: domain.RuntimeVolumeName(app, p.spec.Name, vol.Name),
952 State: domain.AppResourceAttached,
953 }
954 if idx, ok := seen[vol.Name]; ok {
955 volumes[idx] = entry
956 } else {
957 seen[vol.Name] = len(volumes)
958 volumes = append(volumes, entry)
959 }
960 }
961 ownership.Volumes = volumes
962 secrets := append([]domain.AppOwnedSecret(nil), ownership.Secrets...)
963 secretSeen := map[string]int{}
964 for i, secret := range secrets {
965 secretSeen[secret.Service+"\x00"+secret.Env] = i
966 }
967 for envKey, name := range p.spec.Secrets {
968 entry := domain.AppOwnedSecret{
969 Service: p.spec.Name,
970 Env: envKey,
971 Name: name,
972 Path: domain.AppSecretPathForID(ownership.ID, app, p.spec.Name, name),
973 State: domain.AppResourceAttached,
974 }
975 key := p.spec.Name + "\x00" + envKey
976 if idx, ok := secretSeen[key]; ok {
977 secrets[idx] = entry
978 } else {
979 secretSeen[key] = len(secrets)
980 secrets = append(secrets, entry)
981 }
982 }
983 ownership.Secrets = secrets
984 ownership.Images = recordImageOwnership(ownership.Images, p)
985 recordNetworkOwnership(&ownership, s.deps.Networks.Prefix, p.sharedNetworks)
986 if ownership.Services == nil {
987 ownership.Services = map[string]domain.AppServiceRecovery{}
988 }
989 ownership.Services[p.spec.Name] = domain.AppServiceRecovery{RestartUnsafe: singleWriterRequired(p.spec)}
990 if err := s.deps.State.SaveOwnership(ctx, ownership); err != nil {
991 return fmt.Errorf("deployment: record ownership: %w", err)
992 }
993 return nil
994}
995
996// recordImageOwnership stamps one service's pinned image into the app's
997// owned-image list. Positive durable ownership is what later authorizes
998// runtime image prune; image labels alone never do.
999func recordImageOwnership(existing []domain.AppOwnedImage, p pinnedService) []domain.AppOwnedImage {
1000 entry := domain.AppOwnedImage{
1001 Service: p.spec.Name,
1002 Reference: p.spec.Image,
1003 Digest: p.digest,
1004 State: domain.AppResourceAttached,
1005 }
1006 out := append([]domain.AppOwnedImage(nil), existing...)
1007 for i, image := range out {
1008 if image.Service == entry.Service {
1009 out[i] = entry
1010 return out
1011 }
1012 }
1013 return append(out, entry)
1014}
1015
1016// failResult builds a preflight/create failure before any container exists.
1017func (s *Service) failResult(revision, before, after string, err error) ServiceResult {
1018 return ServiceResult{
1019 Result: "failed",
1020 EffectiveRevision: revision,
1021 Before: before,
1022 After: after,
1023 Error: err.Error(),
1024 }
1025}
1026
1027// deploymentReadiness resolves the readiness check a deploy or restart
1028// runs. Readiness never depends on how a service is replaced: a declared
1029// http, tcp, or log check is used as declared, and an undeclared check
1030// ("none" or unset) keeps the implicit HTTP probe on a service Gordon can
1031// probe over the published loopback bind — at least one effective-public
1032// HTTP interface, no L4 interface, no volume, and no bind. An undeclared
1033// probe on a stateful or L4 workload stays unchecked instead: an HTTP GET on
1034// the published port proves nothing there, so Gordon never guesses one.
1035//
1036// A stateless raw TCP service (see implicitTCPProbe) that declares no check
1037// receives the implicit TCP probe, so the overlap switch never routes to a
1038// container whose port does not accept yet.
1039func deploymentReadiness(spec domain.AppService) domain.AppService {
1040 switch {
1041 case implicitHTTPProbe(spec):
1042 spec.Readiness.Type = domain.AppReadinessHTTP
1043 case implicitTCPProbe(spec):
1044 spec.Readiness.Type = domain.AppReadinessTCP
1045 }
1046 return spec
1047}
1048
1049// implicitTCPProbe reports whether a service that declares no readiness check
1050// receives the implicit TCP probe: exactly one TCP interface, no HTTP or UDP
1051// interface, no volume, bind, or device. Such a service is replaced with an
1052// overlap, and its single published loopback bind is the probe target.
1053func implicitTCPProbe(spec domain.AppService) bool {
1054 if spec.Readiness.Type != "" && spec.Readiness.Type != domain.AppReadinessNone {
1055 return false
1056 }
1057 return len(spec.TCP) == 1 && len(spec.HTTP) == 0 && overlapEligible(spec)
1058}
1059
1060// implicitHTTPProbe reports whether a service that declares no readiness
1061// check receives the implicit HTTP probe when a deploy or restart starts it.
1062// Recovery verification of an already running generation keeps the declared
1063// readiness unchanged: it re-checks a generation it did not replace.
1064func implicitHTTPProbe(spec domain.AppService) bool {
1065 if spec.Readiness.Type != "" && spec.Readiness.Type != domain.AppReadinessNone {
1066 return false
1067 }
1068 if len(spec.HTTP) == 0 || len(spec.TCP) > 0 || len(spec.UDP) > 0 || len(spec.Volumes) > 0 || len(spec.Binds) > 0 {
1069 return false
1070 }
1071 for _, h := range spec.HTTP {
1072 if !h.IsPublic() {
1073 return false
1074 }
1075 }
1076 return true
1077}