1package deployment
2
3import (
4 "context"
5 "fmt"
6
7 "github.com/bnema/zerowrap"
8
9 "github.com/bnema/gordon/internal/domain"
10)
11
12// reconcileInterruptedDeploy converges the journal of a deployment that a
13// crash, shutdown, or failed cleanup interrupted. A candidate that was
14// created but never published is removed and its loopback claims released, so
15// recovery can rebuild the published generation without ever running two
16// generations of one service. The interrupted operation is then finalized as
17// failed: a repeat of its key replays an explicit failure instead of a
18// permanent in-flight claim.
19//
20// It also retries the candidates of a failed operation whose cleanup did not
21// complete, so a leftover stays visible and is eventually removed. The
22// published generation recorded in ACTIVE is never touched.
23//
24// It runs at boot and before every mutation of the app, always under the app
25// lock, so a non-terminal journal is definitively interrupted, and a mutation
26// that cannot reconcile a predecessor does not proceed. It reads the
27// operation journal first and only then ACTIVE, so the common case of a
28// successful operation costs one state read.
29//
30// Reconciliation fails closed on a candidate that cannot be removed: an
31// orphan that may still run aborts the caller instead of being journaled as a
32// recoverable leftover. Because the journal is never finalized while such a
33// leftover exists, it stays the latest operation and a newer operation can
34// never be created on top of it and mask it.
35func (s *Service) reconcileInterruptedDeploy(ctx context.Context, app string) error {
36 op, ok, err := s.deps.State.LoadLatestOperation(ctx, app)
37 if err != nil {
38 return fmt.Errorf("deployment: load interrupted operation for %q: %w", app, err)
39 }
40 if !ok {
41 return nil
42 }
43 if !op.Terminal() && s.operationLive(app, op.Op) {
44 // The operation is owned by a live goroutine in this process that
45 // may still be persisting its outcome. Reconcile it only after that
46 // owner has exited; until then the store record is authoritative and
47 // must not be finalized here.
48 return nil
49 }
50 terminal := op.Terminal()
51 needsActive := false
52 for _, step := range op.Steps {
53 if reconcileNeedsContainer(step, terminal) {
54 needsActive = true
55 break
56 }
57 }
58 var active domain.AppActive
59 if needsActive {
60 active, _, err = s.deps.State.LoadActive(ctx, app)
61 if err != nil {
62 return fmt.Errorf("deployment: load active for interrupted operation: %w", err)
63 }
64 }
65 log := zerowrap.FromCtx(ctx)
66 // write is true when the journal itself changes: an interrupted operation
67 // is always finalized, a terminal one only when a leftover must be
68 // reported.
69 changed, err := s.convergeInterruptedSteps(ctx, app, &op, active, terminal, log)
70 if err != nil {
71 return err
72 }
73 write := !terminal || changed
74 if !write {
75 return nil
76 }
77 if !terminal {
78 // The effects ran and only the outcome write was interrupted: that is
79 // a success, not a failure.
80 op.Outcome = interruptedOutcome(op)
81 }
82 if err := s.deps.State.SaveOperation(ctx, op); err != nil {
83 return fmt.Errorf("deployment: finalize interrupted operation for %q: %w", app, err)
84 }
85 log.Warn().Str("app", app).Str("op", op.Op).Msg("deployment: reconciled an unfinished operation")
86 return nil
87}
88
89// reconcileNeedsContainer reports whether one step still needs container
90// work: a step that recorded a candidate. A terminal operation only retries
91// the step that recorded a failed candidate; a pending step exists only in an
92// interrupted operation.
93func reconcileNeedsContainer(step domain.AppOperationStep, terminal bool) bool {
94 if step.After == "" {
95 return false
96 }
97 if terminal {
98 return step.State == domain.AppStepFailed
99 }
100 return true
101}
102
103// convergeInterruptedSteps retries the recorded candidate of every step that
104// needs container work and accumulates the bounded leftovers in the journal.
105// It reports whether the journal changed and fails closed on the first
106// candidate that cannot be removed, before any later step or mutation runs.
107func (s *Service) convergeInterruptedSteps(ctx context.Context, app string, op *domain.AppOperation, active domain.AppActive, terminal bool, log zerowrap.Logger) (bool, error) {
108 changed := false
109 for i := range op.Steps {
110 step := &op.Steps[i]
111 if !reconcileNeedsContainer(*step, terminal) {
112 continue
113 }
114 warnings, convErr := s.convergeCandidate(ctx, app, active, step, terminal, log)
115 if step.After == "" {
116 // Converging the candidate clears its recorded After, so the
117 // journal changed even when nothing was left over to report: the
118 // converged step must be persisted, or every later pass reconciles
119 // it again.
120 changed = true
121 }
122 if len(warnings) > 0 {
123 // A candidate that will not go away is operator-visible instead of
124 // staying untracked.
125 op.Warnings = append(op.Warnings, journalWarnings(warnings)...)
126 changed = true
127 }
128 if convErr != nil {
129 // Fail closed: the orphan may still be running, so no later
130 // mutation may proceed. The journal keeps the leftover (and the
131 // failed step) for the next pass; a terminal operation stays the
132 // latest record because no newer claim is opened.
133 if saveErr := s.deps.State.SaveOperation(ctx, *op); saveErr != nil {
134 log.Warn().Err(saveErr).Msg("deployment: failed to record reconciliation failure")
135 }
136 return changed, fmt.Errorf("deployment: reconcile incomplete operation for %q: %w", app, convErr)
137 }
138 }
139 return changed, nil
140}
141
142// convergeCandidate converges the container one step recorded. The published
143// generation is kept; a never-published candidate is removed so recovery can
144// rebuild the generation recorded in ACTIVE without two generations running
145// at once. An interrupted operation's step is marked failed.
146//
147// A candidate that cannot be removed is returned as an error, not a
148// recoverable warning: the orphan may still run, so every later mutation must
149// fail closed instead of proceeding. The returned warnings are the leftovers
150// of that removal.
151//
152// Success clears the step's recorded After and keeps the removed (or kept)
153// container id in Detail, so the audit trail survives while the converged step
154// is no longer retried on every later pass. Only the failure paths leave After
155// in place, keeping a failed convergence eligible for retry.
156//
157// Once the candidate is converged, the inhibition the replacement wrote for
158// the superseded container is cleared. A sequential replacement confirmed the
159// superseded container gone before the candidate was created (deployService
160// retires it first), so its marker protects nothing and would otherwise
161// refuse the boot/start/restart rebuild from ACTIVE. An overlap replacement
162// leaves its superseded container running: when the candidate is published,
163// its retirement-pending marker stays, and the periodic reconciliation
164// retires that container once the candidate is verified and routed. When the
165// candidate was never published the superseded container still serves, so
166// the marker is dropped and the container is kept.
167//
168// Reconciliation never touches traffic: the persisted binds of a published
169// candidate are unverified here, and the verified start/recovery path
170// publishes them.
171func (s *Service) convergeCandidate(ctx context.Context, app string, active domain.AppActive, step *domain.AppOperationStep, terminal bool, log zerowrap.Logger) ([]CleanupWarning, error) {
172 candidate := step.After
173 if publishedContainer(active, candidate) {
174 // The published generation: keep it. A step that already succeeded is
175 // left as recorded; a routing apply that never ran is republished by
176 // the periodic recovery pass.
177 if !terminal && step.State != domain.AppStepSucceeded {
178 step.State = domain.AppStepFailed
179 step.Error = "interrupted before traffic publication; the published container is kept and routing is republished"
180 }
181 step.Detail = "converged: published container " + candidate + " kept"
182 } else {
183 retired := s.retireContainer(ctx, app, retireOptions{Service: step.Service, Force: true}, candidate)
184 if !retired.Gone {
185 if !terminal {
186 step.State = domain.AppStepFailed
187 step.Error = "interrupted before publication; the unpublished candidate could not be removed: " + cleanupDetail(retired)
188 log.Warn().Str("app", app).Str("container", candidate).Msg("deployment: unpublished candidate could not be removed")
189 }
190 return retired.Warnings, fmt.Errorf("deployment: candidate %s of service %q could not be removed: %s", candidate, step.Service, cleanupDetail(retired))
191 }
192 if !terminal {
193 step.State = domain.AppStepFailed
194 step.Error = "interrupted before publication; the unpublished candidate was removed"
195 }
196 step.Detail = "converged: removed unpublished candidate " + candidate
197 }
198 // The candidate is gone (or published): the replacement-pending
199 // inhibition of the superseded container is stale. Dropping it lets
200 // boot/start/restart rebuild that generation from ACTIVE; the single
201 // writer is preserved because the superseded container was proven gone
202 // before the candidate existed.
203 if step.Service != "" && step.Before != "" && step.Before != candidate {
204 if err := s.clearSupersededInhibition(ctx, app, step.Service, step.Before, publishedContainer(active, candidate)); err != nil {
205 log.Warn().Err(err).Str("app", app).Str("container", step.Before).Msg("deployment: clear stale replacement inhibition")
206 return []CleanupWarning{{Service: step.Service, Leftover: step.Before, Detail: "clear recovery inhibition: " + err.Error()}}, nil
207 }
208 }
209 // Only reconciliation removes a recorded candidate: clearing After here is
210 // what marks the step converged.
211 step.After = ""
212 return nil, nil
213}
214
215// clearSupersededInhibition drops the inhibition the interrupted replacement
216// recorded for the superseded container. A retirement-pending marker of a
217// published candidate is kept: the superseded container of an overlap
218// replacement may still be running, and that marker is what makes the
219// periodic reconciliation retire it. Every other marker is stale.
220func (s *Service) clearSupersededInhibition(ctx context.Context, app, service, before string, published bool) error {
221 if published {
222 inhibitions, err := s.deps.State.LoadRecoveryInhibitions(ctx, app)
223 if err != nil {
224 return fmt.Errorf("deployment: load recovery inhibitions: %w", err)
225 }
226 for _, inhibition := range inhibitions {
227 if inhibition.Service == service && inhibition.ContainerID == before && inhibition.Reason == domain.AppInhibitRetirementPending {
228 return nil
229 }
230 }
231 }
232 return s.clearRecoveryInhibition(ctx, app, service, before)
233}
234
235// interruptedOutcome classifies an interrupted operation: a failure, unless
236// every step had already succeeded.
237func interruptedOutcome(op domain.AppOperation) string {
238 if operationSucceeded(op) {
239 return domain.AppOutcomeSuccess
240 }
241 return domain.AppOutcomeFailed
242}
243
244// operationSucceeded reports whether every step of an interrupted operation
245// reached success: its effects ran and only the final outcome write was
246// interrupted.
247func operationSucceeded(op domain.AppOperation) bool {
248 if len(op.Steps) == 0 {
249 return false
250 }
251 for _, step := range op.Steps {
252 if step.State != domain.AppStepSucceeded {
253 return false
254 }
255 }
256 return true
257}
258
259// publishedContainer reports whether any service of the ACTIVE record names
260// the container. A container that is the published generation is never a
261// leftover, whichever service recorded it.
262func publishedContainer(active domain.AppActive, containerID string) bool {
263 if containerID == "" {
264 return false
265 }
266 for _, service := range active.Services {
267 if service.Container == containerID {
268 return true
269 }
270 }
271 return false
272}