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) {
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// File-delivered secrets are compared against the container's host files.
459// Values are compared in memory only and never logged.
460func (s *Service) envCurrent(ctx context.Context, app string, p pinnedService, container *domain.Container) bool {
461 desired, files, err := s.serviceEnv(ctx, app, p)
462 if err != nil {
463 return false
464 }
465 have := make(map[string]struct{}, len(container.Env))
466 for _, entry := range container.Env {
467 have[entry] = struct{}{}
468 }
469 for _, entry := range desired {
470 if _, ok := have[entry]; !ok {
471 return false
472 }
473 }
474 return s.secretFilesCurrent(ctx, app, container.Name, files)
475}
476
477// secretFilesCurrent reports whether every host secret file of the
478// container holds the current value. Values are compared in memory only.
479func (s *Service) secretFilesCurrent(ctx context.Context, app, container string, files map[string]string) bool {
480 if len(files) == 0 {
481 return true
482 }
483 if s.deps.SecretFiles == nil {
484 return false
485 }
486 for key, want := range files {
487 got, err := s.deps.SecretFiles.Read(ctx, app, container, key)
488 if err != nil || got != want {
489 return false
490 }
491 }
492 return true
493}
494
495// bindsCover reports whether the recorded loopback binds publish every
496// backend port the spec requires, so a kept container stays routable.
497func bindsCover(eff domain.AppEffectiveService, ports []domain.ContainerBackendPort) bool {
498 for _, port := range ports {
499 binds := eff.BackendBinds
500 if port.Protocol == domain.NetworkProtocolUDP {
501 binds = eff.UDPBackendBinds
502 }
503 if binds[port.ContainerPort] == 0 {
504 return false
505 }
506 }
507 return true
508}
509
510// failOperation records a terminal failure on an already claimed
511// journal, so a keyed repeat replays a terminal outcome instead of
512// reading a permanently in-flight claim. stepID names the failing phase
513// (a service step, a traffic publication, or a bare error). Journal-write
514// failures are logged, never returned: the caller already holds a more
515// specific error, and the next mutation's recovery pass converges state.
516func (s *Service) failOperation(ctx context.Context, op *domain.AppOperation, err error, stepID string) {
517 op.Outcome = domain.AppOutcomeFailed
518 op.Steps = append(op.Steps, domain.AppOperationStep{
519 ID: stepID, State: domain.AppStepFailed, Error: err.Error(),
520 })
521 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
522 log := zerowrap.FromCtx(ctx)
523 log.Warn().Err(saveErr).Str("op", op.Op).Msg("deployment: failed to record operation failure")
524 }
525}
526
527// failTrafficPublication records that the final graph apply was rejected.
528// The workload steps already ran and stay accurate, but the operation is a
529// failure: routing did not accept the published state, so a later replay
530// of the same key must never read success.
531func (s *Service) failTrafficPublication(ctx context.Context, op *domain.AppOperation, err error) {
532 s.failOperation(ctx, op, err, "traffic.publish")
533}
534
535// journaledOrNil reports a preflight that already answered the request
536// key: the stored journal when there is one, the bare error otherwise.
537func journaledOrNil(input DeployInput, op *domain.AppOperation, err error) (*DeployResult, error) {
538 if op == nil {
539 return nil, err
540 }
541 return journaledDeployResult(input, op), err
542}
543
544func journaledDeployResult(input DeployInput, op *domain.AppOperation) *DeployResult {
545 return &DeployResult{
546 Op: op.Op,
547 App: input.App,
548 Revision: op.InputRevision,
549 Outcome: op.Outcome,
550 }
551}
552
553// candidateJournal durably records a freshly created replacement container
554// before it can be started or join a network. An interrupted deployment then
555// leaves a trace boot recovery removes, instead of an untracked running
556// generation that could overlap the one recovery rebuilds.
557//
558// The window between the runtime creating a container and this write is the
559// irreducible one: a container ID cannot be recorded before it exists.
560// sync/atomic is not needed here: the journal write is synchronous.
561type candidateJournal func(ctx context.Context, containerID string) error
562
563// journalCandidate records one created candidate in the operation journal.
564// The step is addressed by index and read at write time, so a later append to
565// the operation's steps can never leave the recorder writing to a stale
566// element. A nil operation (an internal path with no journal) records nothing.
567func (s *Service) journalCandidate(op *domain.AppOperation, index int) candidateJournal {
568 if op == nil || index < 0 || index >= len(op.Steps) {
569 return nil
570 }
571 return func(ctx context.Context, containerID string) error {
572 op.Steps[index].After = containerID
573 if err := s.deps.State.SaveOperation(ctx, *op); err != nil {
574 return fmt.Errorf("deployment: record candidate %s: %w", containerID, err)
575 }
576 return nil
577 }
578}
579
580// deployService replaces one service sequentially and returns its terminal
581// result. before is the ACTIVE container being superseded (empty on a first
582// deployment); beforeGrace is its effective stop grace.
583//
584// The order is fixed: withdraw from traffic and confirm it, durably inhibit
585// a single-writer generation, retire the superseded container and confirm
586// it is gone, then create and start the replacement and wait for its
587// readiness probe. Nothing is created before the previous generation is
588// confirmed gone, so two generations never overlap; a withdrawal or
589// retirement that cannot be confirmed aborts without creating a candidate.
590func (s *Service) deployService(ctx context.Context, app, revision string, p pinnedService, opID, before string, beforeGrace time.Duration, journal candidateJournal) ServiceResult {
591 // Revalidate the current authorization and source before withdrawing or
592 // retiring the serving generation. A reload between preflight and this
593 // service step must fail without mutating the existing workload.
594 if _, err := s.resolveServiceBinds(app, p.spec); err != nil {
595 return s.failResult(revision, before, "", err)
596 }
597 if _, err := s.resolveServiceDevices(app, p.spec); err != nil {
598 return s.failResult(revision, before, "", err)
599 }
600 // Withdraw the service from traffic first and confirm it: a withdrawal
601 // that cannot be applied must block every container mutation, so no
602 // unverified generation stays routable. A first deployment has no
603 // published generation to withdraw.
604 if before != "" {
605 if err := s.withdrawForRecovery(ctx, app, p.name); err != nil {
606 return s.failResult(revision, before, "", err)
607 }
608 }
609 // A volume- or bind-owning replacement may write while the old
610 // generation still exists and could be revived by native restart policy.
611 // Inhibit that generation durably BEFORE the write can happen, so boot or
612 // periodic recovery can never restart the old writer on top of the new
613 // one. Cleared once the safe generation is published (publishService) or
614 // the operator removes the app.
615 if before != "" && singleWriterRequired(p.spec) {
616 if err := s.inhibitRecovery(ctx, app, p.name, before, domain.AppInhibitReplacementPending, opID); err != nil {
617 return s.failResult(revision, before, "", err)
618 }
619 }
620 if before != "" {
621 // The superseded generation must be confirmed gone before a new
622 // generation starts: a failed retirement aborts the replacement
623 // instead of allowing overlapping generations, and the generation
624 // keeps its inhibition and its claims.
625 retired := s.retireContainer(ctx, app, retireOptions{
626 Service: p.name, Grace: beforeGrace,
627 }, before)
628 if !retired.Gone {
629 return s.failResult(revision, before, "", fmt.Errorf("deployment: retire superseded container %s: %s", before, cleanupDetail(retired)))
630 }
631 }
632 candidate, err := s.createAndStart(ctx, app, revision, p, opID, journal)
633 if err != nil {
634 result := s.failResult(revision, before, "", err)
635 if candidate.Container != nil {
636 // The candidate exists: keep its ID in the terminal result so the
637 // journal and the operator can still trace it.
638 result.After = candidate.Container.ID
639 }
640 return result
641 }
642 created := candidate.Container
643 if err := s.waitServiceReady(ctx, app, created.ID, deploymentReadiness(p.spec), candidate.TCPBinds); err != nil {
644 // Capture redacted diagnostics while the failed replacement still
645 // exists, then remove only that candidate. The superseded container
646 // is already gone and is never recreated. Cleanup is best effort: a
647 // canceled context can leave the candidate behind, which is then
648 // reported as a leftover warning instead of a false success.
649 tail := s.redactDiagnostics(ctx, app, p, s.logTail(ctx, created.ID))
650 cleanup := s.retireCandidate(ctx, app, p.name, created.ID)
651 return ServiceResult{
652 Result: "failed",
653 EffectiveRevision: revision,
654 Before: before,
655 After: created.ID,
656 RestartUnsafe: singleWriterRequired(p.spec),
657 Error: err.Error(),
658 Diagnostics: tail,
659 CleanupWarnings: cleanup,
660 }
661 }
662 return ServiceResult{
663 Result: "deployed",
664 EffectiveRevision: revision,
665 Before: before,
666 After: created.ID,
667 RestartUnsafe: singleWriterRequired(p.spec),
668 BackendBinds: candidate.TCPBinds,
669 UDPBackendBinds: candidate.UDPBinds,
670 }
671}
672
673// singleWriterRequired reports services that must never have two generations
674// running concurrently. Persistent volumes and any bind (especially a
675// writable one) may be written by both, so they are treated identically.
676func singleWriterRequired(spec domain.AppService) bool {
677 return len(spec.Volumes) > 0 || len(spec.Binds) > 0
678}
679
680// refreshTraffic rebuilds the proxy host index and applies the full
681// HTTP/L4 graph after activation. A failure is fatal to the caller:
682// ACTIVE is published but the new service is not routable, so the
683// operation must not report success. Any service of this app whose
684// fail-closed withdrawal was never applied is withdrawn again first, so
685// a later publication never runs on top of unproven forwarding.
686func (s *Service) refreshTraffic(ctx context.Context, app string) error {
687 if s.deps.Traffic == nil {
688 return nil
689 }
690 for _, service := range s.publication.pending(app) {
691 if err := s.withdrawForRecovery(ctx, app, service); err != nil {
692 return err
693 }
694 }
695 if err := s.deps.Traffic.RebuildTraffic(ctx); err != nil {
696 return fmt.Errorf("deployment: rebuild traffic for %q: %w", app, err)
697 }
698 return nil
699}
700
701// publishService writes the per-service effective record once the
702// replacement passed readiness.
703func (s *Service) publishService(ctx context.Context, app, revision string, p pinnedService, result ServiceResult, opID string) error {
704 if app == "" {
705 return fmt.Errorf("deployment: publish active: empty app: %w", domain.ErrAppStateConflict)
706 }
707 active, ok, err := s.deps.State.LoadActive(ctx, app)
708 if err != nil {
709 return fmt.Errorf("deployment: load active: %w", err)
710 }
711 if !ok {
712 active = domain.AppActive{App: app, Services: map[string]domain.AppEffectiveService{}}
713 }
714 if active.Services == nil {
715 active.Services = map[string]domain.AppEffectiveService{}
716 }
717 activatedBy, activatedAt := opID, time.Now().UTC()
718 if previous, ok := active.Services[p.name]; ok && result.Result == ServiceResultUnchanged {
719 // A kept container keeps the activation of the operation that created it.
720 activatedBy, activatedAt = previous.ActivatedBy, previous.ActivatedAt
721 }
722 active.Services[p.name] = domain.AppEffectiveService{
723 EffectiveRevision: revision,
724 ActivatedBy: activatedBy,
725 ActivatedAt: activatedAt,
726 Image: p.spec.Image,
727 Digest: p.digest,
728 Container: result.After,
729 Spec: p.spec,
730 BackendBinds: result.BackendBinds,
731 UDPBackendBinds: result.UDPBackendBinds,
732 }
733 active.ConvergedRevision = revision
734 converged := true
735 for _, svc := range active.Services {
736 if svc.EffectiveRevision != revision {
737 converged = false
738 break
739 }
740 }
741 active.Converged = converged
742 if converged {
743 active.Networks = append([]domain.AppSharedNetwork(nil), p.appNetworks...)
744 }
745 if err := s.deps.State.SaveActive(ctx, active); err != nil {
746 return fmt.Errorf("deployment: publish active: %w", err)
747 }
748 if err := s.recordOwnership(ctx, app, p); err != nil {
749 return err
750 }
751 // The superseded generation is now safely replaced: clearing its
752 // inhibition is what makes this generation authoritative. A failed
753 // publication leaves the marker in place.
754 if result.Before != "" && result.Before != result.After {
755 if err := s.clearRecoveryInhibition(ctx, app, p.name, result.Before); err != nil {
756 return err
757 }
758 }
759 return nil
760}
761
762// recordOwnership stamps volume/secret/network ownership after publication.
763func (s *Service) recordOwnership(ctx context.Context, app string, p pinnedService) error {
764 ownership, err := s.deps.State.LoadOwnership(ctx, app)
765 if err != nil {
766 return err
767 }
768 if ownership.App == "" {
769 ownership.App = app
770 }
771 // ownership.ID arrives from LoadOwnership (store-assigned UUID);
772 // empty only for legacy records predating UUIDs.
773 volumes := append([]domain.AppOwnedVolume(nil), ownership.Volumes...)
774 seen := map[string]int{}
775 for i, vol := range volumes {
776 seen[vol.Name] = i
777 }
778 for _, vol := range p.spec.Volumes {
779 entry := domain.AppOwnedVolume{
780 Name: vol.Name,
781 Service: p.spec.Name,
782 RuntimeName: domain.RuntimeVolumeName(app, p.spec.Name, vol.Name),
783 State: domain.AppResourceAttached,
784 }
785 if idx, ok := seen[vol.Name]; ok {
786 volumes[idx] = entry
787 } else {
788 seen[vol.Name] = len(volumes)
789 volumes = append(volumes, entry)
790 }
791 }
792 ownership.Volumes = volumes
793 secrets := append([]domain.AppOwnedSecret(nil), ownership.Secrets...)
794 secretSeen := map[string]int{}
795 for i, secret := range secrets {
796 secretSeen[secret.Service+"\x00"+secret.Env] = i
797 }
798 for envKey, name := range p.spec.Secrets {
799 entry := domain.AppOwnedSecret{
800 Service: p.spec.Name,
801 Env: envKey,
802 Name: name,
803 Path: domain.AppSecretPathForID(ownership.ID, app, p.spec.Name, name),
804 State: domain.AppResourceAttached,
805 }
806 key := p.spec.Name + "\x00" + envKey
807 if idx, ok := secretSeen[key]; ok {
808 secrets[idx] = entry
809 } else {
810 secretSeen[key] = len(secrets)
811 secrets = append(secrets, entry)
812 }
813 }
814 ownership.Secrets = secrets
815 ownership.Images = recordImageOwnership(ownership.Images, p)
816 recordNetworkOwnership(&ownership, s.deps.Networks.Prefix, p.sharedNetworks)
817 if ownership.Services == nil {
818 ownership.Services = map[string]domain.AppServiceRecovery{}
819 }
820 ownership.Services[p.spec.Name] = domain.AppServiceRecovery{RestartUnsafe: singleWriterRequired(p.spec)}
821 if err := s.deps.State.SaveOwnership(ctx, ownership); err != nil {
822 return fmt.Errorf("deployment: record ownership: %w", err)
823 }
824 return nil
825}
826
827// recordImageOwnership stamps one service's pinned image into the app's
828// owned-image list. Positive durable ownership is what later authorizes
829// runtime image prune; image labels alone never do.
830func recordImageOwnership(existing []domain.AppOwnedImage, p pinnedService) []domain.AppOwnedImage {
831 entry := domain.AppOwnedImage{
832 Service: p.spec.Name,
833 Reference: p.spec.Image,
834 Digest: p.digest,
835 State: domain.AppResourceAttached,
836 }
837 out := append([]domain.AppOwnedImage(nil), existing...)
838 for i, image := range out {
839 if image.Service == entry.Service {
840 out[i] = entry
841 return out
842 }
843 }
844 return append(out, entry)
845}
846
847// failResult builds a preflight/create failure before any container exists.
848func (s *Service) failResult(revision, before, after string, err error) ServiceResult {
849 return ServiceResult{
850 Result: "failed",
851 EffectiveRevision: revision,
852 Before: before,
853 After: after,
854 Error: err.Error(),
855 }
856}
857
858// deploymentReadiness resolves the readiness check a deploy or restart
859// runs. Readiness never depends on how a service is replaced: a declared
860// http, tcp, or log check is used as declared, and an undeclared check
861// ("none" or unset) keeps the implicit HTTP probe on a service Gordon can
862// probe over the published loopback bind — at least one effective-public
863// HTTP interface, no L4 interface, no volume, and no bind. An undeclared
864// probe on a stateful or L4 workload stays unchecked instead: an HTTP GET on
865// the published port proves nothing there, so Gordon never guesses one.
866func deploymentReadiness(spec domain.AppService) domain.AppService {
867 if !implicitHTTPProbe(spec) {
868 return spec
869 }
870 spec.Readiness.Type = domain.AppReadinessHTTP
871 return spec
872}
873
874// implicitHTTPProbe reports whether a service that declares no readiness
875// check receives the implicit HTTP probe when a deploy or restart starts it.
876// Recovery verification of an already running generation keeps the declared
877// readiness unchanged: it re-checks a generation it did not replace.
878func implicitHTTPProbe(spec domain.AppService) bool {
879 if spec.Readiness.Type != "" && spec.Readiness.Type != domain.AppReadinessNone {
880 return false
881 }
882 if len(spec.HTTP) == 0 || len(spec.TCP) > 0 || len(spec.UDP) > 0 || len(spec.Volumes) > 0 || len(spec.Binds) > 0 {
883 return false
884 }
885 for _, h := range spec.HTTP {
886 if !h.IsPublic() {
887 return false
888 }
889 }
890 return true
891}