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 !domain.SameAppService(p.spec, eff.Spec):
493 return "spec changed"
494 case !bindsCover(eff, backendPorts(p.spec)):
495 return "backend binds missing"
496 case s.publication.inhibited(app, p.name):
497 return "withdrawal pending"
498 }
499 previous, err := s.deps.State.LoadRevision(ctx, app, eff.EffectiveRevision)
500 if err != nil {
501 return "load effective revision: " + err.Error()
502 }
503 if !maps.Equal(previous.Spec.Env, p.appEnv) {
504 return "app env changed"
505 }
506 if !reflect.DeepEqual(domain.AppServiceSharedNetworks(previous.Spec, p.name), p.sharedNetworks) {
507 return "shared networks changed"
508 }
509 if inhibited, err := s.recoveryInhibited(ctx, app, p.name, eff.Container); err != nil || inhibited {
510 return "recovery inhibited"
511 }
512 container, err := s.deps.Runtime.InspectContainer(ctx, eff.Container)
513 if err != nil || container.Status != string(domain.ContainerStatusRunning) {
514 return "container not running"
515 }
516 if !s.envCurrent(ctx, app, p, container) {
517 return "environment or secrets changed"
518 }
519 return ""
520}
521
522// envCurrent reports whether the running container was created with the
523// environment p would get now, including current secret values. The
524// runtime env also holds image-declared keys, so every desired entry must
525// be present; removed keys are caught by the spec and app env comparisons.
526// File-delivered secrets are compared against the container's host files.
527// Values are compared in memory only and never logged.
528func (s *Service) envCurrent(ctx context.Context, app string, p pinnedService, container *domain.Container) bool {
529 desired, files, err := s.serviceEnv(ctx, app, p)
530 if err != nil {
531 return false
532 }
533 have := make(map[string]struct{}, len(container.Env))
534 for _, entry := range container.Env {
535 have[entry] = struct{}{}
536 }
537 for _, entry := range desired {
538 if _, ok := have[entry]; !ok {
539 return false
540 }
541 }
542 return s.secretFilesCurrent(ctx, app, container.Name, files)
543}
544
545// secretFilesCurrent reports whether every host secret file of the
546// container holds the current value. Values are compared in memory only.
547func (s *Service) secretFilesCurrent(ctx context.Context, app, container string, files map[string]string) bool {
548 if len(files) == 0 {
549 return true
550 }
551 if s.deps.SecretFiles == nil {
552 return false
553 }
554 for key, want := range files {
555 got, err := s.deps.SecretFiles.Read(ctx, app, container, key)
556 if err != nil || got != want {
557 return false
558 }
559 }
560 return true
561}
562
563// bindsCover reports whether the recorded loopback binds publish every
564// backend port the spec requires, so a kept container stays routable.
565func bindsCover(eff domain.AppEffectiveService, ports []domain.ContainerBackendPort) bool {
566 for _, port := range ports {
567 binds := eff.BackendBinds
568 if port.Protocol == domain.NetworkProtocolUDP {
569 binds = eff.UDPBackendBinds
570 }
571 if binds[port.ContainerPort] == 0 {
572 return false
573 }
574 }
575 return true
576}
577
578// failOperation records a terminal failure on an already claimed
579// journal, so a keyed repeat replays a terminal outcome instead of
580// reading a permanently in-flight claim. stepID names the failing phase
581// (a service step, a traffic publication, or a bare error). Journal-write
582// failures are logged, never returned: the caller already holds a more
583// specific error, and the next mutation's recovery pass converges state.
584func (s *Service) failOperation(ctx context.Context, op *domain.AppOperation, err error, stepID string) {
585 op.Outcome = domain.AppOutcomeFailed
586 op.Steps = append(op.Steps, domain.AppOperationStep{
587 ID: stepID, State: domain.AppStepFailed, Error: err.Error(),
588 })
589 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
590 log := zerowrap.FromCtx(ctx)
591 log.Warn().Err(saveErr).Str("op", op.Op).Msg("deployment: failed to record operation failure")
592 }
593}
594
595// failTrafficPublication records that the final graph apply was rejected.
596// The workload steps already ran and stay accurate, but the operation is a
597// failure: routing did not accept the published state, so a later replay
598// of the same key must never read success.
599func (s *Service) failTrafficPublication(ctx context.Context, op *domain.AppOperation, err error) {
600 s.failOperation(ctx, op, err, "traffic.publish")
601}
602
603// journaledOrNil reports a preflight that already answered the request
604// key: the stored journal when there is one, the bare error otherwise.
605func journaledOrNil(input DeployInput, op *domain.AppOperation, err error) (*DeployResult, error) {
606 if op == nil {
607 return nil, err
608 }
609 return journaledDeployResult(input, op), err
610}
611
612func journaledDeployResult(input DeployInput, op *domain.AppOperation) *DeployResult {
613 return &DeployResult{
614 Op: op.Op,
615 App: input.App,
616 Revision: op.InputRevision,
617 Outcome: op.Outcome,
618 }
619}
620
621// candidateJournal durably records a freshly created replacement container
622// before it can be started or join a network. An interrupted deployment then
623// leaves a trace boot recovery removes, instead of an untracked running
624// generation that could overlap the one recovery rebuilds.
625//
626// The window between the runtime creating a container and this write is the
627// irreducible one: a container ID cannot be recorded before it exists.
628// sync/atomic is not needed here: the journal write is synchronous.
629type candidateJournal func(ctx context.Context, containerID string) error
630
631// journalCandidate records one created candidate in the operation journal.
632// The step is addressed by index and read at write time, so a later append to
633// the operation's steps can never leave the recorder writing to a stale
634// element. A nil operation (an internal path with no journal) records nothing.
635func (s *Service) journalCandidate(op *domain.AppOperation, index int) candidateJournal {
636 if op == nil || index < 0 || index >= len(op.Steps) {
637 return nil
638 }
639 return func(ctx context.Context, containerID string) error {
640 op.Steps[index].After = containerID
641 if err := s.deps.State.SaveOperation(ctx, *op); err != nil {
642 return fmt.Errorf("deployment: record candidate %s: %w", containerID, err)
643 }
644 return nil
645 }
646}
647
648// deployService replaces one service sequentially and returns its terminal
649// result. before is the ACTIVE container being superseded (empty on a first
650// deployment); beforeGrace is its effective stop grace. It serves every
651// service that is not overlap-eligible, a first deployment, and the
652// rebuild of a missing generation by start and restart; deployOverlap is the
653// overlapping alternative used by a deploy.
654//
655// The order is fixed: withdraw from traffic and confirm it, durably inhibit
656// a single-writer generation, retire the superseded container and confirm
657// it is gone, then create and start the replacement and wait for its
658// readiness probe. Nothing is created before the previous generation is
659// confirmed gone, so two generations never overlap; a withdrawal or
660// retirement that cannot be confirmed aborts without creating a candidate.
661func (s *Service) deployService(ctx context.Context, app, revision string, p pinnedService, opID, before string, beforeGrace time.Duration, journal candidateJournal) ServiceResult {
662 // Revalidate the current authorization and source before withdrawing or
663 // retiring the serving generation. A reload between preflight and this
664 // service step must fail without mutating the existing workload.
665 if failed, ok := s.revalidateService(app, revision, p, before); !ok {
666 return failed
667 }
668 // Withdraw the service from traffic first and confirm it: a withdrawal
669 // that cannot be applied must block every container mutation, so no
670 // unverified generation stays routable. A first deployment has no
671 // published generation to withdraw.
672 if before != "" {
673 if err := s.withdrawForRecovery(ctx, app, p.name); err != nil {
674 return s.failResult(revision, before, "", err)
675 }
676 }
677 // A volume- or bind-owning replacement may write while the old
678 // generation still exists and could be revived by native restart policy.
679 // Inhibit that generation durably BEFORE the write can happen, so boot or
680 // periodic recovery can never restart the old writer on top of the new
681 // one. Cleared once the safe generation is published (publishService) or
682 // the operator removes the app.
683 if before != "" && singleWriterRequired(p.spec) {
684 if err := s.inhibitRecovery(ctx, app, p.name, before, domain.AppInhibitReplacementPending, opID); err != nil {
685 return s.failResult(revision, before, "", err)
686 }
687 }
688 if before != "" {
689 // The superseded generation must be confirmed gone before a new
690 // generation starts: a failed retirement aborts the replacement
691 // instead of allowing overlapping generations, and the generation
692 // keeps its inhibition and its claims.
693 retired := s.retireContainer(ctx, app, retireOptions{
694 Service: p.name, Grace: beforeGrace,
695 }, before)
696 if !retired.Gone {
697 return s.failResult(revision, before, "", fmt.Errorf("deployment: retire superseded container %s: %s", before, cleanupDetail(retired)))
698 }
699 }
700 return s.startCandidate(ctx, app, revision, p, opID, before, journal)
701}
702
703// deployOverlap creates the replacement of an overlap-eligible service while
704// the superseded container keeps running and stays routed: no inhibition and
705// no retirement. Container names carry the operation, so the two generations
706// never collide, and their loopback binds are ephemeral.
707//
708// The only withdrawal it can perform is the retry of one already pending for
709// this service: an earlier withdrawal that could not be proven leaves the
710// service marked, and a marked service would be withdrawn again by the
711// traffic refresh that follows the switch, erasing the binds the switch just
712// published. The retry runs before the candidate exists, so a failure aborts
713// the deploy untouched.
714//
715// It returns the ready candidate with its binds, after durably recording the
716// retirement-pending marker of the superseded container; on any failure the
717// candidate is removed and the superseded container is left exactly as it was.
718// Retiring the superseded container is the caller's job, once the candidate is
719// published and routed (retireSuperseded).
720func (s *Service) deployOverlap(ctx context.Context, app, revision string, p pinnedService, opID, before string, beforeGrace time.Duration, journal candidateJournal) ServiceResult {
721 if failed, ok := s.revalidateService(app, revision, p, before); !ok {
722 return failed
723 }
724 if s.publication.inhibited(app, p.name) {
725 if err := s.withdrawForRecovery(ctx, app, p.name); err != nil {
726 return s.failResult(revision, before, "", err)
727 }
728 }
729 result := s.startCandidate(ctx, app, revision, p, opID, before, journal)
730 if result.Result != "deployed" {
731 return result
732 }
733 // The marker is durable before anything is published: whatever happens
734 // next, the superseded container is retired by the periodic
735 // reconciliation instead of being forgotten, and recovery never revives it.
736 if err := s.inhibitRetirement(ctx, app, p.name, before, opID, beforeGrace); err != nil {
737 result.Result = "failed"
738 result.Error = err.Error()
739 result.CleanupWarnings = append(result.CleanupWarnings, s.retireCandidate(ctx, app, p.name, result.After)...)
740 return result
741 }
742 result.SupersededRunning = true
743 return result
744}
745
746// revalidateService re-resolves the binds and devices of a service against
747// the current policy before any workload mutation. It returns the failed
748// result and false when either was revoked since preflight.
749func (s *Service) revalidateService(app, revision string, p pinnedService, before string) (ServiceResult, bool) {
750 if _, err := s.resolveServiceBinds(app, p.spec); err != nil {
751 return s.failResult(revision, before, "", err), false
752 }
753 if _, err := s.resolveServiceDevices(app, p.spec); err != nil {
754 return s.failResult(revision, before, "", err), false
755 }
756 return ServiceResult{}, true
757}
758
759// startCandidate creates and starts the replacement generation and waits for
760// its readiness probe. A candidate that fails readiness is removed after its
761// redacted diagnostics were captured; whatever generation it supersedes is
762// not touched here.
763func (s *Service) startCandidate(ctx context.Context, app, revision string, p pinnedService, opID, before string, journal candidateJournal) ServiceResult {
764 candidate, err := s.createAndStart(ctx, app, revision, p, opID, journal)
765 if err != nil {
766 result := s.failResult(revision, before, "", err)
767 if candidate.Container != nil {
768 // The candidate exists: keep its ID in the terminal result so the
769 // journal and the operator can still trace it.
770 result.After = candidate.Container.ID
771 }
772 return result
773 }
774 created := candidate.Container
775 if err := s.waitServiceReady(ctx, app, created.ID, deploymentReadiness(p.spec), candidate.TCPBinds); err != nil {
776 // Capture redacted diagnostics while the failed replacement still
777 // exists, then remove only that candidate. A superseded container
778 // that was already retired is never recreated. Cleanup is best
779 // effort: a canceled context can leave the candidate behind, which is
780 // then reported as a leftover warning instead of a false success.
781 tail := s.redactDiagnostics(ctx, app, p, s.logTail(ctx, created.ID))
782 cleanup := s.retireCandidate(ctx, app, p.name, created.ID)
783 return ServiceResult{
784 Result: "failed",
785 EffectiveRevision: revision,
786 Before: before,
787 After: created.ID,
788 RestartUnsafe: singleWriterRequired(p.spec),
789 Error: err.Error(),
790 Diagnostics: tail,
791 CleanupWarnings: cleanup,
792 }
793 }
794 return ServiceResult{
795 Result: "deployed",
796 EffectiveRevision: revision,
797 Before: before,
798 After: created.ID,
799 RestartUnsafe: singleWriterRequired(p.spec),
800 BackendBinds: candidate.TCPBinds,
801 UDPBackendBinds: candidate.UDPBinds,
802 VerifiedStartedAt: candidate.StartedAt,
803 }
804}
805
806// singleWriterRequired reports services that must never have two generations
807// running concurrently. Persistent volumes and any bind (especially a
808// writable one) may be written by both, so they are treated identically.
809// Devices and UDP interfaces are also excluded from the overlap (see
810// overlapEligible).
811func singleWriterRequired(spec domain.AppService) bool {
812 return len(spec.Volumes) > 0 || len(spec.Binds) > 0
813}
814
815// canOverlap reports whether a replacement may run beside its predecessor: both
816// the new spec and the predecessor's effective spec (as recorded in ACTIVE)
817// must be overlap-eligible. A predecessor that owned a volume, bind, device,
818// or UDP interface is a single writer or has pinned UDP sessions whatever the
819// replacement declares, so removing the resource from the spec does not make
820// the transition safe.
821func canOverlap(next, previous domain.AppService) bool {
822 return overlapEligible(next) && overlapEligible(previous)
823}
824
825// overlapEligible reports whether a service may be replaced with a short
826// overlap of the old and the new generation: it has no volume, bind, device,
827// or UDP interface. Volumes, binds, and devices are stateful or exclusive
828// resources two generations must not share. UDP sessions are pinned by the
829// relay to the backend they started on and outlive the switch for the
830// configured drain timeout, so they would keep reaching a backend that is
831// about to be stopped and whose loopback port may be recycled. TCP is fine:
832// established relays do not redial. Every other service keeps the sequential
833// replacement, so two generations of such a service never run together.
834func overlapEligible(spec domain.AppService) bool {
835 return len(spec.Volumes) == 0 && len(spec.Binds) == 0 && len(spec.Devices) == 0 && len(spec.UDP) == 0
836}
837
838// refreshTraffic rebuilds the proxy host index and applies the full
839// HTTP/L4 graph after activation. A failure is fatal to the caller:
840// ACTIVE is published but the new service is not routable, so the
841// operation must not report success. Any service of this app whose
842// fail-closed withdrawal was never applied is withdrawn again first, so
843// a later publication never runs on top of unproven forwarding.
844func (s *Service) refreshTraffic(ctx context.Context, app string) error {
845 if s.deps.Traffic == nil {
846 return nil
847 }
848 for _, service := range s.publication.pending(app) {
849 if err := s.withdrawForRecovery(ctx, app, service); err != nil {
850 return err
851 }
852 }
853 if err := s.deps.Traffic.RebuildTraffic(ctx); err != nil {
854 return fmt.Errorf("deployment: rebuild traffic for %q: %w", app, err)
855 }
856 return nil
857}
858
859// publishService writes the per-service effective record once the
860// replacement passed readiness.
861func (s *Service) publishService(ctx context.Context, app, revision string, p pinnedService, result ServiceResult, opID string) error {
862 if app == "" {
863 return fmt.Errorf("deployment: publish active: empty app: %w", domain.ErrAppStateConflict)
864 }
865 active, ok, err := s.deps.State.LoadActive(ctx, app)
866 if err != nil {
867 return fmt.Errorf("deployment: load active: %w", err)
868 }
869 if !ok {
870 active = domain.AppActive{App: app, Services: map[string]domain.AppEffectiveService{}}
871 }
872 if active.Services == nil {
873 active.Services = map[string]domain.AppEffectiveService{}
874 }
875 activatedBy, activatedAt := opID, time.Now().UTC()
876 if previous, ok := active.Services[p.name]; ok && result.Result == ServiceResultUnchanged {
877 // A kept container keeps the activation of the operation that created it.
878 activatedBy, activatedAt = previous.ActivatedBy, previous.ActivatedAt
879 }
880 active.Services[p.name] = domain.AppEffectiveService{
881 EffectiveRevision: revision,
882 ActivatedBy: activatedBy,
883 ActivatedAt: activatedAt,
884 Image: p.spec.Image,
885 Digest: p.digest,
886 Container: result.After,
887 Spec: p.spec,
888 BackendBinds: result.BackendBinds,
889 UDPBackendBinds: result.UDPBackendBinds,
890 }
891 active.ConvergedRevision = revision
892 converged := true
893 for _, svc := range active.Services {
894 if svc.EffectiveRevision != revision {
895 converged = false
896 break
897 }
898 }
899 active.Converged = converged
900 if converged {
901 active.Networks = append([]domain.AppSharedNetwork(nil), p.appNetworks...)
902 }
903 if err := s.deps.State.SaveActive(ctx, active); err != nil {
904 return fmt.Errorf("deployment: publish active: %w", err)
905 }
906 if err := s.recordOwnership(ctx, app, p); err != nil {
907 return err
908 }
909 return s.clearSupersededMarker(ctx, app, p.name, result)
910}
911
912// clearSupersededMarker clears the recovery inhibition of the generation a
913// sequential replacement superseded: that clearing is what makes the new
914// generation authoritative, and a failed publication leaves the marker in
915// place. An overlap replacement is the exception: its superseded container
916// still runs, and its retirement-pending marker is what gets it retired.
917func (s *Service) clearSupersededMarker(ctx context.Context, app, service string, result ServiceResult) error {
918 if result.Before == "" || result.Before == result.After || result.SupersededRunning {
919 return nil
920 }
921 return s.clearRecoveryInhibition(ctx, app, service, result.Before)
922}
923
924// recordOwnership stamps volume/secret/network ownership after publication.
925func (s *Service) recordOwnership(ctx context.Context, app string, p pinnedService) error {
926 ownership, err := s.deps.State.LoadOwnership(ctx, app)
927 if err != nil {
928 return err
929 }
930 if ownership.App == "" {
931 ownership.App = app
932 }
933 // ownership.ID arrives from LoadOwnership (store-assigned UUID);
934 // empty only for legacy records predating UUIDs.
935 volumes := append([]domain.AppOwnedVolume(nil), ownership.Volumes...)
936 seen := map[string]int{}
937 for i, vol := range volumes {
938 seen[vol.Name] = i
939 }
940 for _, vol := range p.spec.Volumes {
941 entry := domain.AppOwnedVolume{
942 Name: vol.Name,
943 Service: p.spec.Name,
944 RuntimeName: domain.RuntimeVolumeName(app, p.spec.Name, vol.Name),
945 State: domain.AppResourceAttached,
946 }
947 if idx, ok := seen[vol.Name]; ok {
948 volumes[idx] = entry
949 } else {
950 seen[vol.Name] = len(volumes)
951 volumes = append(volumes, entry)
952 }
953 }
954 ownership.Volumes = volumes
955 secrets := append([]domain.AppOwnedSecret(nil), ownership.Secrets...)
956 secretSeen := map[string]int{}
957 for i, secret := range secrets {
958 secretSeen[secret.Service+"\x00"+secret.Env] = i
959 }
960 for envKey, name := range p.spec.Secrets {
961 entry := domain.AppOwnedSecret{
962 Service: p.spec.Name,
963 Env: envKey,
964 Name: name,
965 Path: domain.AppSecretPathForID(ownership.ID, app, p.spec.Name, name),
966 State: domain.AppResourceAttached,
967 }
968 key := p.spec.Name + "\x00" + envKey
969 if idx, ok := secretSeen[key]; ok {
970 secrets[idx] = entry
971 } else {
972 secretSeen[key] = len(secrets)
973 secrets = append(secrets, entry)
974 }
975 }
976 ownership.Secrets = secrets
977 ownership.Images = recordImageOwnership(ownership.Images, p)
978 recordNetworkOwnership(&ownership, s.deps.Networks.Prefix, p.sharedNetworks)
979 if ownership.Services == nil {
980 ownership.Services = map[string]domain.AppServiceRecovery{}
981 }
982 ownership.Services[p.spec.Name] = domain.AppServiceRecovery{RestartUnsafe: singleWriterRequired(p.spec)}
983 if err := s.deps.State.SaveOwnership(ctx, ownership); err != nil {
984 return fmt.Errorf("deployment: record ownership: %w", err)
985 }
986 return nil
987}
988
989// recordImageOwnership stamps one service's pinned image into the app's
990// owned-image list. Positive durable ownership is what later authorizes
991// runtime image prune; image labels alone never do.
992func recordImageOwnership(existing []domain.AppOwnedImage, p pinnedService) []domain.AppOwnedImage {
993 entry := domain.AppOwnedImage{
994 Service: p.spec.Name,
995 Reference: p.spec.Image,
996 Digest: p.digest,
997 State: domain.AppResourceAttached,
998 }
999 out := append([]domain.AppOwnedImage(nil), existing...)
1000 for i, image := range out {
1001 if image.Service == entry.Service {
1002 out[i] = entry
1003 return out
1004 }
1005 }
1006 return append(out, entry)
1007}
1008
1009// failResult builds a preflight/create failure before any container exists.
1010func (s *Service) failResult(revision, before, after string, err error) ServiceResult {
1011 return ServiceResult{
1012 Result: "failed",
1013 EffectiveRevision: revision,
1014 Before: before,
1015 After: after,
1016 Error: err.Error(),
1017 }
1018}
1019
1020// deploymentReadiness resolves the readiness check a deploy or restart
1021// runs. Readiness never depends on how a service is replaced: a declared
1022// http, tcp, or log check is used as declared, and an undeclared check
1023// ("none" or unset) keeps the implicit HTTP probe on a service Gordon can
1024// probe over the published loopback bind — at least one effective-public
1025// HTTP interface, no L4 interface, no volume, and no bind. An undeclared
1026// probe on a stateful or L4 workload stays unchecked instead: an HTTP GET on
1027// the published port proves nothing there, so Gordon never guesses one.
1028//
1029// A stateless raw TCP service (see implicitTCPProbe) that declares no check
1030// receives the implicit TCP probe, so the overlap switch never routes to a
1031// container whose port does not accept yet.
1032func deploymentReadiness(spec domain.AppService) domain.AppService {
1033 switch {
1034 case implicitHTTPProbe(spec):
1035 spec.Readiness.Type = domain.AppReadinessHTTP
1036 case implicitTCPProbe(spec):
1037 spec.Readiness.Type = domain.AppReadinessTCP
1038 }
1039 return spec
1040}
1041
1042// implicitTCPProbe reports whether a service that declares no readiness check
1043// receives the implicit TCP probe: exactly one TCP interface, no HTTP or UDP
1044// interface, no volume, bind, or device. Such a service is replaced with an
1045// overlap, and its single published loopback bind is the probe target.
1046func implicitTCPProbe(spec domain.AppService) bool {
1047 if spec.Readiness.Type != "" && spec.Readiness.Type != domain.AppReadinessNone {
1048 return false
1049 }
1050 return len(spec.TCP) == 1 && len(spec.HTTP) == 0 && overlapEligible(spec)
1051}
1052
1053// implicitHTTPProbe reports whether a service that declares no readiness
1054// check receives the implicit HTTP probe when a deploy or restart starts it.
1055// Recovery verification of an already running generation keeps the declared
1056// readiness unchanged: it re-checks a generation it did not replace.
1057func implicitHTTPProbe(spec domain.AppService) bool {
1058 if spec.Readiness.Type != "" && spec.Readiness.Type != domain.AppReadinessNone {
1059 return false
1060 }
1061 if len(spec.HTTP) == 0 || len(spec.TCP) > 0 || len(spec.UDP) > 0 || len(spec.Volumes) > 0 || len(spec.Binds) > 0 {
1062 return false
1063 }
1064 for _, h := range spec.HTTP {
1065 if !h.IsPublic() {
1066 return false
1067 }
1068 }
1069 return true
1070}