amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package hostops
2
3import (
4 "context"
5 "crypto/ed25519"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "math"
10 "os"
11 "slices"
12 "strconv"
13 "strings"
14 "time"
15
16 "github.com/carlosframework/platform/internal/controlname"
17 "github.com/carlosframework/platform/internal/hostclaim"
18 "github.com/carlosframework/platform/internal/release"
19)
20
21// ErrControlKey is the data boundary's unconditional refusal of a canonical
22// control object. It wraps ErrContract so existing callers treat it as an
23// unavailable contract, never as pending work that may succeed on retry.
24var ErrControlKey = fmt.Errorf("%w: canonical control object is writable only by the typed control writer", ErrContract)
25
26const (
27 MailGateKey = "control/mail-resource-gate.json"
28 registrationPurge = "purge"
29)
30
31// IsControlOnlyKey names every object the data boundary refuses to mutate,
32// whatever the context carries. Malformed app shapes are refused by the
33// classifier, not here, so this stays an exact-key predicate.
34func IsControlOnlyKey(key string) bool {
35 if key == MailGateKey || (strings.HasPrefix(key, WakeResetPrefix) || strings.HasPrefix(key, WakeResetArchivePrefix)) {
36 return true
37 }
38 p := strings.Split(key, "/")
39 if len(p) != 4 || p[0] != "apps" || !controlname.ValidAccount(p[1]) || !controlname.ValidApp(p[2]) {
40 return false
41 }
42 switch p[3] {
43 case "admission.json", "purge.json", "purge-request.json":
44 return true
45 }
46 return false
47}
48
49// ControlWriter is the admission engine's private writer. Its raw store is
50// a named unexported field; there is no accessor, no generic Put/Update and
51// no mutate callback. Every exported method is one typed transition that
52// validates its own precondition against a fresh read.
53type ControlWriter struct {
54 homeRegion string
55 raw release.Store
56 pub ed25519.PublicKey
57 now func() time.Time
58}
59
60// NewControlWriter is called once, by ConfigureClaimRuntime. A source census
61// (TestClaimConsoleMutationInvariant step 5) rejects any other caller.
62func NewControlWriter(raw release.Store, pub ed25519.PublicKey, now func() time.Time, homeRegion ...string) (*ControlWriter, error) {
63 if raw == nil || len(pub) != ed25519.PublicKeySize || now == nil {
64 return nil, ErrContract
65 }
66 home, err := consoleHomeRegion(homeRegion)
67 if err != nil {
68 return nil, err
69 }
70 return &ControlWriter{raw: raw, pub: pub, now: now, homeRegion: home}, nil
71}
72
73func (w *ControlWriter) ready() error {
74 if w == nil || w.raw == nil || w.now == nil {
75 return ErrContract
76 }
77 return nil
78}
79
80// updateAdmission is the single CAS primitive. Lower-case and receiver-only:
81// it is reachable solely from the typed methods in this package.
82func (w *ControlWriter) updateAdmission(ctx context.Context, before AdmissionObserved, mutate func(*AppAdmission) error) (AdmissionObserved, error) {
83 if err := w.ready(); err != nil {
84 return AdmissionObserved{}, err
85 }
86 next, err := nextAdmission(before, mutate)
87 if err != nil {
88 return AdmissionObserved{}, err
89 }
90 body, err := json.Marshal(next)
91 if err != nil {
92 return AdmissionObserved{}, err
93 }
94 etag, err := w.raw.Put(ctx, before.Key, body, before.ETag)
95 if err != nil {
96 return AdmissionObserved{}, err
97 }
98 return AdmissionObserved{Key: before.Key, ETag: etag, Value: next}, nil
99}
100
101// casRetries bounds retryAdmission. Registrations renew every 20s and each
102// effect takes two CASes, so a lost race is routine; eight fresh attempts
103// cover every realistic burst without turning a wedge into a spin.
104const casRetries = 8
105
106// retryAdmission re-reads and re-applies mutate after a lost CAS. Use it only
107// for transitions whose mutate re-validates everything it depends on and is
108// idempotent against its own earlier success (settle, renew, bind, phase,
109// reopen, restore receipt, drain removal). Initial transitions that choose a
110// fresh identity or bind a saved claim observation (RegisterStart,
111// BeginExclusive, ReservePurge, the generation-only claim CAS) stay single
112// CAS: their caller decides what a conflict means. before may be zero, in
113// which case the first attempt reads.
114func (w *ControlWriter) retryAdmission(ctx context.Context, account, app, appID string, before AdmissionObserved, mutate func(*AppAdmission) error) (AdmissionObserved, error) {
115 var err error
116 if before.ETag == "" {
117 if before, err = w.read(ctx, account, app, appID); err != nil {
118 return AdmissionObserved{}, err
119 }
120 }
121 for attempt := 1; ; attempt++ {
122 next, err := w.updateAdmission(ctx, before, mutate)
123 if err == nil || !errors.Is(err, release.ErrCASConflict) || attempt >= casRetries {
124 return next, err
125 }
126 select {
127 case <-ctx.Done():
128 return AdmissionObserved{}, errors.Join(err, ctx.Err())
129 case <-time.After(time.Duration(attempt) * 15 * time.Millisecond):
130 }
131 if before, err = w.read(ctx, account, app, appID); err != nil {
132 return AdmissionObserved{}, err
133 }
134 }
135}
136
137func (w *ControlWriter) read(ctx context.Context, account, app, appID string) (AdmissionObserved, error) {
138 if err := w.ready(); err != nil {
139 return AdmissionObserved{}, err
140 }
141 return ReadAdmission(ctx, w.raw, account, app, appID)
142}
143
144func (w *ControlWriter) purgeAbsent(ctx context.Context, account, app string) error {
145 key, err := PurgeKey(account, app)
146 if err != nil {
147 return err
148 }
149 if _, _, err = w.raw.Get(ctx, key); errors.Is(err, os.ErrNotExist) {
150 return nil
151 } else if err != nil {
152 return err
153 }
154 return ErrPending
155}
156
157// ---- registrations --------------------------------------------------------
158
159// StartHandle is the in-memory capability for one registration this process
160// created. Knowing a registration's owner tuple from the bucket is not
161// enough to issue effects under it; the handle is.
162type StartHandle struct {
163 w *ControlWriter
164 moveStart moveStartSlot
165 account, app, appID, id string
166 owner RegistrationOwner
167 epoch uint64
168 // kind and incarnation are the registration's: a reset-wake-retry
169 // grant is bound to them (review R2).
170 kind, incarnation string
171}
172
173func (h *StartHandle) ID() string { return h.id }
174func (h *StartHandle) AppID() string { return h.appID }
175func (h *StartHandle) Epoch() uint64 { return h.epoch }
176func (h *StartHandle) Owner() RegistrationOwner { return h.owner }
177
178func (w *ControlWriter) RegisterStart(ctx context.Context, account, app, appID, id string, r Registration, now time.Time) (*StartHandle, error) {
179 if r.Kind == registrationPurge || r.Kind == RegistrationIngest {
180 return nil, ErrIdentity
181 }
182 if err := w.purgeAbsent(ctx, account, app); err != nil {
183 return nil, err
184 }
185 before, err := w.read(ctx, account, app, appID)
186 if err != nil {
187 return nil, err
188 }
189 if r.AppID != appID || r.Epoch != before.Value.Epoch {
190 return nil, ErrIdentity
191 }
192 if _, err = w.updateAdmission(ctx, before, func(a *AppAdmission) error { return registerStart(a, id, r, now) }); err != nil {
193 return nil, err
194 }
195 return &StartHandle{w: w, account: account, app: app, appID: appID, id: id, owner: r.Owner, epoch: r.Epoch, kind: r.Kind, incarnation: r.Incarnation}, nil
196}
197
198// Bind and Renew retry a lost CAS: bindStart is idempotent for the same
199// (created, incarnation) and both re-check owner, epoch and liveness.
200func (h *StartHandle) Bind(ctx context.Context, created time.Time, incarnation string, now time.Time) error {
201 _, err := h.w.retryAdmission(ctx, h.account, h.app, h.appID, AdmissionObserved{}, func(a *AppAdmission) error {
202 return bindStart(a, h.id, h.owner, created, incarnation, now)
203 })
204 return err
205}
206
207func (h *StartHandle) Renew(ctx context.Context, now time.Time) error {
208 _, err := h.w.retryAdmission(ctx, h.account, h.app, h.appID, AdmissionObserved{}, func(a *AppAdmission) error {
209 return renewStart(a, h.id, h.owner, now)
210 })
211 return err
212}
213
214type startHandleKey struct{}
215
216// WithStartHandle carries the request's own registration capability to
217// ProductionAuthority.RecordEffects. Only a writer-minted handle exists.
218func WithStartHandle(ctx context.Context, h *StartHandle) context.Context {
219 return context.WithValue(ctx, startHandleKey{}, h)
220}
221
222// startHandleFor returns the context's handle only for the exact
223// registration and incarnation asked for.
224func startHandleFor(ctx context.Context, account, app, appID, id string) (*StartHandle, bool) {
225 h, ok := ctx.Value(startHandleKey{}).(*StartHandle)
226 if !ok || h == nil || h.account != account || h.app != app || h.appID != appID || h.id != id {
227 return nil, false
228 }
229 return h, true
230}
231
232func (h *StartHandle) Runner(now func() time.Time) EffectRunner {
233 return EffectRunner{w: h.w, account: h.account, app: h.app, appID: h.appID, operation: h.id, owner: h.owner, now: now}
234}
235
236// ---- exclusive operations -------------------------------------------------
237
238// OperationHandle is the capability for one exclusive app operation. It is
239// minted only by BeginExclusive, ResumeExclusive and BeginPurge.
240type OperationHandle struct {
241 w *ControlWriter
242 account, app, appID string
243 id, kind string
244 owner RegistrationOwner
245}
246
247func (h *OperationHandle) ID() string { return h.id }
248func (h *OperationHandle) Kind() string { return h.kind }
249func (h *OperationHandle) AppID() string { return h.appID }
250func (h *OperationHandle) Account() string { return h.account }
251func (h *OperationHandle) App() string { return h.app }
252func (h *OperationHandle) Owner() RegistrationOwner { return h.owner }
253
254func (h *OperationHandle) Runner(now func() time.Time) EffectRunner {
255 return EffectRunner{w: h.w, account: h.account, app: h.app, appID: h.appID, operation: h.id, owner: h.owner, now: now, exclusive: true}
256}
257
258func (w *ControlWriter) BeginExclusive(ctx context.Context, account, app, appID string, op AppOperation, owner RegistrationOwner) (*OperationHandle, error) {
259 switch op.Kind {
260 case OpDeleting, OpPlacing, OpAppMutation:
261 default:
262 return nil, ErrIdentity
263 }
264 if !validRegistrationOwner(owner) || op.AppID != appID {
265 return nil, ErrIdentity
266 }
267 if err := w.purgeAbsent(ctx, account, app); err != nil {
268 return nil, err
269 }
270 before, err := w.read(ctx, account, app, appID)
271 if err != nil {
272 return nil, err
273 }
274 if op.Kind == OpAppMutation {
275 // A joined mutation additionally waits for every registration and
276 // effect to settle: it conservatively serializes app mutations.
277 if before.Value.Op == nil && (len(before.Value.Registrations) != 0 || !effectsSettled(&before.Value, "")) {
278 return nil, ErrPending
279 }
280 }
281 if _, err = w.updateAdmission(ctx, before, func(a *AppAdmission) error { return exclusiveAdmission(a, op) }); err != nil {
282 return nil, err
283 }
284 return &OperationHandle{w: w, account: account, app: app, appID: appID, id: op.ID, kind: op.Kind, owner: owner}, nil
285}
286
287func (w *ControlWriter) ResumeExclusive(ctx context.Context, account, app, appID, opID, kind string, owner RegistrationOwner) (*OperationHandle, error) {
288 switch kind {
289 case OpDeleting, OpPlacing, OpPurging:
290 default:
291 return nil, ErrIdentity
292 }
293 if !validRegistrationOwner(owner) || opID == "" {
294 return nil, ErrIdentity
295 }
296 a, err := w.read(ctx, account, app, appID)
297 if err != nil {
298 return nil, err
299 }
300 if a.Value.State != "closed" || a.Value.Op == nil || a.Value.Op.ID != opID || a.Value.Op.Kind != kind {
301 return nil, ErrPending
302 }
303 return &OperationHandle{w: w, account: account, app: app, appID: appID, id: opID, kind: kind, owner: owner}, nil
304}
305
306func (h *OperationHandle) current(ctx context.Context) (AdmissionObserved, error) {
307 a, err := h.w.read(ctx, h.account, h.app, h.appID)
308 if err != nil {
309 return AdmissionObserved{}, err
310 }
311 if a.Value.State != "closed" || a.Value.Op == nil || a.Value.Op.ID != h.id || a.Value.Op.Kind != h.kind {
312 return AdmissionObserved{}, ErrPending
313 }
314 return a, nil
315}
316
317func (h *OperationHandle) mine(a *AppAdmission) error {
318 if a.State != "closed" || a.Op == nil || a.Op.ID != h.id || a.Op.Kind != h.kind || a.Op.AppID != h.appID {
319 return ErrPending
320 }
321 return nil
322}
323
324// Reserved intent keys only the typed methods below write.
325var reservedIntent = []string{"action", "drained.", "account", "app"}
326
327// SetPhase advances this operation's own phase and intent. It cannot touch
328// identity, registrations, effects, last_purge or last_restore.
329func (h *OperationHandle) SetPhase(ctx context.Context, from, to string, set map[string]string) error {
330 if to == "" {
331 return ErrIdentity
332 }
333 for k := range set {
334 for _, r := range reservedIntent {
335 if k == r || (strings.HasSuffix(r, ".") && strings.HasPrefix(k, r)) {
336 return ErrIdentity
337 }
338 }
339 }
340 a, err := h.current(ctx)
341 if err != nil {
342 return err
343 }
344 _, err = h.w.retryAdmission(ctx, h.account, h.app, h.appID, a, func(v *AppAdmission) error {
345 if err := h.mine(v); err != nil {
346 return err
347 }
348 if v.Op.Phase != from {
349 if v.Op.Phase == to && intentHas(v.Op.Intent, set) {
350 return nil
351 }
352 return ErrPending
353 }
354 v.Op.Phase = to
355 if v.Op.Intent == nil {
356 v.Op.Intent = map[string]string{}
357 }
358 for k, val := range set {
359 v.Op.Intent[k] = val
360 }
361 return nil
362 })
363 return err
364}
365
366func intentHas(have, want map[string]string) bool {
367 for k, v := range want {
368 if have[k] != v {
369 return false
370 }
371 }
372 return true
373}
374
375// finish reopens admission. Spec §5.9: every reopen advances the epoch, so
376// no permit, registration or receipt check bound to the closed epoch can
377// act in the reopened one. bumpEpoch is kept for call-site clarity and is
378// always true.
379func (h *OperationHandle) finish(ctx context.Context, check func(*AppAdmission) error, bumpEpoch bool) error {
380 bumpEpoch = true
381 a, err := h.current(ctx)
382 if err != nil {
383 return err
384 }
385 // Retried: h.mine fails once the reopen landed (the retry then reports
386 // ErrPending, never a second epoch bump).
387 _, err = h.w.retryAdmission(ctx, h.account, h.app, h.appID, a, func(v *AppAdmission) error {
388 if err := h.mine(v); err != nil {
389 return err
390 }
391 if len(v.Registrations) != 0 || !effectsSettled(v, "") {
392 return ErrPending
393 }
394 if check != nil {
395 if err := check(v); err != nil {
396 return err
397 }
398 }
399 v.Op = nil
400 v.State = "open"
401 if bumpEpoch {
402 v.Epoch++
403 }
404 return nil
405 })
406 return err
407}
408
409// FinishDelete reopens with a new epoch: no old registration can renew into
410// the next incarnation's admission.
411func (h *OperationHandle) FinishDelete(ctx context.Context) error {
412 if h.kind != OpDeleting {
413 return ErrIdentity
414 }
415 return h.finish(ctx, nil, true)
416}
417
418func (h *OperationHandle) FinishPlacement(ctx context.Context) error {
419 if h.kind != OpPlacing {
420 return ErrIdentity
421 }
422 return h.finish(ctx, func(v *AppAdmission) error {
423 if v.Op.Phase != "releasing" {
424 return ErrPending
425 }
426 return nil
427 }, false)
428}
429
430func (h *OperationHandle) FinishMutation(ctx context.Context) error {
431 if h.kind != OpAppMutation {
432 return ErrIdentity
433 }
434 return h.finish(ctx, func(v *AppAdmission) error {
435 if r := v.LastRestore; r != nil && r.ID == h.id && r.State != "complete" {
436 return ErrPending
437 }
438 return nil
439 }, false)
440}
441
442// ---- restore receipts (review P2-3) ----------------------------------------
443
444// PrepareRestore persists the receipt, bound to the exact tombstone ETag,
445// BEFORE the conditional tombstone delete.
446func (h *OperationHandle) PrepareRestore(ctx context.Context, tombstoneETag string, deletedAt int64) error {
447 if h.kind != OpAppMutation || tombstoneETag == "" || deletedAt <= 0 {
448 return ErrIdentity
449 }
450 a, err := h.current(ctx)
451 if err != nil {
452 return err
453 }
454 if a.Value.Op.Intent["action"] != "restore" {
455 return ErrIdentity
456 }
457 _, err = h.w.retryAdmission(ctx, h.account, h.app, h.appID, a, func(v *AppAdmission) error {
458 if err := h.mine(v); err != nil {
459 return err
460 }
461 if r := v.LastRestore; r != nil && r.ID == h.id {
462 if r.TombstoneETag == tombstoneETag && r.DeletedAt == deletedAt {
463 return nil
464 }
465 return ErrIdentity
466 }
467 v.LastRestore = &RestoreMark{ID: h.id, AppID: h.appID, Epoch: v.Epoch, TombstoneETag: tombstoneETag, DeletedAt: deletedAt, State: "prepared"}
468 return nil
469 })
470 return err
471}
472
473// CompleteRestore marks the receipt complete once the exact tombstone
474// version is gone. A different current tombstone is a refusal.
475func (h *OperationHandle) CompleteRestore(ctx context.Context) error {
476 if h.kind != OpAppMutation {
477 return ErrIdentity
478 }
479 return h.w.completeRestore(ctx, h.account, h.app, h.appID, h.id, false)
480}
481
482func (w *ControlWriter) completeRestore(ctx context.Context, account, app, appID, opID string, finishDelete bool) error {
483 a, err := w.read(ctx, account, app, appID)
484 if err != nil {
485 return err
486 }
487 r := a.Value.LastRestore
488 if a.Value.Op == nil || a.Value.Op.ID != opID || a.Value.Op.Kind != OpAppMutation || r == nil || r.ID != opID || r.AppID != appID || r.Epoch != a.Value.Epoch {
489 return ErrPending
490 }
491 if r.State == "complete" {
492 return nil
493 }
494 key := release.AppTombstoneKey(account, app)
495 _, tag, gerr := w.raw.Get(ctx, key)
496 switch {
497 case errors.Is(gerr, os.ErrNotExist):
498 case gerr != nil:
499 return gerr
500 case tag == r.TombstoneETag && finishDelete:
501 // Recovery finishes the user's restore rather than abandoning it:
502 // the exact-ETag delete is idempotent and cannot erase a successor.
503 if err = release.DeleteIfMatch(ctx, w.raw, key, r.TombstoneETag); err != nil {
504 return err
505 }
506 case tag == r.TombstoneETag:
507 return ErrPending
508 default:
509 // No trash can run while this operation holds the app.
510 return ErrIdentity
511 }
512 // Retried: a concurrent settle or renewal must not leave a deleted
513 // marker behind an incomplete receipt.
514 _, err = w.retryAdmission(ctx, account, app, appID, a, func(v *AppAdmission) error {
515 if v.Op == nil || v.Op.ID != opID || v.LastRestore == nil || v.LastRestore.ID != opID || v.LastRestore.TombstoneETag != r.TombstoneETag {
516 return ErrPending
517 }
518 v.LastRestore.State = "complete"
519 return nil
520 })
521 return err
522}
523
524// recoveryOwner stands in for an issuer when recovery itself records the
525// outcome of an effect the issuer never prepared. It is not a registration
526// owner any handle can hold, so no runner can advance or reuse the effect.
527func recoveryOwner(opID string) RegistrationOwner {
528 return RegistrationOwner{Console: "recovery", Enrollment: "recovery", Controller: opID}
529}
530
531// recoveredMutation reports that opID's app-mutation already ended with a
532// settled outcome (another console's recovery, or the issuer, reopened it).
533func recoveredMutation(v AppAdmission, opID string) bool {
534 if v.Op != nil && v.Op.ID == opID {
535 return false
536 }
537 if r := v.LastRestore; r != nil && r.ID == opID {
538 e, ok := v.Effects[opID]
539 return r.State == "complete" && ok && e.State == "complete"
540 }
541 e, ok := v.Effects[opID]
542 return ok && e.OperationID == opID && e.State == "complete" && effectsSettled(&v, opID)
543}
544
545// RecoverMutation is the only automatic way to reopen an abandoned
546// app-mutation. It never re-runs a backend call. Three shapes:
547//
548// - restore with a saved receipt (any effect state, including no effect:
549// the receipt precedes Issue): the bucket decides. The exact-ETag delete
550// is finished, the op's effect is recorded complete (created when the
551// issuer never prepared it, which also fences a late Issue: it finds a
552// complete effect and gets EffectResult, never a permit), then reopen.
553// A receipt-less restore stays pending here; AbandonRestore owns it.
554// - mail: reopens only once the shared gate no longer holds this op's
555// gate ID AND every effect of the op is complete (a merely prepared one
556// is superseded). An issued or uncertain effect keeps the op; the gate
557// reconciler (ReconcileMailGate) settles those first.
558// - anything else (generic): reopens when nothing of the op is issued or
559// uncertain, superseding its merely prepared effects in the same CAS
560// (abandonUnissuedMutation). Unless the op's own effect is already
561// complete, the op must be at least UnissuedMutationRecoverAfter old,
562// so recovery does not race a live issuer between BeginExclusive and
563// its issued CAS. That age is liveness only; the safety argument is
564// the epoch-advancing reopen, which EffectRunner.Issue's issued CAS
565// re-checks (authorized), so a late issuer is fenced.
566//
567// An issued or uncertain generic effect is never settled here, by age or by
568// readback: it is an activation dependency that needs a quiescence protocol.
569//
570// It is idempotent: an op that some other path already reopened with a
571// settled outcome returns nil.
572func (w *ControlWriter) RecoverMutation(ctx context.Context, account, app, appID, opID string) error {
573 a, err := w.read(ctx, account, app, appID)
574 if err != nil {
575 return err
576 }
577 if a.Value.Op == nil || a.Value.Op.ID != opID {
578 if recoveredMutation(a.Value, opID) {
579 return nil
580 }
581 return ErrPending
582 }
583 if a.Value.State != "closed" || a.Value.Op.Kind != OpAppMutation || len(a.Value.Registrations) != 0 {
584 return ErrPending
585 }
586 switch a.Value.Op.Intent["action"] {
587 case "restore":
588 if r := a.Value.LastRestore; r != nil && r.ID == opID {
589 return w.recoverRestoreReceipt(ctx, account, app, appID, opID)
590 }
591 // No receipt: nothing may have been issued yet, or the issuer is
592 // between BeginExclusive and PrepareRestore. AbandonRestore decides.
593 return ErrPending
594 case "mail":
595 return w.recoverMailMutation(ctx, account, app, appID, opID, a.Value.Op.Intent["gate"])
596 }
597 return w.abandonUnissuedMutation(ctx, account, app, appID, opID, UnissuedMutationRecoverAfter)
598}
599
600// MutationActionStarted is the intent key RunAppMutation sets at
601// BeginExclusive (unix seconds). It is the same key restore uses.
602const MutationActionStarted = RestoreActionStarted
603
604// UnissuedMutationRecoverAfter is how long the sweep leaves a generic
605// app-mutation with no settled effect to its issuer. An op without the
606// started stamp (a lost-response begin from an older build) is eligible at
607// once: the reopen fences it either way.
608const UnissuedMutationRecoverAfter = time.Minute
609
610// abandonUnissued is the issuer's own recovery after Issue failed: f never
611// ran, so no age bound applies.
612func (h *OperationHandle) abandonUnissued(ctx context.Context) error {
613 if h.kind != OpAppMutation {
614 return ErrIdentity
615 }
616 return h.w.abandonUnissuedMutation(ctx, h.account, h.app, h.appID, h.id, 0)
617}
618
619// abandonUnissuedMutation reopens a generic app-mutation (not restore, not
620// mail: those keep their specialized recovery) in one CAS when every effect
621// is complete or one of this op's merely prepared effects, which it
622// supersedes. An issued or uncertain effect of any operation, a
623// registration, or a restore receipt refuses. The reopen advances the epoch,
624// so the op's issuer can no longer issue (EffectRunner.authorized), and a
625// superseded effect fails its prepared->issued CAS.
626func (w *ControlWriter) abandonUnissuedMutation(ctx context.Context, account, app, appID, opID string, minAge time.Duration) error {
627 if err := w.ready(); err != nil {
628 return err
629 }
630 now := w.now()
631 _, err := w.retryAdmission(ctx, account, app, appID, AdmissionObserved{}, func(v *AppAdmission) error {
632 if v.State != "closed" || v.Op == nil || v.Op.ID != opID || v.Op.Kind != OpAppMutation || v.Op.AppID != appID || len(v.Registrations) != 0 {
633 return ErrPending
634 }
635 switch v.Op.Intent["action"] {
636 case "restore", "mail":
637 return ErrPending
638 }
639 if r := v.LastRestore; r != nil && r.ID == opID {
640 return ErrPending
641 }
642 own, ok := v.Effects[opID]
643 if minAge > 0 && !(ok && own.OperationID == opID && own.AppID == appID && own.State == "complete") {
644 sec, perr := strconv.ParseInt(v.Op.Intent[MutationActionStarted], 10, 64)
645 if perr == nil && sec > 0 && now.Before(time.Unix(sec, 0).Add(minAge)) {
646 return ErrPending
647 }
648 }
649 for id, e := range v.Effects {
650 switch {
651 case e.State == "complete":
652 case e.OperationID == opID && e.AppID == appID && e.State == "prepared":
653 e.State, e.ResultReference = "complete", "superseded"
654 v.Effects[id] = e
655 default:
656 return ErrPending // issued, uncertain, or another operation's
657 }
658 }
659 if v.Epoch == math.MaxUint64 {
660 return ErrPending
661 }
662 v.Op, v.State = nil, "open"
663 v.Epoch++
664 return nil
665 })
666 return err
667}
668
669func (w *ControlWriter) recoverRestoreReceipt(ctx context.Context, account, app, appID, opID string) error {
670 if err := w.sqlChildrenSettled(ctx, account, app, appID, opID); err != nil {
671 return err
672 }
673 if err := w.completeRestore(ctx, account, app, appID, opID, true); err != nil {
674 return err
675 }
676 if _, err := w.retryAdmission(ctx, account, app, appID, AdmissionObserved{}, func(v *AppAdmission) error {
677 r := v.LastRestore
678 if v.State != "closed" || v.Op == nil || v.Op.ID != opID || v.Op.Kind != OpAppMutation || r == nil || r.ID != opID || r.State != "complete" {
679 return ErrPending
680 }
681 for id, e := range v.Effects {
682 if e.OperationID != opID || e.State == "complete" || id == opID {
683 continue
684 }
685 if e.State != "prepared" {
686 // An issued/uncertain effect under the restore op that is
687 // not the receipt-bound one: its outcome is not decidable
688 // from the bucket.
689 return ErrPending
690 }
691 e.State, e.ResultReference = "complete", "superseded"
692 v.Effects[id] = e
693 }
694 e, ok := v.Effects[opID]
695 switch {
696 case !ok:
697 v.Effects[opID] = Effect{ID: opID, OperationID: opID, AppID: appID, Resource: OpAppMutation, Action: "restore",
698 Inputs: map[string]string{"tombstone_etag": r.TombstoneETag}, Owner: recoveryOwner(opID),
699 State: "complete", ResultReference: "restore-reconciled"}
700 case e.OperationID != opID || e.AppID != appID:
701 return ErrIdentity
702 case e.State != "complete":
703 // prepared (never called), issued or uncertain: the call's only
704 // external effect is the receipt-bound delete, now decided.
705 e.State, e.ResultReference = "complete", "restore-reconciled"
706 v.Effects[opID] = e
707 }
708 return nil
709 }); err != nil {
710 return err
711 }
712 h := &OperationHandle{w: w, account: account, app: app, appID: appID, id: opID, kind: OpAppMutation}
713 return h.finish(ctx, func(v *AppAdmission) error {
714 if r := v.LastRestore; r == nil || r.ID != opID || r.State != "complete" {
715 return ErrPending
716 }
717 return nil
718 }, true)
719}
720
721// recoverMailMutation ends a mail member operation whose gate was released
722// or reconciled. Never while the gate still holds this gate ID: ops end only
723// after the gate's release CAS.
724func (w *ControlWriter) recoverMailMutation(ctx context.Context, account, app, appID, opID, gateID string) error {
725 if err := w.sqlChildrenSettled(ctx, account, app, appID, opID); err != nil {
726 return err
727 }
728 if gateID == "" {
729 return ErrIdentity
730 }
731 g, _, err := w.readMailGate(ctx)
732 if err != nil {
733 return err
734 }
735 if g.State == "held" && g.ID == gateID {
736 return ErrPending
737 }
738 if _, err = w.retryAdmission(ctx, account, app, appID, AdmissionObserved{}, func(v *AppAdmission) error {
739 if v.State != "closed" || v.Op == nil || v.Op.ID != opID || v.Op.Kind != OpAppMutation || v.Op.Intent["gate"] != gateID {
740 return ErrPending
741 }
742 for id, e := range v.Effects {
743 if e.OperationID != opID || e.State == "complete" {
744 continue
745 }
746 if e.State != "prepared" {
747 return ErrPending // issued or uncertain: ReconcileMailGate first
748 }
749 // A prepared effect never called: its prepared->issued CAS now
750 // fails, and the gate it needed is gone anyway.
751 e.State, e.ResultReference = "complete", "superseded"
752 v.Effects[id] = e
753 }
754 return nil
755 }); err != nil {
756 return err
757 }
758 h := &OperationHandle{w: w, account: account, app: app, appID: appID, id: opID, kind: OpAppMutation}
759 return h.finish(ctx, nil, true)
760}
761
762// ---- drains and settlement --------------------------------------------------
763
764type drainOutcome int
765
766const (
767 drainKeep drainOutcome = iota
768 drainRemove
769)
770
771// judgeRegistration decides whether one saved registration can no longer
772// produce an effect. It may write a generation-only claim CAS that
773// invalidates the saved observation. recoverCreate is the caller's own
774// controller hook for a start that already owns the host.
775func (w *ControlWriter) judgeRegistration(ctx context.Context, a *AppAdmission, account, app, id string, reg Registration, now time.Time, recoverCreate func(context.Context, string) error) (drainOutcome, SavedClaim, error) {
776 if !effectsSettled(a, id) {
777 return drainKeep, SavedClaim{}, nil
778 }
779 if reg.Kind == RegistrationIngest {
780 if len(a.Effects) > 0 {
781 for _, e := range a.Effects {
782 if e.OperationID == id {
783 return drainRemove, SavedClaim{}, nil
784 }
785 }
786 }
787 if RegistrationDrained(reg, now) {
788 return drainRemove, SavedClaim{}, nil
789 }
790 return drainKeep, SavedClaim{}, nil
791 }
792 if reg.Kind == registrationPurge || reg.Kind == RegistrationWakeRetryReset {
793 if RegistrationDrained(reg, now) {
794 return drainRemove, SavedClaim{}, nil
795 }
796 return drainKeep, SavedClaim{}, nil
797 }
798 claim, err := hostclaim.Read(ctx, w.raw, reg.Host)
799 if err != nil {
800 return drainKeep, SavedClaim{}, err
801 }
802 if !claim.Absent && claim.Claim.Op.ID == id {
803 if claim.Claim.Op.Kind == hostclaim.KindCreating && recoverCreate != nil {
804 if err = recoverCreate(ctx, reg.Host); err != nil && !errors.Is(err, ErrPending) {
805 return drainKeep, SavedClaim{}, err
806 }
807 }
808 // An accepted start owns the host until its own recovery settles.
809 return drainKeep, SavedClaim{}, nil
810 }
811 h := claim.Claim
812 completed := !claim.Absent && h.Holder != nil && h.Holder.Account == account && h.Holder.App == app && h.Holder.Incarnation == reg.Incarnation && h.Op.Kind == hostclaim.KindServing
813 cancelled := !claim.Absent && reg.Incarnation != "" && slices.Contains(h.RetiredIncarnations, reg.Incarnation)
814 if completed || cancelled {
815 return drainRemove, SavedObservation(claim), nil
816 }
817 if !RegistrationDrained(reg, now) {
818 return drainKeep, SavedClaim{}, nil
819 }
820 if SavedObservation(claim) == reg.Claim {
821 // Generation-only CAS; for prior absence this writes a free claim.
822 // Never Delete(claim).
823 if claim, err = hostclaim.Update(ctx, w.raw, claim, func(*hostclaim.Claim) error { return nil }); err != nil {
824 return drainKeep, SavedClaim{}, err
825 }
826 }
827 if SavedObservation(claim) == reg.Claim {
828 return drainKeep, SavedClaim{}, nil
829 }
830 return drainRemove, SavedObservation(claim), nil
831}
832
833// Drain removes every saved registration that can no longer act, journaling
834// each witness into the operation intent in the same CAS. ErrPending while
835// any registration or effect is still live.
836func (h *OperationHandle) Drain(ctx context.Context, now time.Time, recoverCreate func(context.Context, string) error) error {
837 if h.kind != OpDeleting && h.kind != OpPlacing {
838 return ErrIdentity
839 }
840 a, err := h.current(ctx)
841 if err != nil {
842 return err
843 }
844 for id, reg := range a.Value.Registrations {
845 out, witness, err := h.w.judgeRegistration(ctx, &a.Value, h.account, h.app, id, reg, now, recoverCreate)
846 if err != nil {
847 return err
848 }
849 if out != drainRemove {
850 return ErrPending
851 }
852 evidence, err := json.Marshal(struct {
853 Registration Registration `json:"registration"`
854 Witness SavedClaim `json:"witness"`
855 }{reg, witness})
856 if err != nil {
857 return err
858 }
859 fresh, err := h.current(ctx)
860 if err != nil {
861 return err
862 }
863 if a, err = h.w.retryAdmission(ctx, h.account, h.app, h.appID, fresh, func(v *AppAdmission) error {
864 if err := h.mine(v); err != nil {
865 return err
866 }
867 saved, ok := v.Registrations[id]
868 if !ok {
869 return nil
870 }
871 if saved != reg {
872 return ErrPending
873 }
874 if v.Op.Intent == nil {
875 v.Op.Intent = map[string]string{}
876 }
877 v.Op.Intent["drained."+id] = string(evidence)
878 delete(v.Registrations, id)
879 return nil
880 }); err != nil {
881 return err
882 }
883 }
884 if len(a.Value.Registrations) != 0 || !effectsSettled(&a.Value, "") {
885 return ErrPending
886 }
887 return nil
888}
889
890// SettleRegistration removes one finished registration while no exclusive
891// operation exists (the recovery sweep).
892func (w *ControlWriter) SettleRegistration(ctx context.Context, account, app, appID, id string, now time.Time) error {
893 a, err := w.read(ctx, account, app, appID)
894 if err != nil {
895 return err
896 }
897 reg, ok := a.Value.Registrations[id]
898 if !ok {
899 return nil
900 }
901 if a.Value.Op != nil {
902 return ErrPending
903 }
904 out, _, err := w.judgeRegistration(ctx, &a.Value, account, app, id, reg, now, nil)
905 if err != nil {
906 return err
907 }
908 if out != drainRemove {
909 return ErrPending
910 }
911 _, err = w.retryAdmission(ctx, account, app, appID, AdmissionObserved{}, func(v *AppAdmission) error {
912 if v.Op != nil || v.Registrations[id] != reg || !effectsSettled(v, id) {
913 return ErrPending
914 }
915 delete(v.Registrations, id)
916 return nil
917 })
918 return err
919}
920