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