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// executionStart returns the start time of the container's current
299// execution, or the zero time when it cannot be read: an unknown execution is
300// simply never recorded as verified, so the periodic pass verifies it.
301func (s *Service) executionStart(ctx context.Context, containerID string) time.Time {
302 if s.deps.Runtime == nil || containerID == "" {
303 return time.Time{}
304 }
305 container, err := s.deps.Runtime.InspectContainer(ctx, containerID)
306 if err != nil || container == nil {
307 return time.Time{}
308 }
309 return container.StartedAt
310}
311
312// recordVerifiedExecution records the execution a deploy or restart verified
313// and routed, so the first periodic pass after it does not treat the new
314// generation as an unseen execution, withdraw it, and verify it again. Only a
315// successful, non-zero observation is recorded; failure paths never are, and
316// a later native restart still differs from the recorded start time.
317func (s *Service) recordVerifiedExecution(app, service string, result ServiceResult) {
318 if result.Result == "failed" || result.After == "" || result.VerifiedStartedAt.IsZero() {
319 return
320 }
321 s.executions.record(recoveryKey{app: app, service: service, container: result.After}, result.VerifiedStartedAt)
322}
323
324// ReconcileRunning is the daemon-owned periodic recovery pass. It never
325// consults DESIRED to select content, never changes intent, never
326// pulls, creates, or removes a container, and never starts a
327// container whose ID is absent from ACTIVE. Every app is attempted;
328// failures are aggregated so one failing app cannot hide the others.
329func (s *Service) ReconcileRunning(ctx context.Context) error {
330 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
331 zerowrap.FieldLayer: "usecase",
332 zerowrap.FieldUseCase: "ReconcileRunning",
333 })
334 if s.deps.State == nil {
335 return nil
336 }
337 apps, err := s.deps.State.ListApps(ctx)
338 if err != nil {
339 return fmt.Errorf("deployment: list apps for reconciliation: %w", err)
340 }
341 var failures []error
342 for _, app := range apps {
343 // Bound the whole app, not each service cumulatively. A multi-service
344 // app with hung runtime calls must not starve every later app forever.
345 appCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
346 err := s.reconcileAppRunning(appCtx, app)
347 cancel()
348 if err != nil {
349 failures = append(failures, err)
350 }
351 }
352 return errors.Join(failures...)
353}
354
355// reconcileAppRunning recovers one app. A busy app (an in-flight user
356// mutation) is skipped, never waited on: the periodic pass must not
357// starve later apps or shutdown.
358func (s *Service) reconcileAppRunning(ctx context.Context, app string) error {
359 release, ok := s.tryAcquireAppContext(ctx, app)
360 if !ok {
361 s.log.Debug().Str("app", app).Msg("deployment: reconciliation skipped, app is busy")
362 return nil
363 }
364 defer release()
365
366 ctx = zerowrap.CtxWithFields(ctx, map[string]any{
367 zerowrap.FieldLayer: "usecase",
368 zerowrap.FieldUseCase: "ReconcileRunning",
369 "app": app,
370 })
371 intent, err := s.deps.State.LoadIntent(ctx, app)
372 if err != nil {
373 return fmt.Errorf("deployment: reconcile %q: load intent: %w", app, err)
374 }
375 active, ok, err := s.deps.State.LoadActive(ctx, app)
376 if err != nil {
377 return fmt.Errorf("deployment: reconcile %q: load active: %w", app, err)
378 }
379 if !ok || len(active.Services) == 0 {
380 s.pruneRecoveryHistory(app, active)
381 return nil
382 }
383 inhibitions, err := s.deps.State.LoadRecoveryInhibitions(ctx, app)
384 if err != nil {
385 return fmt.Errorf("deployment: reconcile %q: load recovery inhibitions: %w", app, err)
386 }
387 inhibited, pending := splitRecoveryInhibitions(app, inhibitions)
388 var failures []error
389 // verified records, per service, the execution this pass verified and
390 // routed. A successor that is not in it is not known to serve.
391 verified := map[string]time.Time{}
392 for _, name := range sortedServiceNames(active) {
393 eff := active.Services[name]
394 if intent.Stopped {
395 if err := s.convergeStoppedService(ctx, app, name, eff); err != nil {
396 failures = append(failures, err)
397 }
398 continue
399 }
400 startedAt, ok, err := s.reconcileRunningService(ctx, app, name, eff, inhibited)
401 if err != nil {
402 failures = append(failures, err)
403 }
404 if ok {
405 verified[name] = startedAt
406 }
407 }
408 if intent.Stopped {
409 // A stopped app must not stay published: rebuild the projection
410 // so no revoked backend remains routable. The publication is
411 // bounded like any other recovery effect.
412 publishCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
413 if err := s.refreshTraffic(publishCtx, app); err != nil {
414 failures = append(failures, err)
415 }
416 cancel()
417 }
418 // Superseded containers are retired last: only after their successors
419 // were verified and routed in this pass.
420 failures = append(failures, s.retirePendingContainers(ctx, app, intent.Stopped, active, pending, verified)...)
421 s.pruneRecoveryHistory(app, active)
422 return errors.Join(failures...)
423}
424
425// splitRecoveryInhibitions separates the generations recovery must never
426// start from the superseded containers waiting to be retired.
427func splitRecoveryInhibitions(app string, inhibitions []domain.AppRecoveryInhibition) (map[recoveryKey]struct{}, []domain.AppRecoveryInhibition) {
428 inhibited := map[recoveryKey]struct{}{}
429 var pending []domain.AppRecoveryInhibition
430 for _, inhibition := range inhibitions {
431 if inhibition.Reason == domain.AppInhibitRetirementPending {
432 pending = append(pending, inhibition)
433 continue
434 }
435 inhibited[recoveryKey{app: app, service: inhibition.Service, container: inhibition.ContainerID}] = struct{}{}
436 }
437 return inhibited, pending
438}
439
440// retirePendingContainers retires the superseded containers an overlap
441// replacement left running, with the stop grace recorded at the marker
442// (the app default for a record that has none). A container is retired only
443// when it can no longer be the serving one:
444// - ACTIVE names it (the replacement never took over): it is the serving
445// generation, so the marker is dropped and the container is kept;
446// - its successor was not verified and routed by THIS pass, or restarted
447// since: the superseded container is retried later, so a successor that
448// is not serving never costs the working generation.
449//
450// A stopped app has no successor to wait for: nothing may run. Cleanup
451// warnings of a retirement (claims not released, secret files) are logged and
452// returned, and the marker stays until the container is gone with its claims
453// released, so the next pass retries.
454func (s *Service) retirePendingContainers(ctx context.Context, app string, stopped bool, active domain.AppActive, pending []domain.AppRecoveryInhibition, verified map[string]time.Time) []error {
455 var failures []error
456 for _, marker := range pending {
457 if publishedContainer(active, marker.ContainerID) {
458 if err := s.clearRecoveryInhibition(ctx, app, marker.Service, marker.ContainerID); err != nil {
459 failures = append(failures, err)
460 }
461 continue
462 }
463 if !stopped && !s.successorVerified(ctx, app, marker.Service, active, verified) {
464 continue
465 }
466 retired := s.retireContainer(ctx, app, retireOptions{
467 Service: marker.Service, Grace: marker.StopGrace, ClearInhibition: true, RetryOnLeak: true,
468 }, marker.ContainerID)
469 if !retired.Settled {
470 failures = append(failures, fmt.Errorf("deployment: retirement pending for %s/%s: %s", app, marker.ContainerID, cleanupDetail(retired)))
471 continue
472 }
473 for _, warning := range retired.Warnings {
474 s.log.Warn().Str("app", app).Str("service", warning.Service).Str("container", warning.Leftover).
475 Str("detail", warning.Detail).Msg("deployment: superseded container retired with leftovers")
476 failures = append(failures, fmt.Errorf("deployment: retired superseded container %s/%s with leftovers: %s", app, warning.Leftover, warning.Detail))
477 }
478 }
479 return failures
480}
481
482// successorVerified reports whether the container ACTIVE names for the
483// service was verified and routed by this very pass, and is still the same
484// running execution: a final inspection that shows a different start time
485// means the successor restarted after it was verified, so it is unverified
486// again. A service ACTIVE no longer carries has no successor to wait for.
487func (s *Service) successorVerified(ctx context.Context, app, service string, active domain.AppActive, verified map[string]time.Time) bool {
488 successor, ok := active.Services[service]
489 if !ok || successor.Container == "" {
490 return true
491 }
492 startedAt, ok := verified[service]
493 if !ok || s.deps.Runtime == nil {
494 return false
495 }
496 container, err := s.deps.Runtime.InspectContainer(ctx, successor.Container)
497 return err == nil && container.Status == string(domain.ContainerStatusRunning) && container.StartedAt.Equal(startedAt)
498}
499
500// convergeStoppedService enforces durable stopped intent: it never
501// starts or publishes, and it stops a verified running exact ACTIVE ID
502// without deleting the container or its volumes.
503func (s *Service) convergeStoppedService(ctx context.Context, app, name string, eff domain.AppEffectiveService) error {
504 key := recoveryKey{app: app, service: name, container: eff.Container}
505 s.backoff.forget(key)
506 s.executions.forget(key)
507 if eff.Container == "" || s.deps.Runtime == nil {
508 return nil
509 }
510 serviceCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
511 defer cancel()
512 container, err := s.deps.Runtime.InspectContainer(serviceCtx, eff.Container)
513 if err != nil {
514 if errors.Is(err, domain.ErrContainerNotFound) {
515 return nil
516 }
517 return fmt.Errorf("deployment: stopped app %q service %q: inspect %s: %w", app, name, eff.Container, err)
518 }
519 if container.Status != string(domain.ContainerStatusRunning) && !isTransitionalStatus(container.Status) {
520 return nil
521 }
522 if err := s.deps.Runtime.StopContainer(serviceCtx, eff.Container, serviceStopGrace(eff)); err != nil && !errors.Is(err, domain.ErrContainerNotFound) {
523 return fmt.Errorf("deployment: stopped app %q service %q: stop %s: %w", app, name, eff.Container, err)
524 }
525 s.log.Info().Str("app", app).Str("service", name).Str("container", eff.Container).
526 Msg("deployment: stopped intent enforced on a revived container")
527 return nil
528}
529
530// reconcileRunningService recovers one service to the running intent
531// carried by ACTIVE. It only ever acts on the exact ACTIVE container ID.
532func (s *Service) reconcileRunningService(ctx context.Context, app, name string, eff domain.AppEffectiveService, inhibited map[recoveryKey]struct{}) (time.Time, bool, error) {
533 if eff.Container == "" {
534 // Absent ACTIVE: never act, never reconstruct.
535 return time.Time{}, false, nil
536 }
537 if _, blocked := inhibited[recoveryKey{app: app, service: name, container: eff.Container}]; blocked {
538 return time.Time{}, false, s.refuseInhibitedRecovery(ctx, app, name, eff.Container)
539 }
540 if s.deps.Runtime == nil {
541 return time.Time{}, false, nil
542 }
543 serviceCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
544 defer cancel()
545 key := recoveryKey{app: app, service: name, container: eff.Container}
546 container, fresh, err := s.inspectForRecovery(serviceCtx, app, name, eff, key)
547 if err != nil {
548 return time.Time{}, false, err
549 }
550 return s.convergeRecoveredService(serviceCtx, app, name, eff, container, key, fresh)
551}
552
553// inspectForRecovery retries a pending withdrawal, classifies the exact
554// ACTIVE container, and makes sure nothing unverifiable stays routable
555// before the workload is touched. A transient inspection error never
556// changes the workload or its traffic.
557func (s *Service) inspectForRecovery(ctx context.Context, app, name string, eff domain.AppEffectiveService, key recoveryKey) (*domain.Container, bool, error) {
558 // A previous withdrawal that could not be applied must be retried
559 // before anything else: never publish on top of stale forwarding.
560 withdrawn := false
561 if s.publication.inhibited(app, name) {
562 if err := s.withdrawForRecovery(ctx, app, name); err != nil {
563 return nil, false, err
564 }
565 withdrawn = true
566 }
567 container, err := s.deps.Runtime.InspectContainer(ctx, eff.Container)
568 if err != nil {
569 return nil, false, s.recoveryInspectError(ctx, app, name, eff, err)
570 }
571 // Never serve a backend that is not verifiably serving: a native
572 // restart (fresh execution) or a container that is not running must
573 // be withdrawn, and that withdrawal must succeed, before Gordon
574 // touches the workload. A transitional state keeps its traffic: the
575 // runtime is restarting it right now and it will be back within the
576 // pass cadence.
577 fresh := s.executions.isNewExecution(key, container.StartedAt)
578 unverified := container.Status != string(domain.ContainerStatusRunning) && !isTransitionalStatus(container.Status)
579 if (fresh || unverified) && !withdrawn {
580 if err := s.withdrawForRecovery(ctx, app, name); err != nil {
581 return nil, false, err
582 }
583 }
584 if fresh {
585 s.backoff.clearUnhealthy(key)
586 }
587 return container, fresh, nil
588}
589
590// convergeRecoveredService drives the exact ACTIVE container to a
591// verified, published state, or leaves it withdrawn with a reported
592// reason.
593func (s *Service) convergeRecoveredService(ctx context.Context, app, name string, eff domain.AppEffectiveService, container *domain.Container, key recoveryKey, fresh bool) (time.Time, bool, error) {
594 started, verify, err := s.driveRecoveredContainer(ctx, app, name, eff, container, key)
595 if err != nil {
596 return time.Time{}, false, errors.Join(s.withdrawForRecovery(ctx, app, name), err)
597 }
598 if !verify {
599 // Transitional or backoff-delayed: observe now, converge later.
600 s.backoff.invalidateStable(key)
601 return time.Time{}, false, nil
602 }
603 restarted, stable, err := s.recoverHealth(ctx, app, name, eff, key)
604 if err != nil {
605 s.backoff.invalidateStable(key)
606 return time.Time{}, false, errors.Join(s.withdrawForRecovery(ctx, app, name), err)
607 }
608 if err := s.publishRecoveredService(ctx, app, name, eff, container, key, started, restarted, fresh); err != nil {
609 return time.Time{}, false, err
610 }
611 if stable {
612 s.backoff.observeStable(key)
613 } else {
614 s.backoff.invalidateStable(key)
615 }
616 // Verified and published: report the execution this pass vouches for.
617 return container.StartedAt, true, nil
618}
619
620// refuseInhibitedRecovery reports a generation whose recovery is durably
621// inhibited and withdraws its backend best-effort: it must not stay
622// routable, and it must never be started.
623func (s *Service) refuseInhibitedRecovery(ctx context.Context, app, name, containerID string) error {
624 withdrawCtx, cancel := context.WithTimeout(ctx, reconcileServiceDeadline)
625 defer cancel()
626 return errors.Join(
627 fmt.Errorf(
628 "deployment: app %q service %q container %s has recovery inhibited; no start attempted",
629 app, name, containerID,
630 ),
631 s.withdrawForRecovery(withdrawCtx, app, name),
632 )
633}
634
635// recoveryInspectError classifies one failed inspection: a confirmed
636// missing container withdraws traffic and reports without reconstruction,
637// while a transient error leaves the workload and its traffic untouched.
638func (s *Service) recoveryInspectError(ctx context.Context, app, name string, eff domain.AppEffectiveService, err error) error {
639 if !errors.Is(err, domain.ErrContainerNotFound) {
640 return fmt.Errorf("deployment: app %q service %q inspect %s: %w", app, name, eff.Container, err)
641 }
642 // The container is confirmed gone: its loopback reservations must not
643 // outlive it, or a later workload that reuses the port would collide
644 // with a claim no container holds.
645 releaseErr := s.deps.State.ReleaseBackendBinds(ctx, app, eff.Container)
646 if releaseErr != nil {
647 releaseErr = fmt.Errorf("deployment: release claims of gone container %s: %w", eff.Container, releaseErr)
648 }
649 return errors.Join(
650 s.withdrawForRecovery(ctx, app, name),
651 releaseErr,
652 fmt.Errorf("deployment: app %q service %q container %s is gone; traffic withdrawn, no reconstruction: %w",
653 app, name, eff.Container, domain.ErrContainerNotFound),
654 )
655}
656
657// driveRecoveredContainer brings the exact ACTIVE container to a running
658// state: an already running container is left alone, a transitional one
659// is observed and retried later, an unsafe state errors non-destructively,
660// and a stopped one is started under the generation's backoff budget.
661// verify reports whether the caller may continue to binds, readiness, and
662// publication: a transitional or delayed generation must not be published
663// on this pass.
664func (s *Service) driveRecoveredContainer(ctx context.Context, app, name string, eff domain.AppEffectiveService, container *domain.Container, key recoveryKey) (started bool, verify bool, err error) {
665 switch {
666 case container.Status == string(domain.ContainerStatusRunning):
667 return false, true, nil
668 case isTransitionalStatus(container.Status):
669 // restarting/starting: observe and retry on a later pass.
670 return false, false, nil
671 case container.Status == string(domain.ContainerStatusPaused) || container.Status == string(domain.ContainerStatusUnknown) || container.Status == "":
672 return false, false, fmt.Errorf("deployment: app %q service %q container %s is %q; refusing non-destructive recovery",
673 app, name, eff.Container, container.Status)
674 default:
675 // created/exited/dead/stopped with running intent: start the
676 // exact ACTIVE ID regardless of exit code.
677 if !s.backoff.allow(key) {
678 return false, false, nil
679 }
680 // A generation found down again before a previous recovery was
681 // ever confirmed stable is crash-looping: charge it, so a
682 // workload that dies after every successful start cannot be
683 // restarted on every pass forever.
684 if s.backoff.attemptPending(key) {
685 s.backoff.recordFailure(key)
686 if !s.backoff.allow(key) {
687 return false, false, nil
688 }
689 }
690 started, err := s.startRecovered(ctx, app, name, eff, key)
691 return started, err == nil, err
692 }
693}
694
695// publishRecoveredService verifies one generation in the order the plan
696// requires: read the container's live binds, probe readiness against
697// them, revalidate execution identity and ACTIVE, persist the verified
698// binds, and only then publish. Binds are never persisted to ACTIVE
699// before readiness, so no other publication can project an unverified
700// backend while this pass waits for the workload. The execution is
701// recorded only on success, so a failed verification is retried on the
702// next pass instead of being silently accepted.
703func (s *Service) publishRecoveredService(ctx context.Context, app, name string, eff domain.AppEffectiveService, observed *domain.Container, key recoveryKey, started, restarted, fresh bool) error {
704 binds, udpBinds, err := s.inspectBackendBinds(ctx, app, name, eff.Container, backendPorts(eff.Spec))
705 if err != nil {
706 s.backoff.recordFailure(key)
707 return errors.Join(s.withdrawForRecovery(ctx, app, name), err)
708 }
709 if started || restarted || fresh {
710 if err := s.waitServiceReady(ctx, app, eff.Container, eff.Spec, binds); err != nil {
711 s.backoff.recordFailure(key)
712 return errors.Join(s.withdrawForRecovery(ctx, app, name), err)
713 }
714 }
715 verifiedBinds, verifiedUDPBinds, err := s.revalidateAndPersistRecoveredBinds(ctx, app, name, eff, observed, binds, udpBinds)
716 if err != nil {
717 s.backoff.recordFailure(key)
718 return errors.Join(s.withdrawForRecovery(ctx, app, name), err)
719 }
720 if started || restarted || fresh || !equalBinds(verifiedBinds, eff.BackendBinds) || !equalBinds(verifiedUDPBinds, eff.UDPBackendBinds) {
721 if err := s.refreshTraffic(ctx, app); err != nil {
722 return err
723 }
724 }
725 s.executions.record(key, observed.StartedAt)
726 return nil
727}
728
729// startRecovered starts the exact ACTIVE container. A failed start is
730// re-inspected: if native restart policy already revived it, no failure
731// is charged and no second restart is issued.
732func (s *Service) startRecovered(serviceCtx context.Context, app, name string, eff domain.AppEffectiveService, key recoveryKey) (bool, error) {
733 // Fail closed before mutating the runtime if a bind's policy was
734 // revoked or became invalid since ACTIVE was published.
735 if _, err := s.resolveServiceBinds(app, eff.Spec); err != nil {
736 return false, err
737 }
738 // Revoked device grants fail before any runtime mutation.
739 if _, err := s.resolveServiceDevices(app, eff.Spec); err != nil {
740 return false, err
741 }
742 if err := s.deps.Runtime.StartContainer(serviceCtx, eff.Container); err != nil {
743 if errors.Is(err, domain.ErrContainerNotFound) {
744 s.backoff.forget(key)
745 return false, errors.Join(
746 fmt.Errorf("deployment: app %q service %q container %s is gone: %w", app, name, eff.Container, domain.ErrContainerNotFound),
747 )
748 }
749 recovered, inspectErr := s.deps.Runtime.InspectContainer(serviceCtx, eff.Container)
750 if inspectErr == nil && (recovered.Status == string(domain.ContainerStatusRunning) || isTransitionalStatus(recovered.Status)) {
751 s.log.Info().Str("app", app).Str("service", name).Str("container", eff.Container).
752 Msg("deployment: native restart won the race, no recovery charged")
753 s.backoff.clearUnhealthy(key)
754 return true, nil
755 }
756 s.backoff.recordFailure(key)
757 return false, fmt.Errorf("deployment: app %q service %q start %s: %w", app, name, eff.Container, err)
758 }
759 s.backoff.markAttempt(key)
760 s.log.Info().Str("app", app).Str("service", name).Str("container", eff.Container).
761 Msg("deployment: recovery started the exact active container")
762 return true, nil
763}
764
765// recoverHealth restarts a workload that reported `unhealthy` on
766// consecutive passes. `starting` waits; a missing healthcheck never
767// triggers a restart. Restarts share the generation's backoff budget and
768// are marked as unconfirmed attempts. The second result reports whether
769// the observation counts as stable ("running, ready, published"): an
770// unhealthy, still-starting, or unreadable health state must not extend
771// the five-minute stability window.
772func (s *Service) recoverHealth(serviceCtx context.Context, app, name string, eff domain.AppEffectiveService, key recoveryKey) (restarted bool, stable bool, err error) {
773 status, hasHealthcheck, statusErr := s.deps.Runtime.GetContainerHealthStatus(serviceCtx, eff.Container)
774 if statusErr != nil || !hasHealthcheck {
775 // A transient health error is not an unhealthy workload, but it
776 // is also not a proven healthy one.
777 return false, statusErr == nil, nil
778 }
779 if status != "unhealthy" {
780 s.backoff.clearUnhealthy(key)
781 return false, status != "starting", nil
782 }
783 if s.backoff.observeUnhealthy(key) < unhealthyThreshold {
784 return false, false, nil
785 }
786 if !s.backoff.allow(key) {
787 return false, false, nil
788 }
789 // Fail closed before mutating the runtime if a bind's policy was
790 // revoked or became invalid since ACTIVE was published.
791 if _, err := s.resolveServiceBinds(app, eff.Spec); err != nil {
792 return false, false, err
793 }
794 // Revoked device grants fail before any runtime mutation.
795 if _, err := s.resolveServiceDevices(app, eff.Spec); err != nil {
796 return false, false, err
797 }
798 if err := s.deps.Runtime.RestartContainer(serviceCtx, eff.Container, serviceStopGrace(eff)); err != nil {
799 s.backoff.recordFailure(key)
800 return false, false, fmt.Errorf("deployment: app %q service %q restart unhealthy %s: %w", app, name, eff.Container, err)
801 }
802 s.backoff.markAttempt(key)
803 s.backoff.clearUnhealthy(key)
804 s.log.Info().Str("app", app).Str("service", name).Str("container", eff.Container).
805 Msg("deployment: restarted an unhealthy container")
806 return true, false, nil
807}
808
809// revalidateAndPersistRecoveredBinds revalidates ACTIVE and the observed
810// execution immediately before publication, then persists the verified
811// binds (already re-read from the exact container) into the ACTIVE
812// record. Stale evidence never publishes: a native restart that happened
813// while this pass waited for readiness changes the observed start and
814// aborts the pass instead of publishing the previous execution's binds.
815func (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) {
816 current, err := s.deps.Runtime.InspectContainer(ctx, eff.Container)
817 if err != nil {
818 return nil, nil, fmt.Errorf("deployment: re-inspect %s before publication: %w", eff.Container, err)
819 }
820 if observed != nil && !current.StartedAt.Equal(observed.StartedAt) {
821 return nil, nil, fmt.Errorf("deployment: app %q service %q container %s restarted during recovery: %w",
822 app, name, eff.Container, domain.ErrAppStateConflict)
823 }
824 active, ok, err := s.deps.State.LoadActive(ctx, app)
825 if err != nil {
826 return nil, nil, fmt.Errorf("deployment: reload active for recovery: %w", err)
827 }
828 if !ok {
829 return nil, nil, fmt.Errorf("deployment: app %q has no active state: %w", app, domain.ErrAppStateConflict)
830 }
831 recorded, ok := active.Services[name]
832 if !ok || recorded.Container != eff.Container || recorded.EffectiveRevision != eff.EffectiveRevision {
833 return nil, nil, fmt.Errorf("deployment: app %q service %q generation changed during recovery: %w", app, name, domain.ErrAppStateConflict)
834 }
835 if !equalBinds(binds, recorded.BackendBinds) || !equalBinds(udpBinds, recorded.UDPBackendBinds) {
836 recorded.BackendBinds = binds
837 recorded.UDPBackendBinds = udpBinds
838 active.Services[name] = recorded
839 if err := s.deps.State.SaveActive(ctx, active); err != nil {
840 return nil, nil, fmt.Errorf("deployment: persist recovered binds: %w", err)
841 }
842 }
843 return binds, udpBinds, nil
844}
845
846// withdrawForRecovery applies one serialized fail-closed withdrawal of
847// a service. A failed application retains an in-memory publication
848// inhibition and never releases backend claims.
849func (s *Service) withdrawForRecovery(ctx context.Context, app, service string) error {
850 if s.deps.Traffic == nil {
851 return nil
852 }
853 if err := s.deps.Traffic.WithdrawService(ctx, app, service); err != nil {
854 s.publication.mark(app, service)
855 return fmt.Errorf("deployment: withdraw %s/%s: %w", app, service, err)
856 }
857 s.publication.clear(app, service)
858 return nil
859}
860
861// recoveryInhibited reports whether one exact generation is durably
862// inhibited for recovery. A retirement-pending marker does not count: it
863// names a superseded container that is being retired, not a generation
864// recovery must keep stopped, and when ACTIVE still names that container
865// (the replacement never took over) it must stay startable.
866func (s *Service) recoveryInhibited(ctx context.Context, app, service, containerID string) (bool, error) {
867 if containerID == "" {
868 return false, nil
869 }
870 inhibitions, err := s.deps.State.LoadRecoveryInhibitions(ctx, app)
871 if err != nil {
872 return false, fmt.Errorf("deployment: load recovery inhibitions: %w", err)
873 }
874 for _, inhibition := range inhibitions {
875 if inhibition.Service == service && inhibition.ContainerID == containerID &&
876 inhibition.Reason != domain.AppInhibitRetirementPending {
877 return true, nil
878 }
879 }
880 return false, nil
881}
882
883// inhibitRecovery durably records a generation-scoped recovery
884// inhibition for one exact container ID.
885func (s *Service) inhibitRecovery(ctx context.Context, app, service, containerID, reason, operation string) error {
886 if containerID == "" {
887 return nil
888 }
889 inhibition := domain.AppRecoveryInhibition{
890 App: app,
891 Service: service,
892 ContainerID: containerID,
893 Reason: reason,
894 Operation: operation,
895 CreatedAt: time.Now().UTC(),
896 }
897 if err := s.deps.State.SaveRecoveryInhibition(ctx, inhibition); err != nil {
898 return fmt.Errorf("deployment: record recovery inhibition for %s/%s: %w", app, containerID, err)
899 }
900 return nil
901}
902
903// inhibitRetirement durably marks a superseded container that an overlap
904// replacement left running beside its successor. The periodic reconciliation
905// retires it with grace, once the successor is verified and routed.
906func (s *Service) inhibitRetirement(ctx context.Context, app, service, containerID, operation string, grace time.Duration) error {
907 if containerID == "" {
908 return nil
909 }
910 inhibition := domain.AppRecoveryInhibition{
911 App: app,
912 Service: service,
913 ContainerID: containerID,
914 Reason: domain.AppInhibitRetirementPending,
915 Operation: operation,
916 CreatedAt: time.Now().UTC(),
917 StopGrace: grace,
918 }
919 if err := s.deps.State.SaveRecoveryInhibition(ctx, inhibition); err != nil {
920 return fmt.Errorf("deployment: record retirement of %s/%s: %w", app, containerID, err)
921 }
922 return nil
923}
924
925// clearRecoveryInhibition drops the inhibition of a safely superseded
926// generation.
927func (s *Service) clearRecoveryInhibition(ctx context.Context, app, service, containerID string) error {
928 if containerID == "" {
929 return nil
930 }
931 if err := s.deps.State.ClearRecoveryInhibition(ctx, app, service, containerID); err != nil {
932 return fmt.Errorf("deployment: clear recovery inhibition for %s/%s: %w", app, containerID, err)
933 }
934 return nil
935}
936
937// pruneRecoveryHistory drops in-memory state of generations that are no
938// longer present in ACTIVE. Durable inhibition records are never
939// dropped here: only an explicit supersede or removal clears them.
940func (s *Service) pruneRecoveryHistory(app string, active domain.AppActive) {
941 live := map[recoveryKey]struct{}{}
942 liveServices := map[string]struct{}{}
943 for name, svc := range active.Services {
944 live[recoveryKey{app: app, service: name, container: svc.Container}] = struct{}{}
945 liveServices[name] = struct{}{}
946 }
947 s.backoff.mu.Lock()
948 for key := range s.backoff.entries {
949 if key.app != app {
950 continue
951 }
952 if _, ok := live[key]; !ok {
953 delete(s.backoff.entries, key)
954 }
955 }
956 s.backoff.mu.Unlock()
957 s.executions.mu.Lock()
958 for key := range s.executions.seen {
959 if key.app != app {
960 continue
961 }
962 if _, ok := live[key]; !ok {
963 delete(s.executions.seen, key)
964 }
965 }
966 s.executions.mu.Unlock()
967 s.publication.retainAppServices(app, liveServices)
968}
969
970// isTransitionalStatus reports runtime states that must be observed and
971// retried rather than acted on.
972func isTransitionalStatus(status string) bool {
973 return status == "restarting" || status == "starting"
974}