1package deployment
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "sort"
8 "sync"
9 "time"
10
11 "github.com/bnema/zerowrap"
12
13 "github.com/bnema/gordon/internal/domain"
14)
15
16// Recovery pass bounds and policy constants.
17const (
18 // reconcileServiceDeadline bounds one service's recovery: inspect,
19 // start/restart, bind inspection, readiness, and publication. A hung
20 // service times out, releases its app lock, and never starves the
21 // apps or the services that follow it.
22 reconcileServiceDeadline = 2 * time.Minute
23 // stableReadyWindow is how long a generation must be observed
24 // running, ready, and published before its failure history resets.
25 stableReadyWindow = 5 * time.Minute
26 // unhealthyThreshold is the number of consecutive unhealthy
27 // observations required before recovery restarts a workload.
28 unhealthyThreshold = 2
29)
30
31// backoffDelays are the delays applied after the third consecutive
32// failure: 1, 2, 4, 8, then a 15-minute cap.
33var backoffDelays = []time.Duration{
34 1 * time.Minute,
35 2 * time.Minute,
36 4 * time.Minute,
37 8 * time.Minute,
38 15 * time.Minute,
39}
40
41// recoveryKey identifies one workload generation for recovery state.
42type recoveryKey struct {
43 app string
44 service string
45 container string
46}
47
48// recoveryBackoff is the in-memory crash-loop budget, keyed by
49// (app, service, exact container ID). It is intentionally not durable:
50// a daemon restart re-arms one real attempt per generation.
51type recoveryBackoff struct {
52 mu sync.Mutex
53 now func() time.Time
54 entries map[recoveryKey]*backoffEntry
55}
56
57type backoffEntry struct {
58 failures int
59 nextAttempt time.Time
60 healthySince time.Time
61 unhealthy int
62 // attemptPending marks a recovery intervention (start/restart) that
63 // has not yet been confirmed by a stable, verified observation. A
64 // generation found down again while an attempt is still pending is
65 // a crash loop and is charged, so a workload cannot restart every
66 // pass forever just because each individual start call succeeded.
67 attemptPending bool
68}
69
70func newRecoveryBackoff(now func() time.Time) *recoveryBackoff {
71 return &recoveryBackoff{now: now, entries: map[recoveryKey]*backoffEntry{}}
72}
73
74// allow reports whether one real attempt may run now.
75func (b *recoveryBackoff) allow(key recoveryKey) bool {
76 b.mu.Lock()
77 defer b.mu.Unlock()
78 entry, ok := b.entries[key]
79 if !ok {
80 return true
81 }
82 return !b.now().Before(entry.nextAttempt)
83}
84
85// recordFailure charges one failed start/restart or unhealthy restart.
86// The first three consecutive failures are real attempts; the fourth and
87// later ones delay the next attempt by 1, 2, 4, 8, then 15 minutes.
88// Only five continuously stable minutes (observeStable) or an explicit
89// forget reset the escalation, so a crash-looping generation cannot
90// escape its budget by waiting.
91func (b *recoveryBackoff) recordFailure(key recoveryKey) {
92 b.mu.Lock()
93 defer b.mu.Unlock()
94 now := b.now()
95 entry := b.ensure(key)
96 entry.failures++
97 entry.healthySince = time.Time{}
98 if entry.failures >= 3 {
99 index := entry.failures - 3
100 if index >= len(backoffDelays) {
101 index = len(backoffDelays) - 1
102 }
103 entry.nextAttempt = now.Add(backoffDelays[index])
104 }
105}
106
107// observeUnhealthy records one consecutive unhealthy observation and
108// returns the current streak length. `starting` and transient health
109// errors must not reach this method.
110func (b *recoveryBackoff) observeUnhealthy(key recoveryKey) int {
111 b.mu.Lock()
112 defer b.mu.Unlock()
113 entry := b.ensure(key)
114 entry.unhealthy++
115 return entry.unhealthy
116}
117
118// clearUnhealthy drops the unhealthy streak after a non-unhealthy
119// observation.
120func (b *recoveryBackoff) clearUnhealthy(key recoveryKey) {
121 b.mu.Lock()
122 defer b.mu.Unlock()
123 if entry, ok := b.entries[key]; ok {
124 entry.unhealthy = 0
125 }
126}
127
128// observeStable records a running, ready, published observation and
129// resets the generation's history after stableReadyWindow. It never
130// creates an entry: a generation with no recorded history is stable by
131// definition, so healthy services do not accumulate monitor state.
132func (b *recoveryBackoff) observeStable(key recoveryKey) {
133 b.mu.Lock()
134 defer b.mu.Unlock()
135 entry, ok := b.entries[key]
136 if !ok {
137 return
138 }
139 now := b.now()
140 entry.unhealthy = 0
141 if entry.healthySince.IsZero() {
142 entry.healthySince = now
143 return
144 }
145 if now.Sub(entry.healthySince) >= stableReadyWindow {
146 // Five continuously stable minutes: the generation is considered
147 // recovered, so its failure history and unconfirmed-attempt flag
148 // are dropped. An intervention is therefore never confirmed by a
149 // single good pass.
150 delete(b.entries, key)
151 }
152}
153
154// invalidateStable drops the stability streak without charging a
155// failure: an unhealthy, transitional, or unverifiable observation is
156// not five continuously stable minutes.
157func (b *recoveryBackoff) invalidateStable(key recoveryKey) {
158 b.mu.Lock()
159 defer b.mu.Unlock()
160 if entry, ok := b.entries[key]; ok {
161 entry.healthySince = time.Time{}
162 }
163}
164
165// markAttempt records a recovery intervention that must be confirmed by
166// a later stable observation.
167func (b *recoveryBackoff) markAttempt(key recoveryKey) {
168 b.mu.Lock()
169 defer b.mu.Unlock()
170 b.ensure(key).attemptPending = true
171}
172
173// attemptPending reports whether a previous recovery intervention of
174// this generation was never confirmed stable.
175func (b *recoveryBackoff) attemptPending(key recoveryKey) bool {
176 b.mu.Lock()
177 defer b.mu.Unlock()
178 entry, ok := b.entries[key]
179 return ok && entry.attemptPending
180}
181
182// ensure returns the entry for one key, creating it on first use. The
183// caller must hold mu.
184func (b *recoveryBackoff) ensure(key recoveryKey) *backoffEntry {
185 entry, ok := b.entries[key]
186 if !ok {
187 entry = &backoffEntry{}
188 b.entries[key] = entry
189 }
190 return entry
191}
192
193// forget drops one generation's history (stopped, removed, or replaced).
194func (b *recoveryBackoff) forget(key recoveryKey) {
195 b.mu.Lock()
196 defer b.mu.Unlock()
197 delete(b.entries, key)
198}
199
200// publicationInhibition is the in-memory record of services whose
201// fail-closed traffic withdrawal could not be applied. Such a service
202// must withdraw successfully again before any later publication.
203type publicationInhibition struct {
204 mu sync.Mutex
205 entries map[recoveryKey]struct{}
206}
207
208func (p *publicationInhibition) mark(app, service string) {
209 p.mu.Lock()
210 defer p.mu.Unlock()
211 if p.entries == nil {
212 p.entries = map[recoveryKey]struct{}{}
213 }
214 p.entries[recoveryKey{app: app, service: service}] = struct{}{}
215}
216
217func (p *publicationInhibition) clear(app, service string) {
218 p.mu.Lock()
219 defer p.mu.Unlock()
220 delete(p.entries, recoveryKey{app: app, service: service})
221}
222
223func (p *publicationInhibition) inhibited(app, service string) bool {
224 p.mu.Lock()
225 defer p.mu.Unlock()
226 _, ok := p.entries[recoveryKey{app: app, service: service}]
227 return ok
228}
229
230// pending returns the services of one app whose withdrawal is still
231// unproven, so a caller can retry them before publishing anything.
232func (p *publicationInhibition) pending(app string) []string {
233 p.mu.Lock()
234 defer p.mu.Unlock()
235 var services []string
236 for key := range p.entries {
237 if key.app == app {
238 services = append(services, key.service)
239 }
240 }
241 sort.Strings(services)
242 return services
243}
244
245// retainAppServices drops entries only for services no longer present in
246// ACTIVE. Marks for active services survive failed passes until a later
247// withdrawal is proven, so another publisher cannot restore stale traffic.
248func (p *publicationInhibition) retainAppServices(app string, live map[string]struct{}) {
249 p.mu.Lock()
250 defer p.mu.Unlock()
251 for key := range p.entries {
252 if key.app != app {
253 continue
254 }
255 if _, ok := live[key.service]; !ok {
256 delete(p.entries, key)
257 }
258 }
259}
260
261// executionTracker remembers the last observed execution start of each
262// generation so a native runtime restart is detected without Gordon
263// restarting the workload itself.
264type executionTracker struct {
265 mu sync.Mutex
266 seen map[recoveryKey]time.Time
267}
268
269// isNewExecution reports whether this observation still needs verification.
270// An unseen generation is treated as fresh: the daemon may have been absent
271// while the runtime restarted it, so accepting it without readiness and bind
272// verification could republish stale traffic. The observation is recorded
273// only after verification succeeds.
274func (t *executionTracker) isNewExecution(key recoveryKey, startedAt time.Time) bool {
275 t.mu.Lock()
276 defer t.mu.Unlock()
277 previous, ok := t.seen[key]
278 return !ok || !previous.Equal(startedAt)
279}
280
281// record stores the verified execution of one generation.
282func (t *executionTracker) record(key recoveryKey, startedAt time.Time) {
283 t.mu.Lock()
284 defer t.mu.Unlock()
285 if t.seen == nil {
286 t.seen = map[recoveryKey]time.Time{}
287 }
288 t.seen[key] = startedAt
289}
290
291// forget drops one generation's observation.
292func (t *executionTracker) forget(key recoveryKey) {
293 t.mu.Lock()
294 defer t.mu.Unlock()
295 delete(t.seen, key)
296}
297
298// ReconcileRunning is the daemon-owned periodic recovery pass. It never
299// consults DESIRED to select content, never changes intent, never
300// pulls, creates, or removes a container, and never starts a
301// container whose ID is absent from ACTIVE. Every app is attempted;
302// failures are aggregated so one failing app cannot hide the others.
303func (s *Service) ReconcileRunning(ctx context.Context) error {
304 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
305 zerowrap.FieldLayer: "usecase",
306 zerowrap.FieldUseCase: "ReconcileRunning",
307 })
308 if s.deps.State == nil {
309 return nil
310 }
311 apps, err := s.deps.State.ListApps(ctx)
312 if err != nil {
313 return fmt.Errorf("deployment: list apps for reconciliation: %w", err)
314 }
315 var failures []error
316 for _, app := range apps {
317 // Bound the whole app, not each service cumulatively. A multi-service
318 // app with hung runtime calls must not starve every later app forever.
319 appCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
320 err := s.reconcileAppRunning(appCtx, app)
321 cancel()
322 if err != nil {
323 failures = append(failures, err)
324 }
325 }
326 return errors.Join(failures...)
327}
328
329// reconcileAppRunning recovers one app. A busy app (an in-flight user
330// mutation) is skipped, never waited on: the periodic pass must not
331// starve later apps or shutdown.
332func (s *Service) reconcileAppRunning(ctx context.Context, app string) error {
333 release, ok := s.tryAcquireAppContext(ctx, app)
334 if !ok {
335 s.log.Debug().Str("app", app).Msg("deployment: reconciliation skipped, app is busy")
336 return nil
337 }
338 defer release()
339
340 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
341 zerowrap.FieldLayer: "usecase",
342 zerowrap.FieldUseCase: "ReconcileRunning",
343 "app": app,
344 })
345 intent, err := s.deps.State.LoadIntent(ctx, app)
346 if err != nil {
347 return fmt.Errorf("deployment: reconcile %q: load intent: %w", app, err)
348 }
349 active, ok, err := s.deps.State.LoadActive(ctx, app)
350 if err != nil {
351 return fmt.Errorf("deployment: reconcile %q: load active: %w", app, err)
352 }
353 if !ok || len(active.Services) == 0 {
354 s.pruneRecoveryHistory(app, active)
355 return nil
356 }
357 inhibitions, err := s.deps.State.LoadRecoveryInhibitions(ctx, app)
358 if err != nil {
359 return fmt.Errorf("deployment: reconcile %q: load recovery inhibitions: %w", app, err)
360 }
361 inhibited, failures := s.processRecoveryInhibitions(ctx, app, inhibitions)
362 for _, name := range sortedServiceNames(active) {
363 eff := active.Services[name]
364 if intent.Stopped {
365 if err := s.convergeStoppedService(ctx, app, name, eff); err != nil {
366 failures = append(failures, err)
367 }
368 continue
369 }
370 if err := s.reconcileRunningService(ctx, app, name, eff, inhibited); err != nil {
371 failures = append(failures, err)
372 }
373 }
374 if intent.Stopped {
375 // A stopped app must not stay published: rebuild the projection
376 // so no revoked backend remains routable. The publication is
377 // bounded like any other recovery effect.
378 publishCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
379 if err := s.refreshTraffic(publishCtx, app); err != nil {
380 failures = append(failures, err)
381 }
382 cancel()
383 }
384 s.pruneRecoveryHistory(app, active)
385 return errors.Join(failures...)
386}
387
388func (s *Service) processRecoveryInhibitions(ctx context.Context, app string, inhibitions []domain.AppRecoveryInhibition) (map[recoveryKey]struct{}, []error) {
389 inhibited := map[recoveryKey]struct{}{}
390 var failures []error
391 for _, inhibition := range inhibitions {
392 if inhibition.Reason != domain.AppInhibitRetirementPending {
393 inhibited[recoveryKey{app: app, service: inhibition.Service, container: inhibition.ContainerID}] = struct{}{}
394 continue
395 }
396 retired := s.retireContainer(ctx, app, retireOptions{Service: inhibition.Service}, inhibition.ContainerID)
397 if !retired.Gone {
398 failures = append(failures, fmt.Errorf("deployment: retirement pending for %s/%s", app, inhibition.ContainerID))
399 }
400 }
401 return inhibited, failures
402}
403
404// convergeStoppedService enforces durable stopped intent: it never
405// starts or publishes, and it stops a verified running exact ACTIVE ID
406// without deleting the container or its volumes.
407func (s *Service) convergeStoppedService(ctx context.Context, app, name string, eff domain.AppEffectiveService) error {
408 key := recoveryKey{app: app, service: name, container: eff.Container}
409 s.backoff.forget(key)
410 s.executions.forget(key)
411 if eff.Container == "" || s.deps.Runtime == nil {
412 return nil
413 }
414 serviceCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
415 defer cancel()
416 container, err := s.deps.Runtime.InspectContainer(serviceCtx, eff.Container)
417 if err != nil {
418 if errors.Is(err, domain.ErrContainerNotFound) {
419 return nil
420 }
421 return fmt.Errorf("deployment: stopped app %q service %q: inspect %s: %w", app, name, eff.Container, err)
422 }
423 if container.Status != string(domain.ContainerStatusRunning) && !isTransitionalStatus(container.Status) {
424 return nil
425 }
426 if err := s.deps.Runtime.StopContainer(serviceCtx, eff.Container, serviceStopGrace(eff)); err != nil && !errors.Is(err, domain.ErrContainerNotFound) {
427 return fmt.Errorf("deployment: stopped app %q service %q: stop %s: %w", app, name, eff.Container, err)
428 }
429 s.log.Info().Str("app", app).Str("service", name).Str("container", eff.Container).
430 Msg("deployment: stopped intent enforced on a revived container")
431 return nil
432}
433
434// reconcileRunningService recovers one service to the running intent
435// carried by ACTIVE. It only ever acts on the exact ACTIVE container ID.
436func (s *Service) reconcileRunningService(ctx context.Context, app, name string, eff domain.AppEffectiveService, inhibited map[recoveryKey]struct{}) error {
437 if eff.Container == "" {
438 // Absent ACTIVE: never act, never reconstruct.
439 return nil
440 }
441 if _, blocked := inhibited[recoveryKey{app: app, service: name, container: eff.Container}]; blocked {
442 return s.refuseInhibitedRecovery(ctx, app, name, eff.Container)
443 }
444 if s.deps.Runtime == nil {
445 return nil
446 }
447 serviceCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
448 defer cancel()
449 key := recoveryKey{app: app, service: name, container: eff.Container}
450 container, fresh, err := s.inspectForRecovery(serviceCtx, app, name, eff, key)
451 if err != nil {
452 return err
453 }
454 return s.convergeRecoveredService(serviceCtx, app, name, eff, container, key, fresh)
455}
456
457// inspectForRecovery retries a pending withdrawal, classifies the exact
458// ACTIVE container, and makes sure nothing unverifiable stays routable
459// before the workload is touched. A transient inspection error never
460// changes the workload or its traffic.
461func (s *Service) inspectForRecovery(ctx context.Context, app, name string, eff domain.AppEffectiveService, key recoveryKey) (*domain.Container, bool, error) {
462 // A previous withdrawal that could not be applied must be retried
463 // before anything else: never publish on top of stale forwarding.
464 withdrawn := false
465 if s.publication.inhibited(app, name) {
466 if err := s.withdrawForRecovery(ctx, app, name); err != nil {
467 return nil, false, err
468 }
469 withdrawn = true
470 }
471 container, err := s.deps.Runtime.InspectContainer(ctx, eff.Container)
472 if err != nil {
473 return nil, false, s.recoveryInspectError(ctx, app, name, eff, err)
474 }
475 // Never serve a backend that is not verifiably serving: a native
476 // restart (fresh execution) or a container that is not running must
477 // be withdrawn, and that withdrawal must succeed, before Gordon
478 // touches the workload. A transitional state keeps its traffic: the
479 // runtime is restarting it right now and it will be back within the
480 // pass cadence.
481 fresh := s.executions.isNewExecution(key, container.StartedAt)
482 unverified := container.Status != string(domain.ContainerStatusRunning) && !isTransitionalStatus(container.Status)
483 if (fresh || unverified) && !withdrawn {
484 if err := s.withdrawForRecovery(ctx, app, name); err != nil {
485 return nil, false, err
486 }
487 }
488 if fresh {
489 s.backoff.clearUnhealthy(key)
490 }
491 return container, fresh, nil
492}
493
494// convergeRecoveredService drives the exact ACTIVE container to a
495// verified, published state, or leaves it withdrawn with a reported
496// reason.
497func (s *Service) convergeRecoveredService(ctx context.Context, app, name string, eff domain.AppEffectiveService, container *domain.Container, key recoveryKey, fresh bool) error {
498 started, verify, err := s.driveRecoveredContainer(ctx, app, name, eff, container, key)
499 if err != nil {
500 return errors.Join(s.withdrawForRecovery(ctx, app, name), err)
501 }
502 if !verify {
503 // Transitional or backoff-delayed: observe now, converge later.
504 s.backoff.invalidateStable(key)
505 return nil
506 }
507 restarted, stable, err := s.recoverHealth(ctx, app, name, eff, key)
508 if err != nil {
509 s.backoff.invalidateStable(key)
510 return errors.Join(s.withdrawForRecovery(ctx, app, name), err)
511 }
512 if err := s.publishRecoveredService(ctx, app, name, eff, container, key, started, restarted, fresh); err != nil {
513 return err
514 }
515 if stable {
516 s.backoff.observeStable(key)
517 } else {
518 s.backoff.invalidateStable(key)
519 }
520 return nil
521}
522
523// refuseInhibitedRecovery reports a generation whose recovery is durably
524// inhibited and withdraws its backend best-effort: it must not stay
525// routable, and it must never be started.
526func (s *Service) refuseInhibitedRecovery(ctx context.Context, app, name, containerID string) error {
527 withdrawCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
528 defer cancel()
529 return errors.Join(
530 fmt.Errorf(
531 "deployment: app %q service %q container %s has recovery inhibited; no start attempted",
532 app, name, containerID,
533 ),
534 s.withdrawForRecovery(withdrawCtx, app, name),
535 )
536}
537
538// recoveryInspectError classifies one failed inspection: a confirmed
539// missing container withdraws traffic and reports without reconstruction,
540// while a transient error leaves the workload and its traffic untouched.
541func (s *Service) recoveryInspectError(ctx context.Context, app, name string, eff domain.AppEffectiveService, err error) error {
542 if !errors.Is(err, domain.ErrContainerNotFound) {
543 return fmt.Errorf("deployment: app %q service %q inspect %s: %w", app, name, eff.Container, err)
544 }
545 // The container is confirmed gone: its loopback reservations must not
546 // outlive it, or a later workload that reuses the port would collide
547 // with a claim no container holds.
548 releaseErr := s.deps.State.ReleaseBackendBinds(ctx, app, eff.Container)
549 if releaseErr != nil {
550 releaseErr = fmt.Errorf("deployment: release claims of gone container %s: %w", eff.Container, releaseErr)
551 }
552 return errors.Join(
553 s.withdrawForRecovery(ctx, app, name),
554 releaseErr,
555 fmt.Errorf("deployment: app %q service %q container %s is gone; traffic withdrawn, no reconstruction: %w",
556 app, name, eff.Container, domain.ErrContainerNotFound),
557 )
558}
559
560// driveRecoveredContainer brings the exact ACTIVE container to a running
561// state: an already running container is left alone, a transitional one
562// is observed and retried later, an unsafe state errors non-destructively,
563// and a stopped one is started under the generation's backoff budget.
564// verify reports whether the caller may continue to binds, readiness, and
565// publication: a transitional or delayed generation must not be published
566// on this pass.
567func (s *Service) driveRecoveredContainer(ctx context.Context, app, name string, eff domain.AppEffectiveService, container *domain.Container, key recoveryKey) (started bool, verify bool, err error) {
568 switch {
569 case container.Status == string(domain.ContainerStatusRunning):
570 return false, true, nil
571 case isTransitionalStatus(container.Status):
572 // restarting/starting: observe and retry on a later pass.
573 return false, false, nil
574 case container.Status == string(domain.ContainerStatusPaused) || container.Status == string(domain.ContainerStatusUnknown) || container.Status == "":
575 return false, false, fmt.Errorf("deployment: app %q service %q container %s is %q; refusing non-destructive recovery",
576 app, name, eff.Container, container.Status)
577 default:
578 // created/exited/dead/stopped with running intent: start the
579 // exact ACTIVE ID regardless of exit code.
580 if !s.backoff.allow(key) {
581 return false, false, nil
582 }
583 // A generation found down again before a previous recovery was
584 // ever confirmed stable is crash-looping: charge it, so a
585 // workload that dies after every successful start cannot be
586 // restarted on every pass forever.
587 if s.backoff.attemptPending(key) {
588 s.backoff.recordFailure(key)
589 if !s.backoff.allow(key) {
590 return false, false, nil
591 }
592 }
593 started, err := s.startRecovered(ctx, app, name, eff, key)
594 return started, err == nil, err
595 }
596}
597
598// publishRecoveredService verifies one generation in the order the plan
599// requires: read the container's live binds, probe readiness against
600// them, revalidate execution identity and ACTIVE, persist the verified
601// binds, and only then publish. Binds are never persisted to ACTIVE
602// before readiness, so no other publication can project an unverified
603// backend while this pass waits for the workload. The execution is
604// recorded only on success, so a failed verification is retried on the
605// next pass instead of being silently accepted.
606func (s *Service) publishRecoveredService(ctx context.Context, app, name string, eff domain.AppEffectiveService, observed *domain.Container, key recoveryKey, started, restarted, fresh bool) error {
607 binds, udpBinds, err := s.inspectBackendBinds(ctx, app, name, eff.Container, backendPorts(eff.Spec))
608 if err != nil {
609 s.backoff.recordFailure(key)
610 return errors.Join(s.withdrawForRecovery(ctx, app, name), err)
611 }
612 if started || restarted || fresh {
613 if err := s.waitServiceReady(ctx, app, eff.Container, eff.Spec, binds); err != nil {
614 s.backoff.recordFailure(key)
615 return errors.Join(s.withdrawForRecovery(ctx, app, name), err)
616 }
617 }
618 verifiedBinds, verifiedUDPBinds, err := s.revalidateAndPersistRecoveredBinds(ctx, app, name, eff, observed, binds, udpBinds)
619 if err != nil {
620 s.backoff.recordFailure(key)
621 return errors.Join(s.withdrawForRecovery(ctx, app, name), err)
622 }
623 if started || restarted || fresh || !equalBinds(verifiedBinds, eff.BackendBinds) || !equalBinds(verifiedUDPBinds, eff.UDPBackendBinds) {
624 if err := s.refreshTraffic(ctx, app); err != nil {
625 return err
626 }
627 }
628 s.executions.record(key, observed.StartedAt)
629 return nil
630}
631
632// startRecovered starts the exact ACTIVE container. A failed start is
633// re-inspected: if native restart policy already revived it, no failure
634// is charged and no second restart is issued.
635func (s *Service) startRecovered(serviceCtx context.Context, app, name string, eff domain.AppEffectiveService, key recoveryKey) (bool, error) {
636 // Fail closed before mutating the runtime if a bind's policy was
637 // revoked or became invalid since ACTIVE was published.
638 if _, err := s.resolveServiceBinds(app, eff.Spec); err != nil {
639 return false, err
640 }
641 // Revoked device grants fail before any runtime mutation.
642 if _, err := s.resolveServiceDevices(app, eff.Spec); err != nil {
643 return false, err
644 }
645 if err := s.deps.Runtime.StartContainer(serviceCtx, eff.Container); err != nil {
646 if errors.Is(err, domain.ErrContainerNotFound) {
647 s.backoff.forget(key)
648 return false, errors.Join(
649 fmt.Errorf("deployment: app %q service %q container %s is gone: %w", app, name, eff.Container, domain.ErrContainerNotFound),
650 )
651 }
652 recovered, inspectErr := s.deps.Runtime.InspectContainer(serviceCtx, eff.Container)
653 if inspectErr == nil && (recovered.Status == string(domain.ContainerStatusRunning) || isTransitionalStatus(recovered.Status)) {
654 s.log.Info().Str("app", app).Str("service", name).Str("container", eff.Container).
655 Msg("deployment: native restart won the race, no recovery charged")
656 s.backoff.clearUnhealthy(key)
657 return true, nil
658 }
659 s.backoff.recordFailure(key)
660 return false, fmt.Errorf("deployment: app %q service %q start %s: %w", app, name, eff.Container, err)
661 }
662 s.backoff.markAttempt(key)
663 s.log.Info().Str("app", app).Str("service", name).Str("container", eff.Container).
664 Msg("deployment: recovery started the exact active container")
665 return true, nil
666}
667
668// recoverHealth restarts a workload that reported `unhealthy` on
669// consecutive passes. `starting` waits; a missing healthcheck never
670// triggers a restart. Restarts share the generation's backoff budget and
671// are marked as unconfirmed attempts. The second result reports whether
672// the observation counts as stable ("running, ready, published"): an
673// unhealthy, still-starting, or unreadable health state must not extend
674// the five-minute stability window.
675func (s *Service) recoverHealth(serviceCtx context.Context, app, name string, eff domain.AppEffectiveService, key recoveryKey) (restarted bool, stable bool, err error) {
676 status, hasHealthcheck, statusErr := s.deps.Runtime.GetContainerHealthStatus(serviceCtx, eff.Container)
677 if statusErr != nil || !hasHealthcheck {
678 // A transient health error is not an unhealthy workload, but it
679 // is also not a proven healthy one.
680 return false, statusErr == nil, nil
681 }
682 if status != "unhealthy" {
683 s.backoff.clearUnhealthy(key)
684 return false, status != "starting", nil
685 }
686 if s.backoff.observeUnhealthy(key) < unhealthyThreshold {
687 return false, false, nil
688 }
689 if !s.backoff.allow(key) {
690 return false, false, nil
691 }
692 // Fail closed before mutating the runtime if a bind's policy was
693 // revoked or became invalid since ACTIVE was published.
694 if _, err := s.resolveServiceBinds(app, eff.Spec); err != nil {
695 return false, false, err
696 }
697 // Revoked device grants fail before any runtime mutation.
698 if _, err := s.resolveServiceDevices(app, eff.Spec); err != nil {
699 return false, false, err
700 }
701 if err := s.deps.Runtime.RestartContainer(serviceCtx, eff.Container, serviceStopGrace(eff)); err != nil {
702 s.backoff.recordFailure(key)
703 return false, false, fmt.Errorf("deployment: app %q service %q restart unhealthy %s: %w", app, name, eff.Container, err)
704 }
705 s.backoff.markAttempt(key)
706 s.backoff.clearUnhealthy(key)
707 s.log.Info().Str("app", app).Str("service", name).Str("container", eff.Container).
708 Msg("deployment: restarted an unhealthy container")
709 return true, false, nil
710}
711
712// revalidateAndPersistRecoveredBinds revalidates ACTIVE and the observed
713// execution immediately before publication, then persists the verified
714// binds (already re-read from the exact container) into the ACTIVE
715// record. Stale evidence never publishes: a native restart that happened
716// while this pass waited for readiness changes the observed start and
717// aborts the pass instead of publishing the previous execution's binds.
718func (s *Service) revalidateAndPersistRecoveredBinds(ctx context.Context, app, name string, eff domain.AppEffectiveService, observed *domain.Container, binds, udpBinds map[int]int) (map[int]int, map[int]int, error) {
719 current, err := s.deps.Runtime.InspectContainer(ctx, eff.Container)
720 if err != nil {
721 return nil, nil, fmt.Errorf("deployment: re-inspect %s before publication: %w", eff.Container, err)
722 }
723 if observed != nil && !current.StartedAt.Equal(observed.StartedAt) {
724 return nil, nil, fmt.Errorf("deployment: app %q service %q container %s restarted during recovery: %w",
725 app, name, eff.Container, domain.ErrAppStateConflict)
726 }
727 active, ok, err := s.deps.State.LoadActive(ctx, app)
728 if err != nil {
729 return nil, nil, fmt.Errorf("deployment: reload active for recovery: %w", err)
730 }
731 if !ok {
732 return nil, nil, fmt.Errorf("deployment: app %q has no active state: %w", app, domain.ErrAppStateConflict)
733 }
734 recorded, ok := active.Services[name]
735 if !ok || recorded.Container != eff.Container || recorded.EffectiveRevision != eff.EffectiveRevision {
736 return nil, nil, fmt.Errorf("deployment: app %q service %q generation changed during recovery: %w", app, name, domain.ErrAppStateConflict)
737 }
738 if !equalBinds(binds, recorded.BackendBinds) || !equalBinds(udpBinds, recorded.UDPBackendBinds) {
739 recorded.BackendBinds = binds
740 recorded.UDPBackendBinds = udpBinds
741 active.Services[name] = recorded
742 if err := s.deps.State.SaveActive(ctx, active); err != nil {
743 return nil, nil, fmt.Errorf("deployment: persist recovered binds: %w", err)
744 }
745 }
746 return binds, udpBinds, nil
747}
748
749// withdrawForRecovery applies one serialized fail-closed withdrawal of
750// a service. A failed application retains an in-memory publication
751// inhibition and never releases backend claims.
752func (s *Service) withdrawForRecovery(ctx context.Context, app, service string) error {
753 if s.deps.Traffic == nil {
754 return nil
755 }
756 if err := s.deps.Traffic.WithdrawService(ctx, app, service); err != nil {
757 s.publication.mark(app, service)
758 return fmt.Errorf("deployment: withdraw %s/%s: %w", app, service, err)
759 }
760 s.publication.clear(app, service)
761 return nil
762}
763
764// recoveryInhibited reports whether one exact generation is durably
765// inhibited for recovery.
766func (s *Service) recoveryInhibited(ctx context.Context, app, service, containerID string) (bool, error) {
767 if containerID == "" {
768 return false, nil
769 }
770 inhibitions, err := s.deps.State.LoadRecoveryInhibitions(ctx, app)
771 if err != nil {
772 return false, fmt.Errorf("deployment: load recovery inhibitions: %w", err)
773 }
774 for _, inhibition := range inhibitions {
775 if inhibition.Service == service && inhibition.ContainerID == containerID {
776 return true, nil
777 }
778 }
779 return false, nil
780}
781
782// inhibitRecovery durably records a generation-scoped recovery
783// inhibition for one exact container ID.
784func (s *Service) inhibitRecovery(ctx context.Context, app, service, containerID, reason, operation string) error {
785 if containerID == "" {
786 return nil
787 }
788 inhibition := domain.AppRecoveryInhibition{
789 App: app,
790 Service: service,
791 ContainerID: containerID,
792 Reason: reason,
793 Operation: operation,
794 CreatedAt: time.Now().UTC(),
795 }
796 if err := s.deps.State.SaveRecoveryInhibition(ctx, inhibition); err != nil {
797 return fmt.Errorf("deployment: record recovery inhibition for %s/%s: %w", app, containerID, err)
798 }
799 return nil
800}
801
802// clearRecoveryInhibition drops the inhibition of a safely superseded
803// generation.
804func (s *Service) clearRecoveryInhibition(ctx context.Context, app, service, containerID string) error {
805 if containerID == "" {
806 return nil
807 }
808 if err := s.deps.State.ClearRecoveryInhibition(ctx, app, service, containerID); err != nil {
809 return fmt.Errorf("deployment: clear recovery inhibition for %s/%s: %w", app, containerID, err)
810 }
811 return nil
812}
813
814// pruneRecoveryHistory drops in-memory state of generations that are no
815// longer present in ACTIVE. Durable inhibition records are never
816// dropped here: only an explicit supersede or removal clears them.
817func (s *Service) pruneRecoveryHistory(app string, active domain.AppActive) {
818 live := map[recoveryKey]struct{}{}
819 liveServices := map[string]struct{}{}
820 for name, svc := range active.Services {
821 live[recoveryKey{app: app, service: name, container: svc.Container}] = struct{}{}
822 liveServices[name] = struct{}{}
823 }
824 s.backoff.mu.Lock()
825 for key := range s.backoff.entries {
826 if key.app != app {
827 continue
828 }
829 if _, ok := live[key]; !ok {
830 delete(s.backoff.entries, key)
831 }
832 }
833 s.backoff.mu.Unlock()
834 s.executions.mu.Lock()
835 for key := range s.executions.seen {
836 if key.app != app {
837 continue
838 }
839 if _, ok := live[key]; !ok {
840 delete(s.executions.seen, key)
841 }
842 }
843 s.executions.mu.Unlock()
844 s.publication.retainAppServices(app, liveServices)
845}
846
847// isTransitionalStatus reports runtime states that must be observed and
848// retried rather than acted on.
849func isTransitionalStatus(status string) bool {
850 return status == "restarting" || status == "starting"
851}