| 1 | package hostops |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "time" |
| 7 | ) |
| 8 | |
| 9 | // ErrNotApplied marks a call that returned before issuing any backend or |
| 10 | // store effect (validation, refusal). Its effect settles complete with |
| 11 | // result "not-applied" instead of uncertain. |
| 12 | var ErrNotApplied = errors.New("hostops: effect not applied") |
| 13 | |
| 14 | // EffectRunner issues durable effects under one registration or exclusive |
| 15 | // operation. Its fields are unexported: runners come only from a |
| 16 | // StartHandle or OperationHandle, so a caller cannot impersonate an owner |
| 17 | // it merely read from the bucket. |
| 18 | type EffectRunner struct { |
| 19 | w *ControlWriter |
| 20 | account, app, appID string |
| 21 | operation string |
| 22 | owner RegistrationOwner |
| 23 | now func() time.Time |
| 24 | exclusive bool |
| 25 | } |
| 26 | |
| 27 | func (r EffectRunner) AppID() string { return r.appID } |
| 28 | func (r EffectRunner) Operation() string { return r.operation } |
| 29 | func (r EffectRunner) Owner() RegistrationOwner { return r.owner } |
| 30 | |
| 31 | func (r EffectRunner) authorized(a *AppAdmission) bool { |
| 32 | if r.exclusive { |
| 33 | return a.State == "closed" && a.Op != nil && a.Op.ID == r.operation && a.Op.AppID == r.appID |
| 34 | } |
| 35 | reg, ok := a.Registrations[r.operation] |
| 36 | return ok && reg.Owner == r.owner && reg.AppID == r.appID && reg.Epoch == a.Epoch && RegistrationLive(reg, r.now()) |
| 37 | } |
| 38 | |
| 39 | // Issue prepares and issues e, then returns a context carrying a live permit |
| 40 | // and the settle function. settle MUST be called exactly once after every |
| 41 | // call made with the context has joined; it revokes the permit first, then |
| 42 | // records complete or uncertain. A deadline is never settlement. |
| 43 | func (r EffectRunner) Issue(ctx context.Context, e Effect) (context.Context, func(result string, callErr error) error, error) { |
| 44 | if r.w == nil || r.now == nil || r.operation == "" { |
| 45 | return nil, nil, ErrContract |
| 46 | } |
| 47 | if e.AppID == "" { |
| 48 | e.AppID = r.appID |
| 49 | } |
| 50 | if e.OperationID == "" { |
| 51 | e.OperationID = r.operation |
| 52 | } |
| 53 | if e.Owner == (RegistrationOwner{}) { |
| 54 | e.Owner = r.owner |
| 55 | } |
| 56 | if e.Owner != r.owner || e.AppID != r.appID || e.OperationID != r.operation { |
| 57 | return nil, nil, ErrContract |
| 58 | } |
| 59 | if err := CheckCreateFreeze(ctx, r.w.raw, r.w.pub); err != nil { |
| 60 | return nil, nil, err |
| 61 | } |
| 62 | a, err := r.w.read(ctx, r.account, r.app, r.appID) |
| 63 | if err != nil { |
| 64 | return nil, nil, err |
| 65 | } |
| 66 | if old, ok := a.Value.Effects[e.ID]; ok { |
| 67 | // A recorded ID is a result or pending work, never a fresh permit. |
| 68 | if old.AppID != e.AppID || old.OperationID != e.OperationID || old.Resource != e.Resource || old.Action != e.Action { |
| 69 | return nil, nil, ErrIdentity |
| 70 | } |
| 71 | if old.State == "complete" { |
| 72 | return nil, nil, errEffectComplete{result: old.ResultReference} |
| 73 | } |
| 74 | return nil, nil, ErrPending |
| 75 | } |
| 76 | e.State = "prepared" |
| 77 | // Authorization is rechecked inside each CAS, so an operation reopened by |
| 78 | // recovery between read and write can never gain this effect. Both CASes |
| 79 | // below run BEFORE the call, so replaying them after a lost race is safe: |
| 80 | // prepareEffect is idempotent for the identical effect, and |
| 81 | // advanceEffect(prepared->issued) refuses an effect someone superseded. |
| 82 | if a, err = r.w.retryAdmission(ctx, r.account, r.app, r.appID, a, func(v *AppAdmission) error { |
| 83 | if !r.authorized(v) { |
| 84 | return ErrPending |
| 85 | } |
| 86 | if r.exclusive { |
| 87 | // A same-operation effect that was only prepared never issued its |
| 88 | // call (a crashed or competing worker). Supersede it in this CAS; |
| 89 | // that worker's prepared->issued CAS then fails. Issued and |
| 90 | // uncertain effects still block. |
| 91 | for id, old := range v.Effects { |
| 92 | if old.OperationID == r.operation && old.Resource == e.Resource && old.State == "prepared" && id != e.ID { |
| 93 | old.State, old.ResultReference = "complete", "superseded" |
| 94 | v.Effects[id] = old |
| 95 | } |
| 96 | } |
| 97 | } |
| 98 | return prepareEffect(v, e) |
| 99 | }); err != nil { |
| 100 | return nil, nil, err |
| 101 | } |
| 102 | if a, err = r.w.retryAdmission(ctx, r.account, r.app, r.appID, a, func(v *AppAdmission) error { |
| 103 | if !r.authorized(v) { |
| 104 | return ErrPending |
| 105 | } |
| 106 | if cur, ok := v.Effects[e.ID]; !ok || cur.State != "prepared" && cur.State != "issued" { |
| 107 | // Superseded or reconciled while we raced: never call. |
| 108 | return ErrPending |
| 109 | } |
| 110 | return advanceEffect(v, e.ID, r.owner, "issued", "") |
| 111 | }); err != nil { |
| 112 | return nil, nil, err |
| 113 | } |
| 114 | issuedCtx, revoke := withAppMutationEffect(ctx, r, e, a.Value.Epoch) |
| 115 | issuedCtx = context.WithValue(issuedCtx, effectPermitKey{}, effectPermit{Resource: e.Resource}) |
| 116 | settled := false |
| 117 | settle := func(result string, callErr error) error { |
| 118 | if settled { |
| 119 | return ErrContract |
| 120 | } |
| 121 | settled = true |
| 122 | revoke() |
| 123 | state := "complete" |
| 124 | switch { |
| 125 | case callErr != nil && errors.Is(callErr, ErrNotApplied): |
| 126 | result = "not-applied" |
| 127 | case callErr != nil || result == "": |
| 128 | // A transport error does not prove the server did not act. |
| 129 | state = "uncertain" |
| 130 | } |
| 131 | decided, saveErr := r.saveSettlement(ctx, e.ID, "", state, result) |
| 132 | if saveErr != nil && !decided { |
| 133 | return errors.Join(callErr, saveErr) |
| 134 | } |
| 135 | if decided { |
| 136 | return callErr |
| 137 | } |
| 138 | if callErr != nil { |
| 139 | return callErr |
| 140 | } |
| 141 | if state != "complete" { |
| 142 | return ErrPending |
| 143 | } |
| 144 | return nil |
| 145 | } |
| 146 | return issuedCtx, settle, nil |
| 147 | } |
| 148 | |
| 149 | // saveSettlement records a settle step's outcome for effect id. The outcome |
| 150 | // is known now and only now: a lost CAS against a routine renewal or a |
| 151 | // sibling effect re-reads and re-applies (bounded), so the effect does not |
| 152 | // stay "issued" for want of one retry. Settle needs no live registration: |
| 153 | // advanceEffect checks the owner only. A non-empty from is the only state |
| 154 | // the effect may leave. decided reports that recovery already recorded a |
| 155 | // different complete decision, which stands. |
| 156 | func (r EffectRunner) saveSettlement(ctx context.Context, id, from, state, result string) (decided bool, err error) { |
| 157 | saveCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 20*time.Second) |
| 158 | defer cancel() |
| 159 | _, err = r.w.retryAdmission(saveCtx, r.account, r.app, r.appID, AdmissionObserved{}, func(v *AppAdmission) error { |
| 160 | cur, ok := v.Effects[id] |
| 161 | if ok && cur.State == "complete" && (cur.State != state || cur.ResultReference != result) { |
| 162 | // Recovery already decided this effect (restore receipt, gate |
| 163 | // reconciliation). Its decision stands; never overwrite it. |
| 164 | decided = true |
| 165 | return errSettledElsewhere |
| 166 | } |
| 167 | if from != "" && ok && cur.State != from && cur.State != state { |
| 168 | return ErrPending |
| 169 | } |
| 170 | if cur.State == "prepared" && cur.Owner == r.owner && cur.OperationID == r.operation && cur.Action == sqlIngestAction && state == "complete" && result == "sql-fenced" { |
| 171 | // SQL's durable fence blocks a delayed issued CAS from ever reaching a |
| 172 | // write. Superseding prepared also makes that CAS refuse immediately. |
| 173 | cur.State, cur.ResultReference = state, result |
| 174 | v.Effects[id] = cur |
| 175 | return nil |
| 176 | } |
| 177 | return advanceEffect(v, id, r.owner, state, result) |
| 178 | }) |
| 179 | return decided, err |
| 180 | } |
| 181 | |
| 182 | // resolveUncertain is the settle step for a move-start effect that Issue |
| 183 | // left uncertain, once ResolveMoveStart has had the bucket decide its record |
| 184 | // Put: complete with the move id, or "not-applied". It is scoped to this |
| 185 | // runner's own registration and to that one action. An exclusive runner, |
| 186 | // another operation, owner or action refuses, and only an uncertain effect |
| 187 | // (or one already holding this same decision) is written. |
| 188 | func (r EffectRunner) resolveUncertain(ctx context.Context, e Effect, result string) error { |
| 189 | if r.w == nil || r.exclusive || r.operation == "" || e.Action != moveStartAction || e.OperationID != r.operation || e.AppID != r.appID || e.Owner != r.owner || result == "" { |
| 190 | return ErrContract |
| 191 | } |
| 192 | decided, err := r.saveSettlement(ctx, e.ID, "uncertain", "complete", result) |
| 193 | if decided { |
| 194 | return errSettledElsewhere |
| 195 | } |
| 196 | return err |
| 197 | } |
| 198 | |
| 199 | // Run issues e, calls issue exactly once with the permitted context, joins |
| 200 | // it, and settles. issue must not start a goroutine that outlives it. |
| 201 | func (r EffectRunner) Run(ctx context.Context, e Effect, issue func(context.Context) (string, error)) (string, error) { |
| 202 | if issue == nil { |
| 203 | return "", ErrContract |
| 204 | } |
| 205 | issuedCtx, settle, err := r.Issue(ctx, e) |
| 206 | var done errEffectComplete |
| 207 | if errors.As(err, &done) { |
| 208 | return done.result, nil |
| 209 | } |
| 210 | if err != nil { |
| 211 | return "", err |
| 212 | } |
| 213 | var result string |
| 214 | var callErr error |
| 215 | func() { |
| 216 | defer func() { |
| 217 | if p := recover(); p != nil { |
| 218 | _ = settle("", errors.New("hostops: effect call panicked")) |
| 219 | panic(p) |
| 220 | } |
| 221 | }() |
| 222 | result, callErr = issue(issuedCtx) |
| 223 | }() |
| 224 | if err = settle(result, callErr); err != nil { |
| 225 | return "", err |
| 226 | } |
| 227 | return result, nil |
| 228 | } |
| 229 | |
| 230 | // EffectResult reports Issue's "already complete" refusal and its saved |
| 231 | // result. That refusal grants no permit; the caller must not call again. |
| 232 | func EffectResult(err error) (string, bool) { |
| 233 | var done errEffectComplete |
| 234 | if errors.As(err, &done) { |
| 235 | return done.result, true |
| 236 | } |
| 237 | return "", false |
| 238 | } |
| 239 | |
| 240 | type errEffectComplete struct{ result string } |
| 241 | |
| 242 | // errSettledElsewhere aborts a settle CAS whose effect recovery already |
| 243 | // recorded complete. |
| 244 | var errSettledElsewhere = errors.New("hostops: effect settled by recovery") |
| 245 | |
| 246 | func (e errEffectComplete) Error() string { return "hostops: effect already complete" } |
| 247 | |
amadan