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