1// Package deployment implements the app deployment engine: preflight,
2// journaled execution, and terminal-result computation.
3package deployment
4
5import (
6 "context"
7 "encoding/hex"
8 "errors"
9 "fmt"
10 "maps"
11 "sort"
12 "strings"
13 "sync"
14 "time"
15
16 "github.com/bnema/zerowrap"
17 "github.com/google/uuid"
18
19 "github.com/bnema/gordon/internal/boundaries/out"
20 "github.com/bnema/gordon/internal/domain"
21 "github.com/bnema/gordon/pkg/validation"
22)
23
24// Deps are the driven adapters owned by the deployment engine.
25type Deps struct {
26 State out.AppState
27 Runtime out.ContainerRuntime
28 Images out.ImageResolver
29 Secrets out.SecretProvider
30 // Registry configures image pulls from the installation registry.
31 // Domain scopes which refs pull with auth; empty disables pulling.
32 Registry RegistryConfig
33 // Networks is the installation network policy applied to every app
34 // workload: the prefix used to derive Gordon-owned network names and
35 // whether those networks block external egress.
36 Networks NetworkConfig
37 // Limits are the installation container resource limits applied on
38 // every create and recovery path. Zero means unlimited.
39 Limits ResourceLimits
40 // ImagePolicy restricts which registries and references may be
41 // resolved, pulled, or run. It is enforced on every path, including
42 // boot, restart, and recovery.
43 ImagePolicy domain.ImageSourcePolicy
44 // Traffic is the serialized HTTP/L4 publish boundary. Rebuilt after
45 // every activation (deploy/start/restart/stop/remove) so the proxy
46 // host index follows ACTIVE; recovery also withdraws one service
47 // fail-closed before touching a workload. Nil in tests that assert
48 // state effects only. Failures are reported to the caller.
49 Traffic out.AppTrafficRefresher
50 // SecretFiles stores file-delivered secrets on the host. Optional:
51 // nil is valid only while no service delivers secrets as files;
52 // preflight refuses such services otherwise.
53 SecretFiles out.SecretFileStore
54 // Guard is checked on every app container config right before the
55 // runtime create. The zero value still refuses host/shared network
56 // namespaces and host paths passed as volume names.
57 Guard domain.RuntimeGuard
58}
59
60// RegistryConfig carries the installation registry identity for pulls.
61type RegistryConfig struct {
62 Domain string
63 PullAddress string
64 Username string
65 Password string
66}
67
68// NetworkConfig carries the installation network policy for app workloads.
69type NetworkConfig struct {
70 // Prefix names the Gordon-owned networks. Empty uses the domain default.
71 Prefix string
72 // Internal blocks external egress from containers attached to those
73 // networks (Docker's Internal flag).
74 Internal bool
75}
76
77// ResourceLimits are the installation container limits applied on every
78// create and recovery path. Zero values mean unlimited.
79type ResourceLimits struct {
80 MemoryBytes int64
81 NanoCPUs int64
82 PidsLimit int64
83}
84
85// Service orchestrates captured-revision deploys.
86type Service struct {
87 deps Deps
88 log zerowrap.Logger
89 probes *ProbeDeps
90
91 // coord serializes workload mutations per app: deploy/start/
92 // restart/stop/remove/boot lock blocking, periodic reconciliation
93 // skips busy apps.
94 coord *appCoordinator
95 // barrier is the process-wide GC barrier. A shared lease covers
96 // every workload mutation from resource selection through durable
97 // protection publication; prune holds the exclusive lease.
98 barrier out.GCBarrier
99 // backoff is the in-memory, generation-scoped recovery budget.
100 backoff *recoveryBackoff
101 // publication holds the in-memory publication inhibition: a
102 // service whose fail-closed withdrawal could not be applied must be
103 // withdrawn again before any later publication.
104 publication publicationInhibition
105 // executions remembers the last observed execution start per
106 // generation, so a native restart is detected without restarting
107 // the workload ourselves.
108 executions executionTracker
109 // bindMu guards bindPolicies. A config reload replaces the whole
110 // map atomically; deploy paths read the current map without holding
111 // the lock.
112 bindMu sync.RWMutex
113 bindPolicies map[string]domain.AppBindPolicy
114 // deviceMu guards devicePolicies. A config reload replaces the whole
115 // map atomically; deploy paths read the current map without holding
116 // the lock.
117 deviceMu sync.RWMutex
118 devicePolicies map[string]domain.AppDevicePolicy
119 // capabilityMu guards capabilityPolicies, replaced atomically on reload.
120 capabilityMu sync.RWMutex
121 capabilityPolicies map[string]domain.AppCapabilityPolicy
122 // liveMu guards liveOps. liveOps holds the operations this process
123 // owns between the StartDeploy claim and the end of ExecuteDeploy. It
124 // is in-memory only: a fresh Service (boot) starts empty and may
125 // reconcile every persisted non-terminal operation, while a
126 // foreground mutation must never finalize an operation a live
127 // goroutine still owns.
128 liveMu sync.Mutex
129 liveOps map[string]struct{}
130}
131
132// NewService creates the deployment engine. All deps are required;
133// Secrets may be nil only in tests that never reach secret preflight.
134func NewService(deps Deps, log zerowrap.Logger) *Service {
135 if deps.ImagePolicy.InstallationRegistry == "" {
136 deps.ImagePolicy.InstallationRegistry = deps.Registry.Domain
137 }
138 return &Service{
139 deps: deps,
140 log: log,
141 coord: newAppCoordinator(),
142 backoff: newRecoveryBackoff(time.Now),
143 }
144}
145
146// markOperationLive records that this process owns one operation from the
147// StartDeploy claim until ExecuteDeploy finishes. A live operation is the
148// single owner's in-flight work: a foreground reconciliation must not
149// finalize it, or the claim would be settled before the owner can execute it.
150func (s *Service) markOperationLive(app, op string) {
151 if app == "" || op == "" {
152 return
153 }
154 s.liveMu.Lock()
155 defer s.liveMu.Unlock()
156 if s.liveOps == nil {
157 s.liveOps = make(map[string]struct{})
158 }
159 s.liveOps[liveOperationKey(app, op)] = struct{}{}
160}
161
162// clearOperationLive drops the live marker. It is called once ExecuteDeploy
163// has recorded its outcome or left the claim recoverable, so a later
164// reconciliation can converge the operation instead of leaving it in flight.
165func (s *Service) clearOperationLive(app, op string) {
166 s.liveMu.Lock()
167 defer s.liveMu.Unlock()
168 delete(s.liveOps, liveOperationKey(app, op))
169}
170
171// operationLive reports whether this process still owns the operation.
172func (s *Service) operationLive(app, op string) bool {
173 s.liveMu.Lock()
174 defer s.liveMu.Unlock()
175 _, ok := s.liveOps[liveOperationKey(app, op)]
176 return ok
177}
178
179func liveOperationKey(app, op string) string {
180 return app + "\x00" + op
181}
182
183// WithGCBarrier wires the process-wide GC barrier. Every workload
184// mutation holds a shared lease from resource selection through the
185// durable publication of the resource's protection, so prune can never
186// observe a half-acquired resource. Nil is valid for tests that never
187// run prune.
188func (s *Service) WithGCBarrier(barrier out.GCBarrier) *Service {
189 s.barrier = barrier
190 return s
191}
192
193// WithBindPolicies supplies the initial administrative bind policies.
194func (s *Service) WithBindPolicies(policies map[string]domain.AppBindPolicy) *Service {
195 s.SetBindPolicies(policies)
196 return s
197}
198
199// SetBindPolicies atomically replaces the administrative bind policies used
200// to authorize service binds. The map is copied so later caller mutation
201// cannot race an in-flight deploy.
202func (s *Service) SetBindPolicies(policies map[string]domain.AppBindPolicy) {
203 copied := make(map[string]domain.AppBindPolicy, len(policies))
204 for name, policy := range policies {
205 copied[name] = policy
206 }
207 s.bindMu.Lock()
208 s.bindPolicies = copied
209 s.bindMu.Unlock()
210}
211
212// snapshotBindPolicies returns the current policy map. SetBindPolicies never
213// mutates a published map, so the snapshot stays safe after the lock drops.
214func (s *Service) snapshotBindPolicies() map[string]domain.AppBindPolicy {
215 s.bindMu.RLock()
216 defer s.bindMu.RUnlock()
217 return s.bindPolicies
218}
219
220// WithDevicePolicies supplies the initial administrative device policies.
221func (s *Service) WithDevicePolicies(policies map[string]domain.AppDevicePolicy) *Service {
222 s.SetDevicePolicies(policies)
223 return s
224}
225
226// SetDevicePolicies atomically replaces the administrative device policies
227// used to authorize service devices. The map and its slices are copied so
228// later caller mutation cannot race an in-flight deploy.
229func (s *Service) SetDevicePolicies(policies map[string]domain.AppDevicePolicy) {
230 copied := make(map[string]domain.AppDevicePolicy, len(policies))
231 for name, policy := range policies {
232 policy.CDI = append([]string(nil), policy.CDI...)
233 policy.AllowedApps = append([]string(nil), policy.AllowedApps...)
234 policy.AllowedServices = append([]string(nil), policy.AllowedServices...)
235 copied[name] = policy
236 }
237 s.deviceMu.Lock()
238 s.devicePolicies = copied
239 s.deviceMu.Unlock()
240}
241
242// snapshotDevicePolicies returns the current policy map. SetDevicePolicies
243// never mutates a published map, so the snapshot stays safe after the
244// lock drops.
245func (s *Service) snapshotDevicePolicies() map[string]domain.AppDevicePolicy {
246 s.deviceMu.RLock()
247 defer s.deviceMu.RUnlock()
248 return s.devicePolicies
249}
250
251// WithCapabilityPolicies supplies the initial administrative capability policies.
252func (s *Service) WithCapabilityPolicies(policies map[string]domain.AppCapabilityPolicy) *Service {
253 s.SetCapabilityPolicies(policies)
254 return s
255}
256
257// SetCapabilityPolicies atomically replaces the administrative capability
258// policies used to authorize service capabilities. The map and its slices
259// are copied so later caller mutation cannot race an in-flight deploy.
260func (s *Service) SetCapabilityPolicies(policies map[string]domain.AppCapabilityPolicy) {
261 copied := make(map[string]domain.AppCapabilityPolicy, len(policies))
262 for name, policy := range policies {
263 policy.Add = append([]string(nil), policy.Add...)
264 policy.AllowedApps = append([]string(nil), policy.AllowedApps...)
265 policy.AllowedServices = append([]string(nil), policy.AllowedServices...)
266 copied[name] = policy
267 }
268 s.capabilityMu.Lock()
269 s.capabilityPolicies = copied
270 s.capabilityMu.Unlock()
271}
272
273// snapshotCapabilityPolicies returns the current policy map.
274// SetCapabilityPolicies never mutates a published map.
275func (s *Service) snapshotCapabilityPolicies() map[string]domain.AppCapabilityPolicy {
276 s.capabilityMu.RLock()
277 defer s.capabilityMu.RUnlock()
278 return s.capabilityPolicies
279}
280
281// resolveServiceCapabilities resolves every declared capability policy
282// against the current administrative policies into the Linux capabilities
283// to add. Errors name only the app, service, and policy.
284func (s *Service) resolveServiceCapabilities(app string, svc domain.AppService) ([]string, error) {
285 if len(svc.Capabilities) == 0 {
286 return nil, nil
287 }
288 caps, err := domain.ResolveAppCapabilities(app, svc.Name, svc.Capabilities, s.snapshotCapabilityPolicies())
289 if err != nil {
290 return nil, fmt.Errorf("deployment: %w", err)
291 }
292 return caps, nil
293}
294
295// verifyReusableCapabilities gates every path that reuses an existing
296// capability-bearing container (restart, start, health recovery). Policy
297// resolution alone is not enough: the container keeps the capability set it
298// was created with, so a policy narrowed after creation would still run with
299// the revoked capability. The container's configured CapAdd must therefore
300// not exceed Gordon's defaults plus the currently authorized grant;
301// otherwise the reuse is refused and a deploy must recreate it. A container
302// holding fewer capabilities than authorized is fine (least privilege).
303// A container that no longer exists is left to the caller's existing
304// rebuild path, which re-resolves the policy.
305func (s *Service) verifyReusableCapabilities(ctx context.Context, app string, eff domain.AppEffectiveService) error {
306 granted, err := s.resolveServiceCapabilities(app, eff.Spec)
307 if err != nil {
308 return err
309 }
310 if len(eff.Spec.Capabilities) == 0 || eff.Container == "" {
311 return nil
312 }
313 ctr, err := s.deps.Runtime.InspectContainer(ctx, eff.Container)
314 if err != nil {
315 if errors.Is(err, domain.ErrContainerNotFound) {
316 return nil
317 }
318 return fmt.Errorf("deployment: app %q service %q: cannot verify container capabilities: %w", app, eff.Spec.Name, domain.ErrCapabilityPolicy)
319 }
320 authorized := append(domain.DefaultAppCapAdd(), granted...)
321 if excess := domain.ExcessCapabilities(ctr.CapAdd, authorized); len(excess) > 0 {
322 return fmt.Errorf("deployment: app %q service %q: container holds capabilities beyond the current policy (%s); deploy to recreate it: %w",
323 app, eff.Spec.Name, strings.Join(excess, ", "), domain.ErrCapabilityPolicy)
324 }
325 return nil
326}
327
328// resolveServiceGrants resolves the device and capability policies of a
329// service in one fail-closed step.
330func (s *Service) resolveServiceGrants(app string, svc domain.AppService) (devices, caps []string, err error) {
331 if devices, err = s.resolveServiceDevices(app, svc); err != nil {
332 return nil, nil, err
333 }
334 if caps, err = s.resolveServiceCapabilities(app, svc); err != nil {
335 return nil, nil, err
336 }
337 return devices, caps, nil
338}
339
340// resolveServiceDevices resolves every declared logical device against the
341// current administrative policies into runtime CDI IDs. Errors name only
342// the app, service, and device: host inventory never appears.
343func (s *Service) resolveServiceDevices(app string, svc domain.AppService) ([]string, error) {
344 if len(svc.Devices) == 0 {
345 return nil, nil
346 }
347 ids, err := domain.ResolveAppDevices(app, svc.Name, svc.Devices, s.snapshotDevicePolicies())
348 if err != nil {
349 return nil, fmt.Errorf("deployment: %w", err)
350 }
351 return ids, nil
352}
353
354// resolveServiceBinds resolves every declared bind against the current
355// administrative policies into ephemeral runtime binds. Errors name only the
356// app, service, and mount: host source paths never appear.
357func (s *Service) resolveServiceBinds(app string, svc domain.AppService) ([]domain.ContainerBind, error) {
358 if len(svc.Binds) == 0 {
359 return nil, nil
360 }
361 policies := s.snapshotBindPolicies()
362 resolved := make([]domain.ContainerBind, 0, len(svc.Binds))
363 for _, bind := range svc.Binds {
364 policy, ok := policies[bind.Name]
365 if !ok {
366 return nil, fmt.Errorf("deployment: app %q service %q mount %q: no administrative mount policy configured: %w", app, svc.Name, bind.Name, domain.ErrBindPolicy)
367 }
368 r, err := policy.ResolveAppBind(app, svc.Name, bind)
369 if err != nil {
370 return nil, fmt.Errorf("deployment: app %q service %q mount %q: refused by administrative mount policy: %w", app, svc.Name, bind.Name, domain.ErrBindPolicy)
371 }
372 resolved = append(resolved, domain.ContainerBind(r))
373 }
374 // Apply the runtime guard here so a bind overlapping a protected
375 // Gordon path fails at preflight and before the serving generation is
376 // retired, not only at create time. createContainer keeps the full
377 // guard as the final check.
378 if err := s.deps.Guard.Check(&domain.ContainerConfig{Binds: resolved}); err != nil {
379 return nil, fmt.Errorf("deployment: app %q service %q: %w", app, svc.Name, err)
380 }
381 return resolved, nil
382}
383
384// acquireAppContext takes the GC shared lease first and the per-app
385// coordinator second, following the documented lock order
386// (GC barrier → per-app coordinator → registry lock → bbolt
387// transaction). The returned release drops both, coordinator first.
388func (s *Service) acquireAppContext(ctx context.Context, app string) (func(), error) {
389 lease, err := s.acquireSharedGC(ctx)
390 if err != nil {
391 return nil, err
392 }
393 release, err := s.coord.acquire(ctx, app)
394 if err != nil {
395 lease.Release()
396 return nil, err
397 }
398 return func() {
399 release()
400 lease.Release()
401 }, nil
402}
403
404// tryAcquireAppContext takes the GC shared lease, then tries the per-app
405// coordinator without blocking. ok is false when the app is busy; the
406// caller skips it. Periodic reconciliation still waits for an in-flight
407// prune, because a half-planned snapshot is exactly what must not be
408// raced.
409func (s *Service) tryAcquireAppContext(ctx context.Context, app string) (func(), bool) {
410 lease, err := s.acquireSharedGC(ctx)
411 if err != nil {
412 return nil, false
413 }
414 release, ok := s.coord.tryAcquire(app)
415 if !ok {
416 lease.Release()
417 return nil, false
418 }
419 return func() {
420 release()
421 lease.Release()
422 }, true
423}
424
425// acquireSharedGC takes a shared GC lease, or a no-op lease when no
426// barrier is wired.
427func (s *Service) acquireSharedGC(ctx context.Context) (out.GCLease, error) {
428 if s.barrier == nil {
429 return noopGCLease{}, nil
430 }
431 return s.barrier.AcquireShared(ctx)
432}
433
434// noopGCLease is the lease used when no barrier is wired.
435type noopGCLease struct{}
436
437func (noopGCLease) Release() {}
438
439// WithProbeDeps overrides readiness probing (tests inject a mock-backed
440// ProbeDeps via NewProbeDeps; production leaves it nil for the live
441// runtime adapter).
442func (s *Service) WithProbeDeps(probes ProbeDeps) *Service {
443 s.probes = &probes
444 return s
445}
446
447// DeployInput selects the revision and service scope.
448type DeployInput struct {
449 App string
450 Revision string
451 Service string
452 Op string
453}
454
455// DeployResult carries terminal per-service results and cleanup warnings.
456type DeployResult struct {
457 Op string
458 App string
459 Revision string
460 Outcome string
461 Services map[string]ServiceResult
462 CleanupWarnings []CleanupWarning
463 // Removed names services retired because the new revision no longer
464 // declares them.
465 Removed []string
466}
467
468// ServiceResultUnchanged marks a deploy step that kept the running
469// container because its image, spec, and environment were already current.
470const ServiceResultUnchanged = domain.AppServiceUnchanged
471
472// ServiceResult is one service's terminal deployment result.
473type ServiceResult struct {
474 Result string
475 EffectiveRevision string
476 Before string
477 After string
478 RestartUnsafe bool
479 Error string
480 // VerifiedStartedAt is the execution start of After observed before its
481 // readiness ran. The caller records it as a verified execution once the
482 // generation is published and routed. Zero when unknown.
483 VerifiedStartedAt time.Time
484 // SupersededRunning is set by an overlap replacement: the container in
485 // Before still runs beside After and carries a retirement-pending marker
486 // until it is retired.
487 SupersededRunning bool
488 // BackendBinds carries the created container's TCP loopback publishes
489 // (container port -> 127.0.0.1 host port) for the active record.
490 BackendBinds map[int]int
491 // UDPBackendBinds carries the created container's UDP loopback
492 // publishes for the active record. Nil when the service declares
493 // no UDP interface.
494 UDPBackendBinds map[int]int
495 // CleanupWarnings are bounded leftovers of this service's terminal
496 // path: a candidate that could not be removed, or a backend claim
497 // that could not be released.
498 CleanupWarnings []CleanupWarning
499 // Diagnostics is bounded, redacted failure output of the failed
500 // candidate. It is never embedded in Error and is persisted to the
501 // journal for logs-scoped retrieval only.
502 Diagnostics []string
503}
504
505// CleanupWarning records post-publication leftovers for operator action.
506type CleanupWarning struct {
507 Service string
508 Leftover string
509 Detail string
510}
511
512// pinnedService binds a service spec to its resolved digest plus the
513// captured revision's app-wide public env (injected into every service
514// alongside that service's own resolved secrets). appEnv is a copy owned
515// by this pin: never mutated, never persisted with secret values.
516type pinnedService struct {
517 name string
518 spec domain.AppService
519 digest string
520 runtimeImage string
521 appEnv map[string]string
522 appNetworks []domain.AppSharedNetwork
523 // sharedNetworks are the declared shared-network memberships this
524 // service joins. The private incarnation network is derived from the
525 // app UUID at create time so recovery and deploy agree.
526 sharedNetworks []domain.AppSharedNetwork
527}
528
529// newOpID allocates a time-ordered op- identifier (uuid v7).
530func newOpID() string {
531 id, err := uuid.NewV7()
532 if err != nil {
533 id = uuid.New()
534 }
535 var buf [32]byte
536 hex.Encode(buf[:], id[:])
537 return "op-" + string(buf[:])
538}
539
540// claimDeploymentLocked is the first phase of a deploy: it resolves the
541// requested revision, enforces the targeted-deploy convergence rules, and
542// claims the request key, persisting the non-terminal journal. It performs
543// no image pull and no workload mutation, so a crash after it leaves a
544// durable claim that reconciliation can converge. owned is true only for the
545// single claimer of the key; a replay returns the stored journal instead.
546// The caller holds the app lock and has already recovered the store.
547func (s *Service) claimDeploymentLocked(ctx context.Context, input DeployInput) (domain.AppDesiredRevision, domain.AppOperation, bool, error) {
548 rev, err := s.resolveRevision(ctx, input)
549 if err != nil {
550 return domain.AppDesiredRevision{}, domain.AppOperation{}, false, err
551 }
552 if input.Service != "" {
553 if err := s.checkConverged(ctx, input.App, input.Service, rev); err != nil {
554 return domain.AppDesiredRevision{}, domain.AppOperation{}, false, err
555 }
556 }
557
558 op := domain.AppOperation{
559 Kind: "deploy",
560 App: input.App,
561 InputRevision: rev.Revision,
562 StartedAt: time.Now().UTC(),
563 Request: domain.AppOperationRequestFor("deploy", input.App, input.Revision, input.Service),
564 }
565 op, owned, err := s.claimOperation(ctx, input.Op, op, []domain.AppOperationStep{{ID: "preflight", State: domain.AppStepPending}})
566 if err != nil {
567 return domain.AppDesiredRevision{}, domain.AppOperation{}, false, err
568 }
569 return rev, op, owned, nil
570}
571
572// pinPreflightLocked is the second preflight phase: it runs the image
573// pull/pin and resource gates of an already claimed operation and records the
574// pinned table (or the terminal preflight failure) in the journal. The
575// caller holds the app lock and owns the claimed, non-terminal operation.
576func (s *Service) pinPreflightLocked(ctx context.Context, app, onlyService string, rev domain.AppDesiredRevision, op *domain.AppOperation) ([]pinnedService, error) {
577 log := zerowrap.FromCtx(ctx)
578
579 pinned, err := s.preflightServices(ctx, app, rev, onlyService)
580 if err != nil {
581 op.Steps[0] = domain.AppOperationStep{ID: "preflight", State: domain.AppStepFailed, Error: err.Error()}
582 op.Outcome = domain.AppOutcomeFailed
583 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
584 log.Warn().Err(saveErr).Msg("deployment: failed to record preflight failure")
585 }
586 return nil, err
587 }
588 op.Steps[0] = domain.AppOperationStep{ID: "preflight", State: domain.AppStepSucceeded}
589 // A resumed claim can carry the service-step plan of an earlier attempt,
590 // including candidates it recorded. Converge those candidates before
591 // replacing the plan: an unreconciled leftover may still be running, and
592 // the replacement this execution creates must never overlap it.
593 if err := s.convergeResumedCandidates(ctx, app, op, log); err != nil {
594 return nil, err
595 }
596 // Replace the persisted plan instead of appending: the preflight step plus
597 // one step per pinned service, so runServiceStep's index contract matches a
598 // normalized layout and no earlier step is overwritten or left pending.
599 plan := make([]domain.AppOperationStep, 1, len(pinned)+1)
600 plan[0] = op.Steps[0]
601 for _, p := range pinned {
602 plan = append(plan, domain.AppOperationStep{
603 ID: "service." + p.name + ".replace",
604 State: domain.AppStepPending,
605 Service: p.name,
606 Digest: p.digest,
607 Image: p.runtimeImage,
608 })
609 }
610 op.Steps = plan
611 if err := s.deps.State.SaveOperation(ctx, *op); err != nil {
612 return nil, fmt.Errorf("deployment: persist pinned table: %w", err)
613 }
614 log.Info().Str("op", op.Op).Str("revision", rev.Revision).Int("services", len(pinned)).Msg("deployment: preflight passed")
615 return pinned, nil
616}
617
618// convergeResumedCandidates converges any candidate an earlier attempt of a
619// resumed claim recorded, so replacing its service-step plan cannot drop a
620// leftover that may still be running. A fresh claim has only its preflight
621// step and does no work here.
622func (s *Service) convergeResumedCandidates(ctx context.Context, app string, op *domain.AppOperation, log zerowrap.Logger) error {
623 needsActive := false
624 for _, step := range op.Steps {
625 if reconcileNeedsContainer(step, false) {
626 needsActive = true
627 break
628 }
629 }
630 if !needsActive {
631 return nil
632 }
633 active, _, err := s.deps.State.LoadActive(ctx, app)
634 if err != nil {
635 return fmt.Errorf("deployment: load active for resumed claim: %w", err)
636 }
637 if _, err := s.convergeInterruptedSteps(ctx, app, op, active, false, log); err != nil {
638 return err
639 }
640 return nil
641}
642
643// claimOperation is the single claim point before any mutation effect.
644// A keyed request is claimed atomically: the first claim persists the
645// in-flight journal and reports owned=true; a repeat returns the stored
646// journal and reports owned=false. A key that already answered a
647// different request fails with ErrAppStateConflict, and a key for a name
648// with no app identity fails with ErrAppNotFound without writing
649// anything. An unkeyed request (internal recovery paths) starts a fresh
650// journal under a generated id.
651func (s *Service) claimOperation(ctx context.Context, key string, op domain.AppOperation, steps []domain.AppOperationStep) (domain.AppOperation, bool, error) {
652 op.Steps = steps
653 // A journal is never opened while a different operation of this app is
654 // still unfinished: that predecessor must be reconciled first, or this
655 // claim would mask it and its leftover generation could stay untracked.
656 latest, found, err := s.deps.State.LoadLatestOperation(ctx, op.App)
657 if err != nil {
658 return domain.AppOperation{}, false, fmt.Errorf("deployment: load latest operation: %w", err)
659 }
660 if found && !latest.Terminal() && latest.Op != key {
661 return domain.AppOperation{}, false, fmt.Errorf(
662 "deployment: app %q has unfinished operation %s: %w", op.App, latest.Op, domain.ErrAppStateConflict)
663 }
664 if key == "" {
665 op.Op = newOpID()
666 if err := s.deps.State.SaveOperation(ctx, op); err != nil {
667 return domain.AppOperation{}, false, fmt.Errorf("deployment: persist journal before effects: %w", err)
668 }
669 return op, true, nil
670 }
671 op.Op = key
672 existing, claimed, err := s.deps.State.ClaimOperation(ctx, op)
673 if err != nil {
674 return domain.AppOperation{}, false, err
675 }
676 if !claimed {
677 return existing, false, nil
678 }
679 return op, true, nil
680}
681
682// replayError reports a replayed key whose journal never reached a
683// terminal outcome: the operation is still in flight or was interrupted,
684// so its effects must not run again.
685func replayError(op domain.AppOperation) error {
686 if !op.Terminal() {
687 return fmt.Errorf("deployment: operation %s has not reached a terminal outcome: %w", op.Op, domain.ErrAppStateConflict)
688 }
689 if op.Outcome != domain.AppOutcomeSuccess {
690 return fmt.Errorf("deployment: operation %s previously ended with outcome %s: %w", op.Op, op.Outcome, domain.ErrAppStateConflict)
691 }
692 return nil
693}
694
695// resolveRevision loads the captured revision (default: current desired).
696// A revision that cannot be resolved on a name with no app identity at
697// all reports ErrAppNotFound: the failure is the missing app, not the
698// missing revision.
699func (s *Service) resolveRevision(ctx context.Context, input DeployInput) (domain.AppDesiredRevision, error) {
700 if input.Revision != "" {
701 rev, err := s.deps.State.LoadRevision(ctx, input.App, input.Revision)
702 if err != nil && errors.Is(err, domain.ErrAppRevisionNotFound) {
703 if knownErr := s.requireKnownApp(ctx, input.App); knownErr != nil {
704 return domain.AppDesiredRevision{}, knownErr
705 }
706 }
707 return rev, err
708 }
709 rev, ok, err := s.deps.State.LoadDesired(ctx, input.App)
710 if err != nil {
711 return domain.AppDesiredRevision{}, err
712 }
713 if !ok {
714 if knownErr := s.requireKnownApp(ctx, input.App); knownErr != nil {
715 return domain.AppDesiredRevision{}, knownErr
716 }
717 return domain.AppDesiredRevision{}, fmt.Errorf("deployment: app %q has no desired state: %w", input.App, domain.ErrAppRevisionNotFound)
718 }
719 return rev, nil
720}
721
722// requireKnownApp refuses a mutation of a name with no live app identity.
723func (s *Service) requireKnownApp(ctx context.Context, app string) error {
724 exists, err := s.deps.State.AppExists(ctx, app)
725 if err != nil {
726 return fmt.Errorf("deployment: app existence: %w", err)
727 }
728 if !exists {
729 return fmt.Errorf("deployment: app %q does not exist: %w", app, domain.ErrAppNotFound)
730 }
731 return nil
732}
733
734// checkConverged permits a targeted deploy when every desired/effective
735// difference belongs to that service. App-wide and other-service changes
736// must be deployed together so ACTIVE never combines incompatible specs.
737func (s *Service) checkConverged(ctx context.Context, app, service string, rev domain.AppDesiredRevision) error {
738 active, ok, err := s.deps.State.LoadActive(ctx, app)
739 if err != nil {
740 return err
741 }
742 if !ok {
743 return fmt.Errorf("deployment: app %q was never deployed, targeted deploy refused: %w", app, domain.ErrAppStateConflict)
744 }
745 if active.Converged && active.ConvergedRevision == rev.Revision {
746 return nil
747 }
748 effective, err := domain.ActiveAppSpec(active, rev.Spec.Env, func(revision string) (map[string]string, error) {
749 return s.revisionEnv(ctx, app, revision)
750 })
751 if err != nil {
752 return fmt.Errorf("deployment: rebuild effective spec: %w", err)
753 }
754 diff := domain.DiffAppSpec(rev.Spec, effective)
755 prefix := "service/" + service + "/"
756 if len(diff.Added) > 0 {
757 return targetedDivergenceError(rev.Revision, diff.Added[0]+" (added)")
758 }
759 if len(diff.Removed) > 0 {
760 return targetedDivergenceError(rev.Revision, diff.Removed[0]+" (removed)")
761 }
762 for _, path := range diff.Changed {
763 if !strings.HasPrefix(path, prefix) {
764 return targetedDivergenceError(rev.Revision, path)
765 }
766 }
767 return nil
768}
769
770// revisionEnv returns the app env of a stored revision, or nil when the
771// revision was garbage-collected.
772func (s *Service) revisionEnv(ctx context.Context, app, revision string) (map[string]string, error) {
773 rev, err := s.deps.State.LoadRevision(ctx, app, revision)
774 if errors.Is(err, domain.ErrAppRevisionNotFound) {
775 return nil, nil
776 }
777 if err != nil {
778 return nil, err
779 }
780 return rev.Spec.Env, nil
781}
782
783func targetedDivergenceError(revision, path string) error {
784 return fmt.Errorf("%w: service-targeted deploy refused, desired %s also changes %s", domain.ErrAppStateConflict, revision, path)
785}
786
787// selectPreflightServices copies and optionally narrows a revision's services.
788func selectPreflightServices(rev domain.AppDesiredRevision, onlyService string) ([]domain.AppService, error) {
789 services := append([]domain.AppService(nil), rev.Spec.Services...)
790 if onlyService == "" {
791 return services, nil
792 }
793 for _, svc := range services {
794 if svc.Name == onlyService {
795 return []domain.AppService{svc}, nil
796 }
797 }
798 return nil, fmt.Errorf("deployment: service %q not in revision %s: %w", onlyService, rev.Revision, domain.ErrAppStateConflict)
799}
800
801// preflightServices runs the preflight gates in order. No mutation.
802func (s *Service) preflightServices(ctx context.Context, app string, rev domain.AppDesiredRevision, onlyService string) ([]pinnedService, error) {
803 services, err := selectPreflightServices(rev, onlyService)
804 if err != nil {
805 return nil, err
806 }
807 sort.Slice(services, func(i, j int) bool { return services[i].Name < services[j].Name })
808
809 var pinned []pinnedService
810 ownership, err := s.deps.State.LoadOwnership(ctx, app)
811 if err != nil {
812 return nil, err
813 }
814 // Defensive re-check: stored revisions predate or bypass apply
815 // validation; an overlap must fail preflight before any mutation.
816 if err := rev.Spec.CheckEnvSecretCollisions(); err != nil {
817 return nil, err
818 }
819 appEnv := maps.Clone(rev.Spec.Env)
820 if appEnv == nil {
821 appEnv = map[string]string{}
822 }
823 for _, svc := range services {
824 digest, err := s.deps.Images.ResolveDigest(ctx, svc.Image)
825 if err != nil {
826 return nil, fmt.Errorf("deployment: image %q unresolvable: %w", svc.Image, domain.ErrAppImageUnresolvable)
827 }
828 if err := s.checkSecrets(ctx, app, ownership.ID, svc); err != nil {
829 return nil, err
830 }
831 runtimeImage, err := s.preflightImage(ctx, svc.Image, digest)
832 if err != nil {
833 return nil, err
834 }
835 if err := s.checkImageVolumes(ctx, svc, runtimeImage); err != nil {
836 return nil, err
837 }
838 // Bind resolution stays a preflight gate: an unresolvable bind
839 // must fail before any workload mutation. The successful values
840 // are discarded because createAndStart re-resolves against the
841 // current policy immediately before the runtime mutation.
842 if _, err := s.resolveServiceBinds(app, svc); err != nil {
843 return nil, err
844 }
845 if err := s.preflightServiceDevices(ctx, app, svc, pinned); err != nil {
846 return nil, err
847 }
848 if _, err := s.resolveServiceCapabilities(app, svc); err != nil {
849 return nil, err
850 }
851 pinned = append(pinned, pinnedService{
852 name: svc.Name, spec: svc, digest: digest, runtimeImage: runtimeImage, appEnv: appEnv,
853 appNetworks: append([]domain.AppSharedNetwork(nil), rev.Spec.Networks...),
854 sharedNetworks: domain.AppServiceSharedNetworks(rev.Spec, svc.Name),
855 })
856 }
857 if err := s.recheckReservations(ctx, app, rev); err != nil {
858 return nil, err
859 }
860 if err := s.checkResources(ctx, app, ownership, services); err != nil {
861 return nil, err
862 }
863 return pinned, nil
864}
865
866// preflightServiceDevices runs the device authorization and engine
867// capability gates for one service. Split from preflightServices to keep
868// its complexity within budget.
869func (s *Service) preflightServiceDevices(ctx context.Context, app string, svc domain.AppService, pinned []pinnedService) error {
870 if _, err := s.resolveServiceDevices(app, svc); err != nil {
871 return err
872 }
873 // Engine capability is a knowable-before-mutation failure: a
874 // device-bearing revision on an unsupported engine must fail here,
875 // before withdrawal retires the serving generation. The check runs
876 // once per revision, on the first device-bearing service;
877 // CreateContainer keeps the same gate as defense.
878 if len(svc.Devices) > 0 && !devicesEngineChecked(pinned) {
879 if s.deps.Runtime == nil {
880 return fmt.Errorf("deployment: runtime unavailable: %w", domain.ErrRuntimeUnsupported)
881 }
882 // Sanitize the adapter cause: version-probe failures may embed
883 // the daemon endpoint, which must not reach journals or API
884 // responses. Cancellation and deadlines are the exception: they
885 // stay recognizable so aborted deploys are not misread as
886 // incapable engines.
887 if err := s.deps.Runtime.SupportsCDIDevices(ctx); err != nil {
888 return sanitizedEngineProbeError(svc.Name, err)
889 }
890 }
891 return nil
892}
893
894// sanitizedEngineProbeError maps an engine device-capability probe failure
895// onto the caller-visible error. Cancellation and deadlines survive so
896// shutdown and timeout handling keep working; every other cause collapses
897// to ErrRuntimeUnsupported alone, dropping the adapter text that may embed
898// the daemon endpoint.
899func sanitizedEngineProbeError(service string, err error) error {
900 switch {
901 case errors.Is(err, context.Canceled):
902 return fmt.Errorf("deployment: engine device capability check for service %q: %w", service, context.Canceled)
903 case errors.Is(err, context.DeadlineExceeded):
904 return fmt.Errorf("deployment: engine device capability check for service %q: %w", service, context.DeadlineExceeded)
905 default:
906 return fmt.Errorf("deployment: engine device capability check for service %q: %w", service, domain.ErrRuntimeUnsupported)
907 }
908}
909
910// devicesEngineChecked reports whether an earlier pinned service already
911// triggered the engine capability probe for this revision.
912func devicesEngineChecked(pinned []pinnedService) bool {
913 for _, p := range pinned {
914 if len(p.spec.Devices) > 0 {
915 return true
916 }
917 }
918 return false
919}
920
921// checkSecrets reads every required secret path once. Values stay in
922// memory for container creation only; nothing is written anywhere.
923// Paths are UUID-keyed so a removed app's secrets are never adopted
924// by a new app reusing the name.
925func (s *Service) checkSecrets(ctx context.Context, app, appID string, svc domain.AppService) error {
926 if len(svc.SecretFiles) > 0 && s.deps.SecretFiles == nil {
927 return fmt.Errorf("deployment: service %q delivers secrets as files but no secret file store is configured", svc.Name)
928 }
929 if s.deps.Secrets == nil {
930 return fmt.Errorf("deployment: secret provider unavailable: %w", domain.ErrAppSecretMissing)
931 }
932 for envKey, name := range svc.Secrets {
933 path := domain.AppSecretPathForID(appID, app, svc.Name, name)
934 if _, err := s.deps.Secrets.GetSecret(ctx, path); err != nil {
935 return fmt.Errorf("deployment: secret %s (env %s) missing: %w", path, envKey, domain.ErrAppSecretMissing)
936 }
937 }
938 return nil
939}
940
941// appSecretID returns the stable internal UUID for secret paths.
942// Empty means legacy name-keyed paths (records predating UUIDs).
943// Used on the execution path where preflight's ownership snapshot is
944// not in scope; deploys are infrequent, one extra store read per
945// service is acceptable.
946func (s *Service) appSecretID(ctx context.Context, app string) (string, error) {
947 ownership, err := s.deps.State.LoadOwnership(ctx, app)
948 if err != nil {
949 return "", err
950 }
951 return ownership.ID, nil
952}
953
954func (s *Service) preflightImage(ctx context.Context, image, digest string) (string, error) {
955 if err := validation.ValidateImageDigest(digest); err != nil {
956 return "", fmt.Errorf("deployment: image %q resolved to invalid digest: %w", image, domain.ErrAppImageUnresolvable)
957 }
958 if err := s.validateImageSource(image, digest); err != nil {
959 return "", err
960 }
961 if s.deps.Registry.Domain == "" {
962 return image, nil
963 }
964 return s.pullImage(ctx, stripImageTag(image)+"@"+digest)
965}
966
967// validateImageSource enforces the installation image policy on a
968// reference before any pull. Both the manifest reference and the pinned
969// digest are checked, so a digest-pinned ref cannot bypass the hostname
970// and port allowlist on deploy or recovery.
971func (s *Service) validateImageSource(image, digest string) error {
972 ref := strings.TrimSpace(image)
973 if digest != "" && !strings.Contains(ref, "@") {
974 ref = stripImageTag(image) + "@" + digest
975 }
976 if err := s.deps.ImagePolicy.ValidateImageSource(ref); err != nil {
977 return fmt.Errorf("deployment: image %q: %w", image, err)
978 }
979 return nil
980}
981
982// checkImageVolumes rejects images with unmapped VOLUME declarations.
983func (s *Service) checkImageVolumes(ctx context.Context, svc domain.AppService, runtimeImage string) error {
984 declared, err := s.deps.Runtime.InspectImageVolumes(ctx, runtimeImage)
985 if err != nil {
986 return fmt.Errorf("deployment: inspect image volumes: %w", err)
987 }
988 if len(declared) == 0 {
989 return nil
990 }
991 mapped := map[string]struct{}{}
992 for _, vol := range svc.Volumes {
993 mapped[vol.Path] = struct{}{}
994 }
995 // Bind destinations also map image-declared volumes: a bind that
996 // mounts over an image VOLUME is an explicit operator mapping.
997 for _, bind := range svc.Binds {
998 mapped[bind.Path] = struct{}{}
999 }
1000 for _, path := range declared {
1001 if _, ok := mapped[path]; !ok {
1002 return fmt.Errorf(
1003 "deployment: image declares unmanaged volume %q, map it explicitly: %w",
1004 path, domain.ErrAppUnmanagedImageVolume,
1005 )
1006 }
1007 }
1008 return nil
1009}
1010
1011// recheckReservations re-validates against the live global table.
1012func (s *Service) recheckReservations(ctx context.Context, app string, rev domain.AppDesiredRevision) error {
1013 checkpoint, err := s.deps.State.LoadCheckpoint(ctx)
1014 if err != nil {
1015 return err
1016 }
1017 reservations := domain.ReservationsFor(rev.Spec)
1018 if err := domain.CheckReservations(checkpoint.Reservations, reservations, app); err != nil {
1019 return fmt.Errorf("deployment: %w", err)
1020 }
1021 return nil
1022}
1023
1024// checkResources verifies volumes exist-or-creatable and ownership is consistent.
1025// A generated runtime volume that already exists but is NOT recorded in
1026// this app's ownership is refused fail-closed: mounting it would
1027// implicitly adopt (and possibly modify) foreign data. Volumes recorded
1028// in ownership (attached or retained) were created by this app's own
1029// deploys and are safe to reuse.
1030func (s *Service) checkResources(ctx context.Context, app string, ownership domain.AppOwnership, services []domain.AppService) error {
1031 owned := map[string]string{}
1032 ownedRuntime := map[string]struct{}{}
1033 for _, vol := range ownership.Volumes {
1034 owned[vol.Name] = vol.Service
1035 ownedRuntime[vol.RuntimeName] = struct{}{}
1036 }
1037 for _, svc := range services {
1038 for _, vol := range svc.Volumes {
1039 if owner, ok := owned[vol.Name]; ok && owner != svc.Name {
1040 return fmt.Errorf(
1041 "deployment: volume %q owned by %q, not %q: %w",
1042 vol.Name, owner, svc.Name, domain.ErrAppStateConflict,
1043 )
1044 }
1045 runtimeName := domain.RuntimeVolumeName(app, svc.Name, vol.Name)
1046 exists, err := s.deps.Runtime.VolumeExists(ctx, runtimeName)
1047 if err != nil {
1048 return fmt.Errorf("deployment: check volume %q: %w", runtimeName, err)
1049 }
1050 if exists {
1051 if _, ok := ownedRuntime[runtimeName]; !ok {
1052 return fmt.Errorf(
1053 "deployment: runtime volume %q already exists but is not owned by app %q: refusing to adopt foreign data: %w",
1054 runtimeName, app, domain.ErrAppStateConflict,
1055 )
1056 }
1057 }
1058 }
1059 }
1060 return nil
1061}
1062
1063// ComputeOutcome derives the op outcome over terminal per-service results.
1064func ComputeOutcome(results map[string]ServiceResult) string {
1065 deployed := 0
1066 failed := 0
1067 for _, result := range results {
1068 switch result.Result {
1069 case "deployed", ServiceResultUnchanged, domain.AppServiceRemoved:
1070 deployed++
1071 case "failed":
1072 failed++
1073 }
1074 }
1075 switch {
1076 case deployed > 0 && failed > 0:
1077 return domain.AppOutcomePartial
1078 case deployed > 0:
1079 return domain.AppOutcomeSuccess
1080 default:
1081 return domain.AppOutcomeFailed
1082 }
1083}
1084
1085// itoa formats integers without extra imports.
1086func itoa(n int) string {
1087 return fmt.Sprintf("%d", n)
1088}