| 1 | package hostops |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "encoding/hex" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "strconv" |
| 10 | "sync/atomic" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/release" |
| 14 | ) |
| 15 | |
| 16 | type issuedMutationKey struct{} |
| 17 | |
| 18 | // MarkAppMutationIssued is called immediately before an app effect reaches |
| 19 | // a raw sink. It never grants authority; it records whether failure may be |
| 20 | // ambiguous so a validation refusal can release an unissued operation. |
| 21 | func MarkAppMutationIssued(ctx context.Context) { |
| 22 | // Every enclosing operation is marked, not only the innermost: a nested |
| 23 | // scope's raw write is also a possible late effect of each outer operation, |
| 24 | // so none of them may release as unapplied. |
| 25 | marks, _ := ctx.Value(issuedMutationKey{}).([]*atomic.Bool) |
| 26 | for _, p := range marks { |
| 27 | if p != nil { |
| 28 | p.Store(true) |
| 29 | } |
| 30 | } |
| 31 | } |
| 32 | |
| 33 | // ObserveAppMutationEffects records raw effects in this scope and keeps all |
| 34 | // enclosing observers. It grants no authority and does not settle anything. |
| 35 | func ObserveAppMutationEffects(ctx context.Context) (context.Context, func() bool) { |
| 36 | issued := new(atomic.Bool) |
| 37 | outer, _ := ctx.Value(issuedMutationKey{}).([]*atomic.Bool) |
| 38 | marks := append(append([]*atomic.Bool(nil), outer...), issued) |
| 39 | return context.WithValue(ctx, issuedMutationKey{}, marks), issued.Load |
| 40 | } |
| 41 | |
| 42 | // BindMutationRunner runs once at bootstrap, before publishing the boundary. |
| 43 | func BindMutationRunner(s release.Store, w *ControlWriter, owner RegistrationOwner, localAppID func(context.Context, string, string) (string, error)) error { |
| 44 | b, ok := s.(*mutationBoundary) |
| 45 | if !ok || w == nil || !validRegistrationOwner(owner) || b.control != nil || localAppID == nil { |
| 46 | return ErrContract |
| 47 | } |
| 48 | b.localAppID = localAppID |
| 49 | b.control = w |
| 50 | b.owner = owner |
| 51 | return nil |
| 52 | } |
| 53 | |
| 54 | // unissuedRecoveryTimeout bounds the issuer's own uncancelled attempt to end |
| 55 | // an operation whose effect it never issued. |
| 56 | const unissuedRecoveryTimeout = 20 * time.Second |
| 57 | |
| 58 | func (b *mutationBoundary) RunAppMutation(ctx context.Context, account, app, action string, f func(context.Context) error) error { |
| 59 | if b.control == nil || b.localAppID == nil || f == nil || action == "" { |
| 60 | return ErrContract |
| 61 | } |
| 62 | // Resolve the requesting console's identity before reading shared state. |
| 63 | // A current bucket name is not authority to adopt a recreated app. |
| 64 | appID, err := b.localAppID(ctx, account, app) |
| 65 | if err != nil { |
| 66 | return err |
| 67 | } |
| 68 | if appID == "" { |
| 69 | return ErrIdentity |
| 70 | } |
| 71 | a, err := ReadAdmission(ctx, b.raw, account, app, appID) |
| 72 | if err != nil { |
| 73 | return err |
| 74 | } |
| 75 | if RequireAppMutation(ctx, b.raw, account, app, a.Value.AppID) == nil { |
| 76 | return f(ctx) |
| 77 | } |
| 78 | // A callback carrying retired authority cannot acquire a successor's |
| 79 | // incarnation merely by retrying through a high-level adapter. |
| 80 | inherited, _ := ctx.Value(appMutationPermitKey{}).([]*AppMutationPermit) |
| 81 | for _, permit := range inherited { |
| 82 | if permit != nil && permit.account == account && permit.app == app { |
| 83 | return ErrPending |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | var nonce [24]byte |
| 88 | if _, err = rand.Read(nonce[:]); err != nil { |
| 89 | return err |
| 90 | } |
| 91 | id := hex.EncodeToString(nonce[:]) |
| 92 | owner := b.owner |
| 93 | owner.Controller = id |
| 94 | intent := map[string]string{"action": action, MutationActionStarted: strconv.FormatInt(b.control.now().Unix(), 10)} |
| 95 | h, err := b.control.BeginExclusive(ctx, account, app, a.Value.AppID, AppOperation{ID: id, Kind: OpAppMutation, AppID: a.Value.AppID, Phase: "issued", Intent: intent}, owner) |
| 96 | if err != nil { |
| 97 | if !RefusedBeforeEffect(err) { |
| 98 | // The begin CAS may have landed with its response lost. Nothing |
| 99 | // was issued under this fresh ID, so end it if it exists. |
| 100 | // Its own refusal is the normal outcome when the CAS never landed, |
| 101 | // so the caller sees only the begin error. |
| 102 | _ = recoverUnissued(ctx, err, func(rctx context.Context) error { |
| 103 | return b.control.abandonUnissuedMutation(rctx, account, app, a.Value.AppID, id, 0) |
| 104 | }) |
| 105 | } |
| 106 | return err |
| 107 | } |
| 108 | pctx, settle, err := h.Runner(time.Now).Issue(ctx, Effect{ID: id, Resource: OpAppMutation, Action: action}) |
| 109 | if err != nil { |
| 110 | // f was never called. Absent or merely prepared effects are |
| 111 | // superseded and the app reopens with a new epoch; an issued one (a |
| 112 | // lost response on the issued CAS) keeps the operation closed. |
| 113 | return recoverUnissued(ctx, err, h.abandonUnissued) |
| 114 | } |
| 115 | issued := new(atomic.Bool) |
| 116 | outer, _ := ctx.Value(issuedMutationKey{}).([]*atomic.Bool) |
| 117 | marks := append(append([]*atomic.Bool(nil), outer...), issued) |
| 118 | pctx = context.WithValue(pctx, issuedMutationKey{}, marks) |
| 119 | callErr := f(pctx) |
| 120 | settleErr := callErr |
| 121 | if callErr != nil && !issued.Load() { |
| 122 | settleErr = errors.Join(ErrNotApplied, callErr) |
| 123 | } |
| 124 | if err = settle("joined", settleErr); err != nil && !errors.Is(err, ErrNotApplied) { |
| 125 | return errors.Join(callErr, err) |
| 126 | } |
| 127 | settled, cancel := context.WithTimeout(context.WithoutCancel(ctx), unissuedRecoveryTimeout) |
| 128 | defer cancel() |
| 129 | if err = h.FinishMutation(settled); err != nil { |
| 130 | return errors.Join(callErr, err) |
| 131 | } |
| 132 | return callErr |
| 133 | } |
| 134 | |
| 135 | // recoverUnissued runs one bounded, uncancelled recovery attempt after cause |
| 136 | // and returns cause with its classification intact: a recovery failure is |
| 137 | // reported as text only, so it cannot turn, say, context.Canceled into |
| 138 | // ErrPending for the caller. The sweep (RecoverMutation) retries later. |
| 139 | func recoverUnissued(ctx context.Context, cause error, attempt func(context.Context) error) error { |
| 140 | rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), unissuedRecoveryTimeout) |
| 141 | defer cancel() |
| 142 | if err := attempt(rctx); err != nil { |
| 143 | return fmt.Errorf("%w (unissued operation left for recovery: %v)", cause, err) |
| 144 | } |
| 145 | return cause |
| 146 | } |
| 147 | |
amadan