1package deployment
2
3import (
4 "bufio"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "sort"
10 "strings"
11 "time"
12
13 "github.com/bnema/zerowrap"
14
15 "github.com/bnema/gordon/internal/domain"
16)
17
18// failLogTailLines bounds the log tail attached to service failures.
19const failLogTailLines = 50
20
21// failLogTailTimeout bounds the diagnostics read of a failing container.
22const failLogTailTimeout = 10 * time.Second
23
24// Deploy executes a preflighted revision: fail-fast across services in
25// sorted name order. Every service is replaced sequentially — withdraw
26// traffic, retire the superseded container, start the replacement and wait
27// for its readiness probe — so a deployment may briefly interrupt the
28// service, and two Gordon-managed generations of one service never run at
29// the same time. Volumes are never deleted: no RemoveVolume call, no
30// volume-deletion flags on container removal.
31//
32// It is the synchronous composition of the two engine phases: StartDeploy
33// claims the key and journals the operation, then ExecuteDeploy runs the
34// effects for the owner only.
35func (s *Service) Deploy(ctx context.Context, input DeployInput) (*DeployResult, error) {
36 started, err := s.StartDeploy(ctx, input)
37 if err != nil {
38 return nil, err
39 }
40 if !started.Owned {
41 // The key already answered this request: its stored journal is the
42 // result and no workload is touched again.
43 return journaledOrNil(input, &started.Claim.Journal, started.ReplayError())
44 }
45 return s.ExecuteDeploy(ctx, started.Claim)
46}
47
48// DeployClaim is the durable identity of one started deploy operation. It is
49// the hand-off between the two phases: StartDeploy returns it and
50// ExecuteDeploy consumes it. App and Op are the durable identity, so a
51// caller that reconstructs the claim from them alone (for example after a
52// restart, with no in-process state) leaves Journal and Resolved empty and
53// ExecuteDeploy loads and resolves both from the store.
54type DeployClaim struct {
55 App string
56 Op string
57 Service string
58 Revision string
59 // Resolved is the revision StartDeploy resolved for this claim. Empty
60 // when the claim was reconstructed from identity alone.
61 Resolved domain.AppDesiredRevision
62 // Journal is the claimed operation held in process. Its Op is empty when
63 // the claim was reconstructed from identity alone.
64 Journal domain.AppOperation
65}
66
67// StartDeployResult reports whether this caller owns the claimed operation.
68// Owned is true only for the single caller that claimed Op. Every idempotent
69// replay (the same key already answered) reports Owned false, and the stored
70// journal in Claim.Journal is the result. Only an owner may pass the claim to
71// ExecuteDeploy.
72type StartDeployResult struct {
73 Claim DeployClaim
74 Owned bool
75}
76
77// ReplayError is the explicit disposition of a claim this caller does not
78// own: nil for a terminal success, and a conflict for an in-flight,
79// interrupted, or failed operation, so a replay is never mistaken for a
80// second execution. It is nil for an owned operation.
81func (r StartDeployResult) ReplayError() error {
82 if r.Owned {
83 return nil
84 }
85 return replayError(r.Claim.Journal)
86}
87
88// StartDeploy performs the first phase of a deploy. Under the app lock it
89// converges any interrupted predecessor, resolves the request to a revision
90// and runs the targeted-deploy convergence checks, atomically claims the
91// request key, and persists the non-terminal journal. It performs no image
92// pull, pinning, or workload mutation, so a crash right after it leaves a
93// durable claim that reconciliation can converge.
94//
95// The returned claim distinguishes the newly owned operation (Owned true)
96// from an idempotent replay (Owned false): the same key is never handed to
97// two owners, and a replay never executes effects again.
98func (s *Service) StartDeploy(ctx context.Context, input DeployInput) (*StartDeployResult, error) {
99 // Planning and claiming acquire no runtime resource, so StartDeploy takes
100 // only the per-app coordinator. ExecuteDeploy takes the GC shared lease
101 // across resource acquisition and publication, where it is required.
102 release, err := s.coord.acquire(ctx, input.App)
103 if err != nil {
104 return nil, err
105 }
106 defer release()
107 return s.startDeployLocked(ctx, input)
108}
109
110func (s *Service) startDeployLocked(ctx context.Context, input DeployInput) (*StartDeployResult, error) {
111 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
112 zerowrap.FieldLayer: "usecase",
113 zerowrap.FieldUseCase: "StartDeploy",
114 "app": input.App,
115 })
116 log := zerowrap.FromCtx(ctx)
117
118 if err := s.deps.State.Recover(ctx); err != nil {
119 return nil, fmt.Errorf("deployment: recover before deploy: %w", err)
120 }
121 // A mutation of this app never starts on top of an interrupted one: its
122 // never-published candidates must be gone before this mutation creates a
123 // generation of the same service. ACTIVE is materialized first, so the
124 // published generation cannot be mistaken for a leftover.
125 if err := s.reconcileInterruptedDeploy(ctx, input.App); err != nil {
126 return nil, err
127 }
128 rev, op, owned, err := s.claimDeploymentLocked(ctx, input)
129 if err != nil {
130 return nil, err
131 }
132 if owned {
133 // The claimed operation is now live in this process. It is
134 // registered before the caller releases the coordinator, so a
135 // foreground mutation that takes the lock in the gap before
136 // ExecuteDeploy cannot reconcile the claim away.
137 s.markOperationLive(input.App, op.Op)
138 } else {
139 log.Info().Str("op", op.Op).Msg("deployment: replayed an already claimed operation")
140 }
141 return &StartDeployResult{
142 Claim: DeployClaim{
143 App: input.App, Op: op.Op, Service: input.Service, Revision: input.Revision,
144 Resolved: rev, Journal: op,
145 },
146 Owned: owned,
147 }, nil
148}
149
150// ExecuteDeploy performs the second phase of a deploy: it reacquires the app
151// lock, loads and verifies the claimed journal by app/op identity, runs the
152// image pull/pin and the sequential replacement, and records the terminal
153// outcome. A claim that is already terminal, or that does not match its
154// identity, is never executed: it replays its stored outcome instead. An
155// operation that stops mid-execution (shutdown or cancellation) leaves its
156// non-terminal journal behind for reconciliation, so it can never run twice.
157//
158// The operation stays live until this returns: the store record, not the
159// in-process claim, is what executes, and the live marker is cleared only
160// after the outcome was recorded or a recoverable non-terminal claim was
161// left behind, while the app lock is still held.
162func (s *Service) ExecuteDeploy(ctx context.Context, claim DeployClaim) (*DeployResult, error) {
163 release, err := s.acquireAppContext(ctx, claim.App)
164 if err != nil {
165 // The claim is left as persisted: recoverable, but no longer owned
166 // here, so a later reconciliation can converge it.
167 s.clearOperationLive(claim.App, claim.Op)
168 return nil, err
169 }
170 defer release()
171 // Registered after release, so it runs first: the marker is dropped
172 // before the app lock is released, and never while a live goroutine may
173 // still persist state.
174 defer s.clearOperationLive(claim.App, claim.Op)
175 return s.executeLocked(ctx, claim)
176}
177
178// AbandonDeploy settles a claim this process owns but will never execute, such
179// as when daemon shutdown races the hand-off between StartDeploy and
180// ExecuteDeploy. It drops the in-process live marker and converges the still
181// non-terminal journal as an interrupted operation, so no durable claim stays
182// marked live forever and a later reconciliation reads a terminal outcome
183// instead of a permanently in-flight key. It takes the app lock but performs
184// no workload effect.
185func (s *Service) AbandonDeploy(ctx context.Context, claim DeployClaim) error {
186 if claim.App == "" || claim.Op == "" {
187 return fmt.Errorf("deployment: abandon requires an app and operation identity: %w", domain.ErrAppStateConflict)
188 }
189 // The marker is dropped before any fallible step: even if convergence
190 // fails, the claim is no longer owned here, so a later reconciliation
191 // (foreground or boot) can converge it instead of skipping it as live.
192 s.clearOperationLive(claim.App, claim.Op)
193 release, err := s.coord.acquire(ctx, claim.App)
194 if err != nil {
195 return err
196 }
197 defer release()
198 return s.reconcileInterruptedDeploy(ctx, claim.App)
199}
200
201// executeLocked runs the claimed, non-terminal operation. The caller holds
202// the app lock. Every failure that happens before a terminal outcome is
203// recorded (load active, revision, service step, traffic) leaves the journal
204// with its step state, never a silent success.
205func (s *Service) executeLocked(ctx context.Context, claim DeployClaim) (*DeployResult, error) {
206 input := DeployInput{App: claim.App, Revision: claim.Revision, Service: claim.Service}
207 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
208 zerowrap.FieldLayer: "usecase",
209 zerowrap.FieldUseCase: "ExecuteDeploy",
210 "app": claim.App,
211 "op": claim.Op,
212 })
213 log := zerowrap.FromCtx(ctx)
214
215 op, err := s.loadClaim(ctx, claim)
216 if err != nil {
217 return nil, err
218 }
219 if op.Terminal() {
220 // The claim was finalized between the phases: never execute a settled
221 // key, replay its stored outcome instead.
222 return journaledDeployResult(input, &op), replayError(op)
223 }
224 // The claimed revision is authoritative for the whole execution: it was
225 // resolved when the key was claimed and is what the journal records, so a
226 // desired-state change between the two phases must never redirect this
227 // operation to a different revision than the one it pinned.
228 rev, err := s.claimRevision(ctx, claim, op)
229 if err != nil {
230 s.failOperation(ctx, &op, err, "error")
231 return journaledDeployResult(input, &op), err
232 }
233 pinned, err := s.pinPreflightLocked(ctx, input.App, input.Service, rev, &op)
234 if err != nil {
235 return journaledDeployResult(input, &op), err
236 }
237 sort.Slice(pinned, func(i, j int) bool { return pinned[i].name < pinned[j].name })
238
239 active, _, err := s.deps.State.LoadActive(ctx, input.App)
240 if err != nil {
241 loadErr := fmt.Errorf("deployment: load active: %w", err)
242 s.failOperation(ctx, &op, loadErr, "error")
243 return journaledDeployResult(input, &op), loadErr
244 }
245
246 result := &DeployResult{
247 Op: op.Op,
248 App: input.App,
249 Revision: rev.Revision,
250 Services: map[string]ServiceResult{},
251 }
252 // A targeted deploy is intentionally a partial plan: services omitted
253 // from pinned remain active. Full deploys reconcile actual removals.
254 if input.Service == "" {
255 if err := s.reconcileRemovalsForDeploy(ctx, input.App, active, pinned, &op, result, log); err != nil {
256 return result, err
257 }
258 }
259 for i, p := range pinned {
260 if err := s.runServiceStep(ctx, input.App, rev.Revision, i, p, &op, active, result, log); err != nil {
261 return result, err
262 }
263 }
264 op.Outcome = ComputeOutcome(result.Services)
265 op.Warnings = journalWarnings(collectCleanupWarnings(result.Services))
266 result.CleanupWarnings = collectCleanupWarnings(result.Services)
267 if saveErr := s.deps.State.SaveOperation(ctx, op); saveErr != nil {
268 log.Warn().Err(saveErr).Msg("deployment: failed to record deploy outcome")
269 }
270 return result, nil
271}
272
273// loadClaim verifies the claimed journal by app/op identity. The store is
274// authoritative: the operation is always reloaded by app/op ID, so a stale
275// in-process journal can never be executed (for example when another actor
276// finalized the key between the claim and execution). A journal that does
277// not match the identity, or is not a deploy, is refused.
278func (s *Service) loadClaim(ctx context.Context, claim DeployClaim) (domain.AppOperation, error) {
279 if claim.App == "" || claim.Op == "" {
280 return domain.AppOperation{}, fmt.Errorf("deployment: execute requires an app and operation identity: %w", domain.ErrAppStateConflict)
281 }
282 op, err := s.deps.State.LoadOperation(ctx, claim.App, claim.Op)
283 if err != nil {
284 return domain.AppOperation{}, fmt.Errorf("deployment: load claimed operation %q: %w", claim.Op, err)
285 }
286 if op.App != claim.App || op.Op != claim.Op {
287 return domain.AppOperation{}, fmt.Errorf("deployment: claim does not identify app %q operation %q: %w", claim.App, claim.Op, domain.ErrAppStateConflict)
288 }
289 if op.Kind != "deploy" {
290 return domain.AppOperation{}, fmt.Errorf("deployment: operation %q is %q, not a deploy: %w", claim.Op, op.Kind, domain.ErrAppStateConflict)
291 }
292 return op, nil
293}
294
295// claimRevision returns the revision StartDeploy captured. A claim
296// reconstructed from identity alone resolves the revision its journal
297// recorded, so a resumed execution pins the revision it was started for.
298func (s *Service) claimRevision(ctx context.Context, claim DeployClaim, op domain.AppOperation) (domain.AppDesiredRevision, error) {
299 if claim.Resolved.Revision != "" {
300 return claim.Resolved, nil
301 }
302 return s.resolveRevision(ctx, DeployInput{App: claim.App, Revision: op.InputRevision, Service: claim.Service})
303}
304
305// runServiceStep executes one pinned service in the journal: replace it,
306// publish its ACTIVE entry, then publish traffic for the new container. The
307// step is recorded as succeeded only after the runtime effect, the ACTIVE
308// publication, and the traffic apply all succeeded, so a journaled success
309// is never ahead of what is routed.
310func (s *Service) runServiceStep(
311 ctx context.Context,
312 app, revision string,
313 index int,
314 p pinnedService,
315 op *domain.AppOperation,
316 active domain.AppActive,
317 result *DeployResult,
318 log zerowrap.Logger,
319) error {
320 before := ""
321 if eff, ok := active.Services[p.name]; ok {
322 before = eff.Container
323 }
324 stepID := "service." + p.name + ".replace"
325 // The step is journaled before the replacement runs and carries the
326 // candidate's container ID as soon as it exists, so an interrupted deploy
327 // leaves a trace boot recovery can clean up instead of an untracked
328 // generation.
329 step := &op.Steps[index+1]
330 *step = domain.AppOperationStep{
331 ID: stepID, Service: p.name,
332 Digest: p.digest, Image: p.runtimeImage,
333 Before: before, State: domain.AppStepPending,
334 }
335 svcResult := s.deployService(ctx, app, revision, p, op.Op, before, activeStopGrace(active, p.name), s.journalCandidate(op, index+1))
336 result.Services[p.name] = svcResult
337 // A recorded candidate is authoritative while the step is unresolved: a
338 // replacement that failed after creating its container must stay traceable
339 // for reconciliation.
340 if svcResult.After != "" {
341 step.After = svcResult.After
342 }
343 fail := func(err error) error {
344 step.State = domain.AppStepFailed
345 step.Error = err.Error()
346 step.Diagnostics = svcResult.Diagnostics
347 op.Warnings = journalWarnings(collectCleanupWarnings(result.Services))
348 op.Outcome = ComputeOutcome(result.Services)
349 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
350 log.Warn().Err(saveErr).Msg("deployment: failed to record service failure")
351 }
352 return err
353 }
354 if svcResult.Result == "failed" {
355 // Fail fast: later services stay unchanged.
356 return fail(fmt.Errorf("deployment: service %q failed at %s: %s: %w",
357 p.name, stepID, svcResult.Error, domain.ErrAppStateConflict))
358 }
359 // Publish the per-service effective state: the new binds become the
360 // service's recorded backend before the proxy is repointed at them.
361 if err := s.publishService(ctx, app, revision, p, svcResult, op.Op); err != nil {
362 svcResult.Result = "failed"
363 svcResult.Error = err.Error()
364 result.Services[p.name] = svcResult
365 return fail(err)
366 }
367 // Rebuild and publish the traffic graph. ACTIVE already names the new
368 // container, but the operation is a failure until routing accepted it,
369 // so a rejected graph must never be journaled as success.
370 if err := s.refreshTraffic(ctx, app); err != nil {
371 svcResult.Result = "failed"
372 svcResult.Error = err.Error()
373 result.Services[p.name] = svcResult
374 return fail(err)
375 }
376 result.Services[p.name] = svcResult
377 step.State = domain.AppStepSucceeded
378 step.Error = ""
379 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
380 log.Warn().Err(saveErr).Msg("deployment: failed to checkpoint service progress")
381 }
382 return nil
383}
384
385// failOperation records a terminal failure on an already claimed
386// journal, so a keyed repeat replays a terminal outcome instead of
387// reading a permanently in-flight claim. stepID names the failing phase
388// (a service step, a traffic publication, or a bare error). Journal-write
389// failures are logged, never returned: the caller already holds a more
390// specific error, and the next mutation's recovery pass converges state.
391func (s *Service) failOperation(ctx context.Context, op *domain.AppOperation, err error, stepID string) {
392 op.Outcome = domain.AppOutcomeFailed
393 op.Steps = append(op.Steps, domain.AppOperationStep{
394 ID: stepID, State: domain.AppStepFailed, Error: err.Error(),
395 })
396 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
397 log := zerowrap.FromCtx(ctx)
398 log.Warn().Err(saveErr).Str("op", op.Op).Msg("deployment: failed to record operation failure")
399 }
400}
401
402// failTrafficPublication records that the final graph apply was rejected.
403// The workload steps already ran and stay accurate, but the operation is a
404// failure: routing did not accept the published state, so a later replay
405// of the same key must never read success.
406func (s *Service) failTrafficPublication(ctx context.Context, op *domain.AppOperation, err error) {
407 s.failOperation(ctx, op, err, "traffic.publish")
408}
409
410// journaledOrNil reports a preflight that already answered the request
411// key: the stored journal when there is one, the bare error otherwise.
412func journaledOrNil(input DeployInput, op *domain.AppOperation, err error) (*DeployResult, error) {
413 if op == nil {
414 return nil, err
415 }
416 return journaledDeployResult(input, op), err
417}
418
419func journaledDeployResult(input DeployInput, op *domain.AppOperation) *DeployResult {
420 return &DeployResult{
421 Op: op.Op,
422 App: input.App,
423 Revision: op.InputRevision,
424 Outcome: op.Outcome,
425 }
426}
427
428// reconcileRemovalsForDeploy journals and applies the removal of services
429// the new revision no longer declares. A failure marks the operation
430// failed and aborts before any new state is published.
431func (s *Service) reconcileRemovalsForDeploy(ctx context.Context, app string, active domain.AppActive, pinned []pinnedService, op *domain.AppOperation, result *DeployResult, log zerowrap.Logger) error {
432 removalSteps, removed, warnings, err := s.reconcileRemovedServices(ctx, app, active, pinned)
433 op.Warnings = append(op.Warnings, journalWarnings(warnings)...)
434 result.CleanupWarnings = append(result.CleanupWarnings, warnings...)
435 if err != nil {
436 op.Steps = append(op.Steps, removalSteps...)
437 op.Outcome = domain.AppOutcomeFailed
438 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
439 log.Warn().Err(saveErr).Msg("deployment: failed to record removal failure")
440 }
441 return err
442 }
443 if len(removalSteps) == 0 {
444 return nil
445 }
446 op.Steps = append(op.Steps, removalSteps...)
447 result.Removed = removed
448 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
449 log.Warn().Err(saveErr).Msg("deployment: failed to record service removals")
450 }
451 return nil
452}
453
454// reconcileRemovedServices withdraws and retires every ACTIVE service the
455// new revision no longer declares. Traffic is withdrawn first (fail
456// closed), then the exact container is inhibited, stopped, and removed
457// with its data retained, its backend claims released, and its ACTIVE
458// entry deleted. A failure leaves the remaining services untouched and
459// stops the deploy before any new state is published.
460func (s *Service) reconcileRemovedServices(ctx context.Context, app string, active domain.AppActive, pinned []pinnedService) ([]domain.AppOperationStep, []string, []CleanupWarning, error) {
461 desired := make(map[string]struct{}, len(pinned))
462 for _, p := range pinned {
463 desired[p.name] = struct{}{}
464 }
465 var names []string
466 for name := range active.Services {
467 if _, ok := desired[name]; !ok {
468 names = append(names, name)
469 }
470 }
471 sort.Strings(names)
472
473 steps := make([]domain.AppOperationStep, 0, len(names))
474 var cleanupWarnings []CleanupWarning
475 for _, name := range names {
476 eff := active.Services[name]
477 step, warnings, err := s.retireRemovedService(ctx, app, name, eff)
478 steps = append(steps, step)
479 cleanupWarnings = append(cleanupWarnings, warnings...)
480 if err != nil {
481 return steps, names, cleanupWarnings, err
482 }
483 delete(active.Services, name)
484 }
485 if len(names) == 0 {
486 return nil, nil, nil, nil
487 }
488 if err := s.deps.State.SaveActive(ctx, active); err != nil {
489 return steps, names, cleanupWarnings, fmt.Errorf("deployment: persist service removals: %w", err)
490 }
491 return steps, names, cleanupWarnings, nil
492}
493
494// retireRemovedService withdraws one removed service and removes its exact
495// container, releasing its claims only after withdrawal succeeded.
496func (s *Service) retireRemovedService(ctx context.Context, app, name string, eff domain.AppEffectiveService) (domain.AppOperationStep, []CleanupWarning, error) {
497 step := domain.AppOperationStep{
498 ID: "service." + name + ".remove",
499 State: domain.AppStepPending,
500 Service: name,
501 Before: eff.Container,
502 }
503 fail := func(err error) (domain.AppOperationStep, []CleanupWarning, error) {
504 step.State = domain.AppStepFailed
505 step.Error = err.Error()
506 return step, nil, err
507 }
508 if err := s.withdrawForRecovery(ctx, app, name); err != nil {
509 return fail(err)
510 }
511 if eff.Container != "" {
512 if err := s.inhibitRecovery(ctx, app, name, eff.Container, "removed", ""); err != nil {
513 return fail(err)
514 }
515 retired := s.retireContainer(ctx, app, retireOptions{
516 Service: name, Grace: serviceStopGrace(eff), ClearInhibition: true,
517 }, eff.Container)
518 if !retired.Gone {
519 return fail(fmt.Errorf("deployment: remove %s/%s container %s: %s", app, name, eff.Container, cleanupDetail(retired)))
520 }
521 step.State = domain.AppStepSucceeded
522 step.After = ""
523 return step, retired.Warnings, nil
524 }
525 step.State = domain.AppStepSucceeded
526 step.After = ""
527 return step, nil, nil
528}
529
530// candidateJournal durably records a freshly created replacement container
531// before it can be started or join a network. An interrupted deployment then
532// leaves a trace boot recovery removes, instead of an untracked running
533// generation that could overlap the one recovery rebuilds.
534//
535// The window between the runtime creating a container and this write is the
536// irreducible one: a container ID cannot be recorded before it exists.
537// sync/atomic is not needed here: the journal write is synchronous.
538type candidateJournal func(ctx context.Context, containerID string) error
539
540// journalCandidate records one created candidate in the operation journal.
541// The step is addressed by index and read at write time, so a later append to
542// the operation's steps can never leave the recorder writing to a stale
543// element. A nil operation (an internal path with no journal) records nothing.
544func (s *Service) journalCandidate(op *domain.AppOperation, index int) candidateJournal {
545 if op == nil || index < 0 || index >= len(op.Steps) {
546 return nil
547 }
548 return func(ctx context.Context, containerID string) error {
549 op.Steps[index].After = containerID
550 if err := s.deps.State.SaveOperation(ctx, *op); err != nil {
551 return fmt.Errorf("deployment: record candidate %s: %w", containerID, err)
552 }
553 return nil
554 }
555}
556
557// reconcileInterruptedDeploy converges the journal of a deployment that a
558// crash, shutdown, or failed cleanup interrupted. A candidate that was
559// created but never published is removed and its loopback claims released, so
560// recovery can rebuild the published generation without ever running two
561// generations of one service. The interrupted operation is then finalized as
562// failed: a repeat of its key replays an explicit failure instead of a
563// permanent in-flight claim.
564//
565// It also retries the candidates of a failed operation whose cleanup did not
566// complete, so a leftover stays visible and is eventually removed. The
567// published generation recorded in ACTIVE is never touched.
568//
569// It runs at boot and before every mutation of the app, always under the app
570// lock, so a non-terminal journal is definitively interrupted, and a mutation
571// that cannot reconcile a predecessor does not proceed. It reads the
572// operation journal first and only then ACTIVE, so the common case of a
573// successful operation costs one state read.
574//
575// Reconciliation fails closed on a candidate that cannot be removed: an
576// orphan that may still run aborts the caller instead of being journaled as a
577// recoverable leftover. Because the journal is never finalized while such a
578// leftover exists, it stays the latest operation and a newer operation can
579// never be created on top of it and mask it.
580func (s *Service) reconcileInterruptedDeploy(ctx context.Context, app string) error {
581 op, ok, err := s.deps.State.LoadLatestOperation(ctx, app)
582 if err != nil {
583 return fmt.Errorf("deployment: load interrupted operation for %q: %w", app, err)
584 }
585 if !ok {
586 return nil
587 }
588 if !op.Terminal() && s.operationLive(app, op.Op) {
589 // The operation is owned by a live goroutine in this process that
590 // may still be persisting its outcome. Reconcile it only after that
591 // owner has exited; until then the store record is authoritative and
592 // must not be finalized here.
593 return nil
594 }
595 terminal := op.Terminal()
596 needsActive := false
597 for _, step := range op.Steps {
598 if reconcileNeedsContainer(step, terminal) {
599 needsActive = true
600 break
601 }
602 }
603 var active domain.AppActive
604 if needsActive {
605 active, _, err = s.deps.State.LoadActive(ctx, app)
606 if err != nil {
607 return fmt.Errorf("deployment: load active for interrupted operation: %w", err)
608 }
609 }
610 log := zerowrap.FromCtx(ctx)
611 // write is true when the journal itself changes: an interrupted operation
612 // is always finalized, a terminal one only when a leftover must be
613 // reported.
614 changed, err := s.convergeInterruptedSteps(ctx, app, &op, active, terminal, log)
615 if err != nil {
616 return err
617 }
618 write := !terminal || changed
619 if !write {
620 return nil
621 }
622 if !terminal {
623 // The effects ran and only the outcome write was interrupted: that is
624 // a success, not a failure.
625 op.Outcome = interruptedOutcome(op)
626 }
627 if err := s.deps.State.SaveOperation(ctx, op); err != nil {
628 return fmt.Errorf("deployment: finalize interrupted operation for %q: %w", app, err)
629 }
630 log.Warn().Str("app", app).Str("op", op.Op).Msg("deployment: reconciled an unfinished operation")
631 return nil
632}
633
634// reconcileNeedsContainer reports whether one step still needs container
635// work: a step that recorded a candidate. A terminal operation only retries
636// the step that recorded a failed candidate; a pending step exists only in an
637// interrupted operation.
638func reconcileNeedsContainer(step domain.AppOperationStep, terminal bool) bool {
639 if step.After == "" {
640 return false
641 }
642 if terminal {
643 return step.State == domain.AppStepFailed
644 }
645 return true
646}
647
648// convergeInterruptedSteps retries the recorded candidate of every step that
649// needs container work and accumulates the bounded leftovers in the journal.
650// It reports whether the journal changed and fails closed on the first
651// candidate that cannot be removed, before any later step or mutation runs.
652func (s *Service) convergeInterruptedSteps(ctx context.Context, app string, op *domain.AppOperation, active domain.AppActive, terminal bool, log zerowrap.Logger) (bool, error) {
653 changed := false
654 for i := range op.Steps {
655 step := &op.Steps[i]
656 if !reconcileNeedsContainer(*step, terminal) {
657 continue
658 }
659 warnings, convErr := s.convergeCandidate(ctx, app, active, step, terminal, log)
660 if step.After == "" {
661 // Converging the candidate clears its recorded After, so the
662 // journal changed even when nothing was left over to report: the
663 // converged step must be persisted, or every later pass reconciles
664 // it again.
665 changed = true
666 }
667 if len(warnings) > 0 {
668 // A candidate that will not go away is operator-visible instead of
669 // staying untracked.
670 op.Warnings = append(op.Warnings, journalWarnings(warnings)...)
671 changed = true
672 }
673 if convErr != nil {
674 // Fail closed: the orphan may still be running, so no later
675 // mutation may proceed. The journal keeps the leftover (and the
676 // failed step) for the next pass; a terminal operation stays the
677 // latest record because no newer claim is opened.
678 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
679 log.Warn().Err(saveErr).Msg("deployment: failed to record reconciliation failure")
680 }
681 return changed, fmt.Errorf("deployment: reconcile incomplete operation for %q: %w", app, convErr)
682 }
683 }
684 return changed, nil
685}
686
687// convergeCandidate converges the container one step recorded. The published
688// generation is kept; a never-published candidate is removed so recovery can
689// rebuild the generation recorded in ACTIVE without two generations running
690// at once. An interrupted operation's step is marked failed.
691//
692// A candidate that cannot be removed is returned as an error, not a
693// recoverable warning: the orphan may still run, so every later mutation must
694// fail closed instead of proceeding. The returned warnings are the leftovers
695// of that removal.
696//
697// Success clears the step's recorded After and keeps the removed (or kept)
698// container id in Detail, so the audit trail survives while the converged step
699// is no longer retried on every later pass. Only the failure paths leave After
700// in place, keeping a failed convergence eligible for retry.
701//
702// Once the candidate is converged, the inhibition the replacement wrote for
703// the superseded container is cleared. A step carrying both Before and After
704// records that the superseded container was confirmed gone before the
705// candidate was created (deployService retires it first), so the marker
706// protects nothing and would otherwise refuse the boot/start/restart rebuild
707// from ACTIVE.
708func (s *Service) convergeCandidate(ctx context.Context, app string, active domain.AppActive, step *domain.AppOperationStep, terminal bool, log zerowrap.Logger) ([]CleanupWarning, error) {
709 candidate := step.After
710 if publishedContainer(active, candidate) {
711 // The published generation: keep it. A step that already succeeded is
712 // left as recorded; a routing apply that never ran is republished by
713 // the periodic recovery pass.
714 if !terminal && step.State != domain.AppStepSucceeded {
715 step.State = domain.AppStepFailed
716 step.Error = "interrupted before traffic publication; the published container is kept and routing is republished"
717 }
718 step.Detail = "converged: published container " + candidate + " kept"
719 } else {
720 retired := s.retireContainer(ctx, app, retireOptions{Service: step.Service, Force: true}, candidate)
721 if !retired.Gone {
722 if !terminal {
723 step.State = domain.AppStepFailed
724 step.Error = "interrupted before publication; the unpublished candidate could not be removed: " + cleanupDetail(retired)
725 log.Warn().Str("app", app).Str("container", candidate).Msg("deployment: unpublished candidate could not be removed")
726 }
727 return retired.Warnings, fmt.Errorf("deployment: candidate %s of service %q could not be removed: %s", candidate, step.Service, cleanupDetail(retired))
728 }
729 if !terminal {
730 step.State = domain.AppStepFailed
731 step.Error = "interrupted before publication; the unpublished candidate was removed"
732 }
733 step.Detail = "converged: removed unpublished candidate " + candidate
734 }
735 // The candidate is gone (or published): the replacement-pending
736 // inhibition of the superseded container is stale. Dropping it lets
737 // boot/start/restart rebuild that generation from ACTIVE; the single
738 // writer is preserved because the superseded container was proven gone
739 // before the candidate existed.
740 if step.Service != "" && step.Before != "" && step.Before != candidate {
741 if err := s.clearRecoveryInhibition(ctx, app, step.Service, step.Before); err != nil {
742 log.Warn().Err(err).Str("app", app).Str("container", step.Before).Msg("deployment: clear stale replacement inhibition")
743 return []CleanupWarning{{Service: step.Service, Leftover: step.Before, Detail: "clear recovery inhibition: " + err.Error()}}, nil
744 }
745 }
746 // Only reconciliation removes a recorded candidate: clearing After here is
747 // what marks the step converged.
748 step.After = ""
749 return nil, nil
750}
751
752// interruptedOutcome classifies an interrupted operation: a failure, unless
753// every step had already succeeded.
754func interruptedOutcome(op domain.AppOperation) string {
755 if operationSucceeded(op) {
756 return domain.AppOutcomeSuccess
757 }
758 return domain.AppOutcomeFailed
759}
760
761// operationSucceeded reports whether every step of an interrupted operation
762// reached success: its effects ran and only the final outcome write was
763// interrupted.
764func operationSucceeded(op domain.AppOperation) bool {
765 if len(op.Steps) == 0 {
766 return false
767 }
768 for _, step := range op.Steps {
769 if step.State != domain.AppStepSucceeded {
770 return false
771 }
772 }
773 return true
774}
775
776// publishedContainer reports whether any service of the ACTIVE record names
777// the container. A container that is the published generation is never a
778// leftover, whichever service recorded it.
779func publishedContainer(active domain.AppActive, containerID string) bool {
780 if containerID == "" {
781 return false
782 }
783 for _, service := range active.Services {
784 if service.Container == containerID {
785 return true
786 }
787 }
788 return false
789}
790
791// deployService replaces one service sequentially and returns its terminal
792// result. before is the ACTIVE container being superseded (empty on a first
793// deployment); beforeGrace is its effective stop grace.
794//
795// The order is fixed: withdraw from traffic and confirm it, durably inhibit
796// a single-writer generation, retire the superseded container and confirm
797// it is gone, then create and start the replacement and wait for its
798// readiness probe. Nothing is created before the previous generation is
799// confirmed gone, so two generations never overlap; a withdrawal or
800// retirement that cannot be confirmed aborts without creating a candidate.
801func (s *Service) deployService(ctx context.Context, app, revision string, p pinnedService, opID, before string, beforeGrace time.Duration, journal candidateJournal) ServiceResult {
802 // Revalidate the current authorization and source before withdrawing or
803 // retiring the serving generation. A reload between preflight and this
804 // service step must fail without mutating the existing workload.
805 if _, err := s.resolveServiceBinds(app, p.spec); err != nil {
806 return s.failResult(revision, before, "", err)
807 }
808 if _, err := s.resolveServiceDevices(app, p.spec); err != nil {
809 return s.failResult(revision, before, "", err)
810 }
811 // Withdraw the service from traffic first and confirm it: a withdrawal
812 // that cannot be applied must block every container mutation, so no
813 // unverified generation stays routable. A first deployment has no
814 // published generation to withdraw.
815 if before != "" {
816 if err := s.withdrawForRecovery(ctx, app, p.name); err != nil {
817 return s.failResult(revision, before, "", err)
818 }
819 }
820 // A volume- or bind-owning replacement may write while the old
821 // generation still exists and could be revived by native restart policy.
822 // Inhibit that generation durably BEFORE the write can happen, so boot or
823 // periodic recovery can never restart the old writer on top of the new
824 // one. Cleared once the safe generation is published (publishService) or
825 // the operator removes the app.
826 if before != "" && singleWriterRequired(p.spec) {
827 if err := s.inhibitRecovery(ctx, app, p.name, before, domain.AppInhibitReplacementPending, opID); err != nil {
828 return s.failResult(revision, before, "", err)
829 }
830 }
831 if before != "" {
832 // The superseded generation must be confirmed gone before a new
833 // generation starts: a failed retirement aborts the replacement
834 // instead of allowing overlapping generations, and the generation
835 // keeps its inhibition and its claims.
836 retired := s.retireContainer(ctx, app, retireOptions{
837 Service: p.name, Grace: beforeGrace,
838 }, before)
839 if !retired.Gone {
840 return s.failResult(revision, before, "", fmt.Errorf("deployment: retire superseded container %s: %s", before, cleanupDetail(retired)))
841 }
842 }
843 created, binds, udpBinds, err := s.createAndStart(ctx, app, revision, p, opID, journal)
844 if err != nil {
845 result := s.failResult(revision, before, "", err)
846 if created != nil {
847 // The candidate exists: keep its ID in the terminal result so the
848 // journal and the operator can still trace it.
849 result.After = created.ID
850 }
851 return result
852 }
853 if err := s.waitServiceReady(ctx, app, created.ID, deploymentReadiness(p.spec), binds); err != nil {
854 // Capture redacted diagnostics while the failed replacement still
855 // exists, then remove only that candidate. The superseded container
856 // is already gone and is never recreated. Cleanup is best effort: a
857 // canceled context can leave the candidate behind, which is then
858 // reported as a leftover warning instead of a false success.
859 tail := s.redactDiagnostics(ctx, app, p, s.logTail(ctx, created.ID))
860 cleanup := s.retireCandidate(ctx, app, p.name, created.ID)
861 return ServiceResult{
862 Result: "failed",
863 EffectiveRevision: revision,
864 Before: before,
865 After: created.ID,
866 RestartUnsafe: singleWriterRequired(p.spec),
867 Error: err.Error(),
868 Diagnostics: tail,
869 CleanupWarnings: cleanup,
870 }
871 }
872 return ServiceResult{
873 Result: "deployed",
874 EffectiveRevision: revision,
875 Before: before,
876 After: created.ID,
877 RestartUnsafe: singleWriterRequired(p.spec),
878 BackendBinds: binds,
879 UDPBackendBinds: udpBinds,
880 }
881}
882
883// singleWriterRequired reports services that must never have two generations
884// running concurrently. Persistent volumes and any bind (especially a
885// writable one) may be written by both, so they are treated identically.
886func singleWriterRequired(spec domain.AppService) bool {
887 return len(spec.Volumes) > 0 || len(spec.Binds) > 0
888}
889
890// refreshTraffic rebuilds the proxy host index and applies the full
891// HTTP/L4 graph after activation. A failure is fatal to the caller:
892// ACTIVE is published but the new service is not routable, so the
893// operation must not report success. Any service of this app whose
894// fail-closed withdrawal was never applied is withdrawn again first, so
895// a later publication never runs on top of unproven forwarding.
896func (s *Service) refreshTraffic(ctx context.Context, app string) error {
897 if s.deps.Traffic == nil {
898 return nil
899 }
900 for _, service := range s.publication.pending(app) {
901 if err := s.withdrawForRecovery(ctx, app, service); err != nil {
902 return err
903 }
904 }
905 if err := s.deps.Traffic.RebuildTraffic(ctx); err != nil {
906 return fmt.Errorf("deployment: rebuild traffic for %q: %w", app, err)
907 }
908 return nil
909}
910
911// reserveVolumeOwnership durably records one app-owned volume as
912// attached BEFORE the runtime volume is created, so a crash between the
913// two leaves a protected record instead of an unowned volume. The
914// caller passes the validated incarnation ID from ensureIncarnationID,
915// which already assigned it, so no reload is needed for the labels.
916func (s *Service) reserveVolumeOwnership(ctx context.Context, app, appID, service, volumeName, runtimeName string) error {
917 ownership, err := s.deps.State.LoadOwnership(ctx, app)
918 if err != nil {
919 return fmt.Errorf("deployment: load ownership: %w", err)
920 }
921 if ownership.App == "" {
922 ownership.App = app
923 }
924 if ownership.ID == "" {
925 ownership.ID = appID
926 }
927 entry := domain.AppOwnedVolume{
928 Name: volumeName,
929 Service: service,
930 RuntimeName: runtimeName,
931 State: domain.AppResourceAttached,
932 }
933 replaced := false
934 for i, existing := range ownership.Volumes {
935 if existing.RuntimeName == runtimeName {
936 ownership.Volumes[i] = entry
937 replaced = true
938 break
939 }
940 }
941 if !replaced {
942 ownership.Volumes = append(ownership.Volumes, entry)
943 }
944 if err := s.deps.State.SaveOwnership(ctx, ownership); err != nil {
945 return fmt.Errorf("deployment: reserve volume ownership: %w", err)
946 }
947 return nil
948}
949
950// volumeProvenanceLabels is the label set stamped on an app-created
951// volume. Ownership is never inferred from the volume name, so the
952// labels must carry the full provenance: app, incarnation UUID, and
953// service. The managed marker is stamped here as well as merged by the
954// adapter, so provenance never depends on adapter behaviour alone.
955func volumeProvenanceLabels(app, appID, service, revision string) map[string]string {
956 labels := map[string]string{
957 domain.LabelManaged: "true",
958 domain.LabelApp: app,
959 domain.LabelAppService: service,
960 domain.LabelAppRevision: revision,
961 }
962 if appID != "" {
963 labels[domain.LabelAppID] = appID
964 }
965 return labels
966}
967
968// createAndStart builds the runtime config and starts the replacement.
969// Preflight already pulled and inspected the pinned runtime image.
970// Every TCP-capable interface port is published on 127.0.0.1 ephemeral
971// plus every UDP interface port on 127.0.0.1/udp ephemeral: readiness
972// and the proxy dial these loopback binds rootless-first,
973// never container IPs. The runtime keeps native restarts; Gordon
974// reconciles intent at boot and in the monitor (accepted decision:
975// native runtime restarts plus daemon reconciliation).
976func (s *Service) createAndStart(ctx context.Context, app, revision string, p pinnedService, opID string, journal candidateJournal) (*domain.Container, map[int]int, map[int]int, error) {
977 // Re-resolve binds from the current policy immediately before any
978 // runtime mutation: a bind revoked since preflight must fail here,
979 // before volume ownership or container creation.
980 resolvedBinds, err := s.resolveServiceBinds(app, p.spec)
981 if err != nil {
982 return nil, nil, nil, err
983 }
984 // Re-resolve devices from the current policy under the same
985 // fail-closed rule: revoked grants never reach the runtime.
986 resolvedDevices, err := s.resolveServiceDevices(app, p.spec)
987 if err != nil {
988 return nil, nil, nil, err
989 }
990 env, err := s.serviceEnv(ctx, app, p)
991 if err != nil {
992 return nil, nil, nil, err
993 }
994 image := runtimeImageRef(p)
995 // The incarnation network is verified or created before any
996 // container exists, so a workload is only ever started on a
997 // Gordon-owned network that no other app shares.
998 appID, err := s.ensureIncarnationID(ctx, app)
999 if err != nil {
1000 return nil, nil, nil, err
1001 }
1002 nets := s.resolveAppNetworks(appID, p.sharedNetworks)
1003 if err := s.ensureAppNetworks(ctx, app, appID, nets); err != nil {
1004 return nil, nil, nil, err
1005 }
1006 volumes := map[string]string{}
1007 readOnlyVolumes := map[string]string{}
1008 for _, vol := range p.spec.Volumes {
1009 runtimeName := domain.RuntimeVolumeName(app, p.spec.Name, vol.Name)
1010 // Durable ownership is reserved BEFORE the runtime volume
1011 // exists: a crash in between leaves a protected record, never
1012 // an unowned volume that prune could later adopt. The
1013 // incarnation ID is already validated above, so the labels
1014 // carry the same ID without another ownership load.
1015 if err := s.reserveVolumeOwnership(ctx, app, appID, p.spec.Name, vol.Name, runtimeName); err != nil {
1016 return nil, nil, nil, err
1017 }
1018 if err := s.deps.Runtime.CreateVolume(ctx, runtimeName, volumeProvenanceLabels(app, appID, p.spec.Name, revision)); err != nil {
1019 // CreateVolume is idempotent at the adapter; existence was
1020 // checked at preflight, so only real backend errors fail here.
1021 return nil, nil, nil, fmt.Errorf("deployment: create volume %q: %w", vol.Name, err)
1022 }
1023 // A declared read-only mount must reach the runtime as read-only:
1024 // a writable mount would let the service modify protected data.
1025 if vol.ReadOnly {
1026 readOnlyVolumes[vol.Path] = runtimeName
1027 continue
1028 }
1029 volumes[vol.Path] = runtimeName
1030 }
1031 config := &domain.ContainerConfig{
1032 Image: image,
1033 Name: domain.LogicalServiceIdentity(app, p.spec.Name) + "--" + shortOp(opID),
1034 Env: env,
1035 Entrypoint: append([]string(nil), p.spec.Command...),
1036 Volumes: volumes,
1037 ReadOnlyVolumes: readOnlyVolumes,
1038 Binds: resolvedBinds,
1039 CDIDevices: resolvedDevices,
1040 Labels: appLabels(app, p.spec.Name, revision),
1041 AutoRemove: false,
1042 RestartPolicy: domain.RestartPolicyAlways,
1043 PortPublishes: backendPublishes(p.spec),
1044 NetworkMode: nets.private,
1045 Hostname: p.spec.Name,
1046 Aliases: []string{p.spec.Name},
1047 MemoryLimit: s.deps.Limits.MemoryBytes,
1048 NanoCPUs: s.deps.Limits.NanoCPUs,
1049 PidsLimit: s.deps.Limits.PidsLimit,
1050 }
1051 created, err := s.createContainer(ctx, p.name, config)
1052 if err != nil {
1053 return nil, nil, nil, err
1054 }
1055 // The candidate ID is durably journaled before the container is started
1056 // or joins a shared network, so an interrupted deployment can be cleaned
1057 // up instead of leaving an untracked running generation.
1058 if err := s.recordCandidate(ctx, app, p.spec.Name, created.ID, journal); err != nil {
1059 return nil, nil, nil, err
1060 }
1061 if err := s.connectSharedNetworks(ctx, created.ID, nets); err != nil {
1062 s.retireCandidate(ctx, app, p.spec.Name, created.ID)
1063 return created, nil, nil, err
1064 }
1065 if err := s.deps.Runtime.StartContainer(ctx, created.ID); err != nil {
1066 s.retireCandidate(ctx, app, p.spec.Name, created.ID)
1067 return created, nil, nil, fmt.Errorf("deployment: start container: %w", err)
1068 }
1069 binds, udpBinds, err := s.readBackendBinds(ctx, app, p.spec.Name, created.ID, backendPorts(p.spec))
1070 if err != nil {
1071 s.retireCandidate(ctx, app, p.spec.Name, created.ID)
1072 return created, nil, nil, err
1073 }
1074 return created, binds, udpBinds, nil
1075}
1076
1077// runtimeImageRef resolves the exact runtime image reference of one pinned
1078// service: the image preflight resolved, else the manifest reference pinned to
1079// its digest.
1080func runtimeImageRef(p pinnedService) string {
1081 if p.runtimeImage != "" {
1082 return p.runtimeImage
1083 }
1084 if p.digest == "" {
1085 return p.spec.Image
1086 }
1087 return stripImageTag(p.spec.Image) + "@" + p.digest
1088}
1089
1090// recordCandidate durably journals a freshly created candidate before it is
1091// started. A candidate whose ID cannot be recorded must not keep running:
1092// recovery could never find it, so it is removed and the failure surfaced.
1093func (s *Service) recordCandidate(ctx context.Context, app, service, containerID string, journal candidateJournal) error {
1094 if journal == nil {
1095 return nil
1096 }
1097 if err := journal(ctx, containerID); err != nil {
1098 retired := s.retireContainer(ctx, app, retireOptions{Service: service, Force: true}, containerID)
1099 if !retired.Gone {
1100 return fmt.Errorf("%w (candidate cleanup: %s)", err, cleanupDetail(retired))
1101 }
1102 return err
1103 }
1104 return nil
1105}
1106
1107func (s *Service) createContainer(ctx context.Context, service string, config *domain.ContainerConfig) (*domain.Container, error) {
1108 created, err := s.deps.Runtime.CreateContainer(ctx, config)
1109 if err == nil {
1110 return created, nil
1111 }
1112 // The engine-unsupported sentinel survives redaction so callers can
1113 // map it to the structured runtime-unsupported envelope. It carries
1114 // no host inventory (engine family/version only), and the device
1115 // gate runs only for device-bearing creates, so this branch also
1116 // covers services that declare both binds and devices.
1117 if errors.Is(err, domain.ErrRuntimeUnsupported) {
1118 return nil, fmt.Errorf("deployment: create container for service %q with administrative devices: %w", service, domain.ErrRuntimeUnsupported)
1119 }
1120 if len(config.Binds) > 0 {
1121 // A CreateContainer failure is a runtime error, not a bind policy
1122 // violation: policy was already enforced by resolveServiceBinds
1123 // before this call. Runtime errors may embed the resolved host
1124 // path, so redact the whole cause instead of mislabelling it and
1125 // keep it out of operation journals and API/CLI responses.
1126 return nil, fmt.Errorf("deployment: create container for service %q with administrative mounts: runtime error redacted", service)
1127 }
1128 if len(config.CDIDevices) > 0 {
1129 // Same redaction rule for device-bearing creates: runtime errors
1130 // may embed resolved CDI IDs, which are host inventory. Policy
1131 // was already enforced by resolveServiceDevices before this call.
1132 return nil, fmt.Errorf("deployment: create container for service %q with administrative devices: runtime error redacted", service)
1133 }
1134 return nil, fmt.Errorf("deployment: create container: %w", err)
1135}
1136
1137// backendPorts collects every interface container port for loopback
1138// publication: public HTTP + TCP interfaces on tcp plus UDP interfaces on
1139// udp, plus an explicit TCP readiness port. Internal HTTP ports are
1140// excluded: they are reached only over the private network and must keep
1141// no host binding. Each publish is on 127.0.0.1 ephemeral (never public).
1142// Deduplicated by (protocol, container port).
1143func backendPorts(spec domain.AppService) []domain.ContainerBackendPort {
1144 seen := map[domain.ContainerBackendPort]struct{}{}
1145 var ports []domain.ContainerBackendPort
1146 add := func(port int, protocol domain.NetworkProtocol) {
1147 if port <= 0 {
1148 return
1149 }
1150 key := domain.ContainerBackendPort{ContainerPort: port, Protocol: protocol}
1151 if _, ok := seen[key]; !ok {
1152 seen[key] = struct{}{}
1153 ports = append(ports, key)
1154 }
1155 }
1156 for _, h := range spec.HTTP {
1157 if !h.IsPublic() {
1158 continue
1159 }
1160 add(h.Port, domain.NetworkProtocolTCP)
1161 }
1162 for _, t := range spec.TCP {
1163 add(t.Port, domain.NetworkProtocolTCP)
1164 }
1165 for _, u := range spec.UDP {
1166 add(u.Port, domain.NetworkProtocolUDP)
1167 }
1168 // Readiness probes are TCP-only (validation rejects UDP-only
1169 // services with tcp/http readiness); the readiness port matches a
1170 // declared TCP container port. An internal-only port is never
1171 // published, so readiness metadata cannot create a host binding for
1172 // it.
1173 if spec.Readiness.Port > 0 && !spec.InternallyOnlyPort(spec.Readiness.Port) {
1174 add(spec.Readiness.Port, domain.NetworkProtocolTCP)
1175 }
1176 sort.Slice(ports, func(i, j int) bool {
1177 if ports[i].Protocol != ports[j].Protocol {
1178 return ports[i].Protocol < ports[j].Protocol
1179 }
1180 return ports[i].ContainerPort < ports[j].ContainerPort
1181 })
1182 return ports
1183}
1184
1185// backendPublishes maps backend ports to 127.0.0.1 ephemeral publishes.
1186func backendPublishes(spec domain.AppService) []domain.ContainerPortPublish {
1187 ports := backendPorts(spec)
1188 publishes := make([]domain.ContainerPortPublish, 0, len(ports))
1189 for _, port := range ports {
1190 publishes = append(publishes, domain.ContainerPortPublish{
1191 HostIP: "127.0.0.1",
1192 HostPort: 0,
1193 ContainerPort: port.ContainerPort,
1194 Protocol: port.Protocol,
1195 })
1196 }
1197 return publishes
1198}
1199
1200// readBackendBinds resolves each published container port to its
1201// 127.0.0.1 host port and registers the Gordon-generated claims
1202// (owner gordon-backend) in the global checkpoint. Missing binds fail:
1203// an unbound backend can serve neither readiness nor proxy traffic.
1204// A registration conflict removes the CANDIDATE container and fails —
1205// callers must only pass freshly created candidates here, never an
1206// existing ACTIVE container (see inspectBackendBinds).
1207func (s *Service) readBackendBinds(ctx context.Context, app, service, containerID string, ports []domain.ContainerBackendPort) (map[int]int, map[int]int, error) {
1208 if len(ports) == 0 {
1209 return map[int]int{}, map[int]int{}, nil
1210 }
1211 observed, err := s.deps.Runtime.GetContainerBackendBinds(ctx, containerID, ports)
1212 if err != nil {
1213 return nil, nil, fmt.Errorf("deployment: backend binds: %w", err)
1214 }
1215 binds, udpBinds := splitBackendBinds(observed)
1216 claims := backendClaims(app, service, containerID, observed)
1217 if len(claims) == 0 {
1218 return binds, udpBinds, nil
1219 }
1220 if err := s.deps.State.RegisterBackendBinds(ctx, claims); err != nil {
1221 s.retireCandidate(ctx, app, service, containerID)
1222 return nil, nil, err
1223 }
1224 return binds, udpBinds, nil
1225}
1226
1227// inspectBackendBinds re-inspects the loopback publishes of an EXISTING
1228// container and registers the Gordon-generated claims. Unlike
1229// readBackendBinds it never stops or removes the container: a
1230// registration conflict fails without touching the ACTIVE workload.
1231// Missing binds fail: an unbound backend must not stay routable.
1232func (s *Service) inspectBackendBinds(ctx context.Context, app, service, containerID string, ports []domain.ContainerBackendPort) (map[int]int, map[int]int, error) {
1233 if len(ports) == 0 {
1234 return map[int]int{}, map[int]int{}, nil
1235 }
1236 observed, err := s.deps.Runtime.GetContainerBackendBinds(ctx, containerID, ports)
1237 if err != nil {
1238 return nil, nil, fmt.Errorf("deployment: backend binds: %w", err)
1239 }
1240 binds, udpBinds := splitBackendBinds(observed)
1241 claims := backendClaims(app, service, containerID, observed)
1242 if len(claims) == 0 {
1243 return binds, udpBinds, nil
1244 }
1245 if err := s.deps.State.RegisterBackendBinds(ctx, claims); err != nil {
1246 return nil, nil, err
1247 }
1248 return binds, udpBinds, nil
1249}
1250
1251// splitBackendBinds partitions observed binds by protocol. TCP feeds
1252// readiness and HTTP/TCP proxying; UDP feeds UDP relaying.
1253func splitBackendBinds(observed []domain.ContainerBackendBind) (map[int]int, map[int]int) {
1254 binds := map[int]int{}
1255 udpBinds := map[int]int{}
1256 for _, bind := range observed {
1257 if bind.Protocol == domain.NetworkProtocolUDP {
1258 udpBinds[bind.ContainerPort] = bind.HostPort
1259 } else {
1260 binds[bind.ContainerPort] = bind.HostPort
1261 }
1262 }
1263 return binds, udpBinds
1264}
1265
1266// backendClaims maps observed binds to Gordon-generated loopback claims
1267// with the exact protocol (tcp/udp) for the global checkpoint.
1268func backendClaims(app, service, containerID string, observed []domain.ContainerBackendBind) []domain.AppListenerReservation {
1269 claims := make([]domain.AppListenerReservation, 0, len(observed))
1270 for _, bind := range observed {
1271 claims = append(claims, domain.AppListenerReservation{
1272 Proto: string(bind.Protocol),
1273 IP: "127.0.0.1",
1274 Port: bind.HostPort,
1275 Service: service,
1276 App: app,
1277 Owner: domain.OwnerGordonBackend,
1278 ContainerID: containerID,
1279 })
1280 }
1281 return claims
1282}
1283
1284// stripImageTag removes any tag or digest suffix from an image reference.
1285// colon is a registry port (localhost:15500/e2e/web:v1 strips to
1286// localhost:15500/e2e/web).
1287func stripImageTag(ref string) string {
1288 if base, _, ok := strings.Cut(ref, "@"); ok {
1289 return base
1290 }
1291 lastSlash := strings.LastIndex(ref, "/")
1292 if idx := strings.LastIndex(ref, ":"); idx > lastSlash {
1293 return ref[:idx]
1294 }
1295 return ref
1296}
1297
1298// pullImage fetches the pinned image from the installation registry.
1299// External refs pull anonymously; installation-registry refs use the
1300// internal credentials. A missing registry config disables pulling
1301// (tests and pre-pulled environments).
1302func (s *Service) pullImage(ctx context.Context, image string) (string, error) {
1303 if err := s.validateImageSource(image, ""); err != nil {
1304 return "", err
1305 }
1306 if s.deps.Registry.Domain == "" {
1307 return image, nil
1308 }
1309 if !s.deps.ImagePolicy.IsInstallationImage(image) {
1310 if err := s.deps.Runtime.PullImage(ctx, image); err != nil {
1311 return "", fmt.Errorf("deployment: pull image %q: %w", image, err)
1312 }
1313 return image, nil
1314 }
1315 _, remainder, ok := strings.Cut(image, "/")
1316 if !ok || !strings.Contains(remainder, "@sha256:") {
1317 return "", fmt.Errorf("deployment: installation image must include an exact sha256 digest: %w", domain.ErrAppImageNotAllowed)
1318 }
1319 pullImage := s.deps.Registry.PullAddress + "/" + remainder
1320 request := domain.ImagePullRequest{Reference: pullImage, Username: s.deps.Registry.Username, Password: s.deps.Registry.Password, Transport: domain.ImagePullTransportHTTP}
1321 if err := s.deps.Runtime.PullImageWithOptions(ctx, request); err != nil {
1322 return "", fmt.Errorf("deployment: pull installation image %q via configured local HTTP transport %q: %w", image, s.deps.Registry.PullAddress, err)
1323 }
1324 digest := remainder[strings.LastIndex(remainder, "@")+1:]
1325 if err := s.deps.Runtime.VerifyImageDigest(ctx, pullImage, digest); err != nil {
1326 return "", fmt.Errorf("deployment: verify pulled installation image %q: %w", image, err)
1327 }
1328 return pullImage, nil
1329}
1330
1331// serviceEnv resolves the service environment: the captured revision's
1332// app-wide public env on every service, plus that service's own secret
1333// values (memory only, UUID-keyed paths). Keys are emitted sorted and
1334// deterministic. Overlapping public/secret keys are invalid with no
1335// precedence; preflight rejects them before any mutation.
1336func (s *Service) serviceEnv(ctx context.Context, app string, p pinnedService) ([]string, error) {
1337 spec := p.spec
1338 envMap := make(map[string]string, len(p.appEnv)+len(spec.Secrets))
1339 for key, value := range p.appEnv {
1340 envMap[key] = value
1341 }
1342 keys := make([]string, 0, len(spec.Secrets))
1343 for envKey := range spec.Secrets {
1344 keys = append(keys, envKey)
1345 }
1346 sort.Strings(keys)
1347 if len(keys) == 0 {
1348 return sortedEnv(envMap), nil
1349 }
1350 id, err := s.appSecretID(ctx, app)
1351 if err != nil {
1352 return nil, err
1353 }
1354 for _, envKey := range keys {
1355 if _, overlap := envMap[envKey]; overlap {
1356 return nil, fmt.Errorf("deployment: env key %q collides with a secret key in service %q: %w", envKey, spec.Name, domain.ErrInvalidAppSpec)
1357 }
1358 path := domain.AppSecretPathForID(id, app, spec.Name, spec.Secrets[envKey])
1359 value, err := s.deps.Secrets.GetSecret(ctx, path)
1360 if err != nil {
1361 return nil, fmt.Errorf("deployment: secret %s (env %s) missing: %w", path, envKey, domain.ErrAppSecretMissing)
1362 }
1363 envMap[envKey] = value
1364 }
1365 return sortedEnv(envMap), nil
1366}
1367
1368// sortedEnv renders a KEY=value list in sorted key order. Values stay in
1369// memory for container creation only; nothing is persisted or logged.
1370func sortedEnv(envMap map[string]string) []string {
1371 if len(envMap) == 0 {
1372 return nil
1373 }
1374 keys := make([]string, 0, len(envMap))
1375 for key := range envMap {
1376 keys = append(keys, key)
1377 }
1378 sort.Strings(keys)
1379 env := make([]string, 0, len(keys))
1380 for _, key := range keys {
1381 env = append(env, key+"="+envMap[key])
1382 }
1383 return env
1384}
1385
1386// publishService writes the per-service effective record once the
1387// replacement passed readiness.
1388func (s *Service) publishService(ctx context.Context, app, revision string, p pinnedService, result ServiceResult, opID string) error {
1389 if app == "" {
1390 return fmt.Errorf("deployment: publish active: empty app: %w", domain.ErrAppStateConflict)
1391 }
1392 active, ok, err := s.deps.State.LoadActive(ctx, app)
1393 if err != nil {
1394 return fmt.Errorf("deployment: load active: %w", err)
1395 }
1396 if !ok {
1397 active = domain.AppActive{App: app, Services: map[string]domain.AppEffectiveService{}}
1398 }
1399 if active.Services == nil {
1400 active.Services = map[string]domain.AppEffectiveService{}
1401 }
1402 active.Services[p.name] = domain.AppEffectiveService{
1403 EffectiveRevision: revision,
1404 ActivatedBy: opID,
1405 ActivatedAt: time.Now().UTC(),
1406 Image: p.spec.Image,
1407 Digest: p.digest,
1408 Container: result.After,
1409 Spec: p.spec,
1410 BackendBinds: result.BackendBinds,
1411 UDPBackendBinds: result.UDPBackendBinds,
1412 }
1413 active.ConvergedRevision = revision
1414 converged := true
1415 for _, svc := range active.Services {
1416 if svc.EffectiveRevision != revision {
1417 converged = false
1418 break
1419 }
1420 }
1421 active.Converged = converged
1422 if converged {
1423 active.Networks = append([]domain.AppSharedNetwork(nil), p.appNetworks...)
1424 }
1425 if err := s.deps.State.SaveActive(ctx, active); err != nil {
1426 return fmt.Errorf("deployment: publish active: %w", err)
1427 }
1428 if err := s.recordOwnership(ctx, app, p); err != nil {
1429 return err
1430 }
1431 // The superseded generation is now safely replaced: clearing its
1432 // inhibition is what makes this generation authoritative. A failed
1433 // publication leaves the marker in place.
1434 if result.Before != "" && result.Before != result.After {
1435 if err := s.clearRecoveryInhibition(ctx, app, p.name, result.Before); err != nil {
1436 return err
1437 }
1438 }
1439 return nil
1440}
1441
1442// recordOwnership stamps volume/secret/network ownership after publication.
1443func (s *Service) recordOwnership(ctx context.Context, app string, p pinnedService) error {
1444 ownership, err := s.deps.State.LoadOwnership(ctx, app)
1445 if err != nil {
1446 return err
1447 }
1448 if ownership.App == "" {
1449 ownership.App = app
1450 }
1451 // ownership.ID arrives from LoadOwnership (store-assigned UUID);
1452 // empty only for legacy records predating UUIDs.
1453 volumes := append([]domain.AppOwnedVolume(nil), ownership.Volumes...)
1454 seen := map[string]int{}
1455 for i, vol := range volumes {
1456 seen[vol.Name] = i
1457 }
1458 for _, vol := range p.spec.Volumes {
1459 entry := domain.AppOwnedVolume{
1460 Name: vol.Name,
1461 Service: p.spec.Name,
1462 RuntimeName: domain.RuntimeVolumeName(app, p.spec.Name, vol.Name),
1463 State: domain.AppResourceAttached,
1464 }
1465 if idx, ok := seen[vol.Name]; ok {
1466 volumes[idx] = entry
1467 } else {
1468 seen[vol.Name] = len(volumes)
1469 volumes = append(volumes, entry)
1470 }
1471 }
1472 ownership.Volumes = volumes
1473 secrets := append([]domain.AppOwnedSecret(nil), ownership.Secrets...)
1474 secretSeen := map[string]int{}
1475 for i, secret := range secrets {
1476 secretSeen[secret.Service+"\x00"+secret.Env] = i
1477 }
1478 for envKey, name := range p.spec.Secrets {
1479 entry := domain.AppOwnedSecret{
1480 Service: p.spec.Name,
1481 Env: envKey,
1482 Name: name,
1483 Path: domain.AppSecretPathForID(ownership.ID, app, p.spec.Name, name),
1484 State: domain.AppResourceAttached,
1485 }
1486 key := p.spec.Name + "\x00" + envKey
1487 if idx, ok := secretSeen[key]; ok {
1488 secrets[idx] = entry
1489 } else {
1490 secretSeen[key] = len(secrets)
1491 secrets = append(secrets, entry)
1492 }
1493 }
1494 ownership.Secrets = secrets
1495 ownership.Images = recordImageOwnership(ownership.Images, p)
1496 recordNetworkOwnership(&ownership, s.deps.Networks.Prefix, p.sharedNetworks)
1497 if ownership.Services == nil {
1498 ownership.Services = map[string]domain.AppServiceRecovery{}
1499 }
1500 ownership.Services[p.spec.Name] = domain.AppServiceRecovery{RestartUnsafe: singleWriterRequired(p.spec)}
1501 if err := s.deps.State.SaveOwnership(ctx, ownership); err != nil {
1502 return fmt.Errorf("deployment: record ownership: %w", err)
1503 }
1504 return nil
1505}
1506
1507// recordImageOwnership stamps one service's pinned image into the app's
1508// owned-image list. Positive durable ownership is what later authorizes
1509// runtime image prune; image labels alone never do.
1510func recordImageOwnership(existing []domain.AppOwnedImage, p pinnedService) []domain.AppOwnedImage {
1511 entry := domain.AppOwnedImage{
1512 Service: p.spec.Name,
1513 Reference: p.spec.Image,
1514 Digest: p.digest,
1515 State: domain.AppResourceAttached,
1516 }
1517 out := append([]domain.AppOwnedImage(nil), existing...)
1518 for i, image := range out {
1519 if image.Service == entry.Service {
1520 out[i] = entry
1521 return out
1522 }
1523 }
1524 return append(out, entry)
1525}
1526
1527// failResult builds a preflight/create failure before any container exists.
1528func (s *Service) failResult(revision, before, after string, err error) ServiceResult {
1529 return ServiceResult{
1530 Result: "failed",
1531 EffectiveRevision: revision,
1532 Before: before,
1533 After: after,
1534 Error: err.Error(),
1535 }
1536}
1537
1538// logTail returns the bounded recent log tail of the failing container.
1539// The read is bounded by the caller's deadline and by an internal cap so
1540// a slow or hung log stream cannot stall the operation.
1541func (s *Service) logTail(ctx context.Context, containerID string) []string {
1542 if containerID == "" || ctx.Err() != nil {
1543 return nil
1544 }
1545 tailCtx, cancel := context.WithTimeout(ctx, failLogTailTimeout)
1546 defer cancel()
1547 stream, err := s.deps.Runtime.GetContainerLogs(tailCtx, containerID, false)
1548 if err != nil {
1549 return nil
1550 }
1551 defer func() { _ = stream.Close() }()
1552 lines, err := tailLines(stream, failLogTailLines)
1553 if err != nil {
1554 return nil
1555 }
1556 return lines
1557}
1558
1559// redactDiagnostics replaces every value of this service's secrets in the
1560// given log lines. Secret values are read into memory only. When a secret
1561// cannot be read, diagnostics are dropped entirely: unredacted application
1562// output is never persisted.
1563func (s *Service) redactDiagnostics(ctx context.Context, app string, p pinnedService, lines []string) []string {
1564 if len(lines) == 0 {
1565 return nil
1566 }
1567 if len(p.spec.Secrets) == 0 || s.deps.Secrets == nil {
1568 return lines
1569 }
1570 id, err := s.appSecretID(ctx, app)
1571 if err != nil {
1572 return nil
1573 }
1574 redacted := append([]string(nil), lines...)
1575 for _, name := range p.spec.Secrets {
1576 path := domain.AppSecretPathForID(id, app, p.spec.Name, name)
1577 value, err := s.deps.Secrets.GetSecret(ctx, path)
1578 if err != nil || value == "" {
1579 return nil
1580 }
1581 for i, line := range redacted {
1582 redacted[i] = strings.ReplaceAll(line, value, "[redacted]")
1583 }
1584 }
1585 return redacted
1586}
1587
1588// tailLines keeps the last N lines; logs are untrusted app output.
1589func tailLines(r io.Reader, n int) ([]string, error) {
1590 scanner := bufio.NewScanner(r)
1591 scanner.Buffer(make([]byte, 64*1024), 1024*1024)
1592 var lines []string
1593 for scanner.Scan() {
1594 lines = append(lines, scanner.Text())
1595 if len(lines) > n {
1596 lines = lines[len(lines)-n:]
1597 }
1598 }
1599 if err := scanner.Err(); err != nil {
1600 return nil, err
1601 }
1602 return lines, nil
1603}
1604
1605// deploymentReadiness resolves the readiness check a deploy or restart
1606// runs. Readiness never depends on how a service is replaced: a declared
1607// http, tcp, or log check is used as declared, and an undeclared check
1608// ("none" or unset) keeps the implicit HTTP probe on a service Gordon can
1609// probe over the published loopback bind — at least one effective-public
1610// HTTP interface, no L4 interface, no volume, and no bind. An undeclared
1611// probe on a stateful or L4 workload stays unchecked instead: an HTTP GET on
1612// the published port proves nothing there, so Gordon never guesses one.
1613func deploymentReadiness(spec domain.AppService) domain.AppService {
1614 if !implicitHTTPProbe(spec) {
1615 return spec
1616 }
1617 spec.Readiness.Type = domain.AppReadinessHTTP
1618 return spec
1619}
1620
1621// implicitHTTPProbe reports whether a service that declares no readiness
1622// check receives the implicit HTTP probe when a deploy or restart starts it.
1623// Recovery verification of an already running generation keeps the declared
1624// readiness unchanged: it re-checks a generation it did not replace.
1625func implicitHTTPProbe(spec domain.AppService) bool {
1626 if spec.Readiness.Type != "" && spec.Readiness.Type != domain.AppReadinessNone {
1627 return false
1628 }
1629 if len(spec.HTTP) == 0 || len(spec.TCP) > 0 || len(spec.UDP) > 0 || len(spec.Volumes) > 0 || len(spec.Binds) > 0 {
1630 return false
1631 }
1632 for _, h := range spec.HTTP {
1633 if !h.IsPublic() {
1634 return false
1635 }
1636 }
1637 return true
1638}
1639
1640// appLabels stamps engine ownership on every created container.
1641func appLabels(app, service, revision string) map[string]string {
1642 return map[string]string{
1643 domain.LabelManaged: "true",
1644 domain.LabelApp: app,
1645 domain.LabelAppService: service,
1646 domain.LabelAppRevision: revision,
1647 }
1648}
1649
1650// shortOp keeps container names short but unique per op.
1651func shortOp(opID string) string {
1652 trimmed := strings.TrimPrefix(opID, "op-")
1653 if len(trimmed) > 12 {
1654 trimmed = trimmed[:12]
1655 }
1656 if trimmed == "" {
1657 return "op"
1658 }
1659 return trimmed
1660}