| 1 | package hostops |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "encoding/hex" |
| 7 | "sync/atomic" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/release" |
| 11 | ) |
| 12 | |
| 13 | type appMutationPermitKey struct{} |
| 14 | |
| 15 | // AppMutationPermit is effect-time authority for one app incarnation. Its |
| 16 | // constructor is withAppMutationEffect, called only from EffectRunner.Issue |
| 17 | // after the effect is durably issued. |
| 18 | type AppMutationPermit struct { |
| 19 | writer *ControlWriter |
| 20 | account, app, appID string |
| 21 | epoch uint64 |
| 22 | operation, effect string |
| 23 | owner RegistrationOwner |
| 24 | exclusive bool |
| 25 | sqlOnly bool |
| 26 | sqlJoined atomic.Bool |
| 27 | live atomic.Bool |
| 28 | now func() time.Time |
| 29 | } |
| 30 | |
| 31 | // PermitScope is the validated, read-only view the SQL boundary loads into |
| 32 | // its TEMP permit relation. |
| 33 | type PermitScope struct { |
| 34 | Account, App, AppID string |
| 35 | Epoch uint64 |
| 36 | Operation, Effect string |
| 37 | } |
| 38 | |
| 39 | func withAppMutationEffect(ctx context.Context, r EffectRunner, e Effect, epoch uint64) (context.Context, func()) { |
| 40 | p := &AppMutationPermit{writer: r.w, account: r.account, app: r.app, appID: r.appID, epoch: epoch, operation: e.OperationID, effect: e.ID, owner: r.owner, exclusive: r.exclusive, sqlOnly: e.Action == sqlIngestAction, now: r.now} |
| 41 | p.live.Store(true) |
| 42 | inherited, _ := ctx.Value(appMutationPermitKey{}).([]*AppMutationPermit) |
| 43 | permits := append(append([]*AppMutationPermit(nil), inherited...), p) |
| 44 | return context.WithValue(ctx, appMutationPermitKey{}, permits), func() { p.live.Store(false) } |
| 45 | } |
| 46 | |
| 47 | // HasPermitContext is a cheap pre-check for logging and fixtures. It is never |
| 48 | // authority. |
| 49 | func HasPermitContext(ctx context.Context) bool { |
| 50 | permits, _ := ctx.Value(appMutationPermitKey{}).([]*AppMutationPermit) |
| 51 | for _, p := range permits { |
| 52 | if p != nil && p.live.Load() { |
| 53 | return true |
| 54 | } |
| 55 | } |
| 56 | return false |
| 57 | } |
| 58 | |
| 59 | func (p *AppMutationPermit) valid(a AppAdmission) bool { |
| 60 | if p == nil || !p.live.Load() || p.now == nil || a.AppID != p.appID || a.Epoch != p.epoch { |
| 61 | return false |
| 62 | } |
| 63 | e, ok := a.Effects[p.effect] |
| 64 | if !ok || e.State != "issued" || e.AppID != p.appID || e.OperationID != p.operation || e.Owner != p.owner { |
| 65 | return false |
| 66 | } |
| 67 | if p.exclusive { |
| 68 | op := a.Op |
| 69 | return a.State == "closed" && op != nil && op.ID == p.operation && op.AppID == p.appID |
| 70 | } |
| 71 | r, ok := a.Registrations[p.operation] |
| 72 | // Each guarded write is checked against a live registration. A lapsed |
| 73 | // registration issues nothing further; the durable issued effect still |
| 74 | // blocks purge until it settles. |
| 75 | return ok && r.AppID == p.appID && r.Epoch == p.epoch && r.Owner == p.owner && RegistrationLive(r, p.now()) |
| 76 | } |
| 77 | |
| 78 | // RequireAppMutation validates effect-time authority for one exact app |
| 79 | // incarnation. It tries every inherited permit (a shared operation carries a |
| 80 | // complete app set), reads admission at most once, and never opens |
| 81 | // admission, creates a registration or changes canonical state. |
| 82 | func RequireAppMutation(ctx context.Context, s release.Store, account, app, appID string) error { |
| 83 | if err := ctx.Err(); err != nil { |
| 84 | return err |
| 85 | } |
| 86 | if appID == "" { |
| 87 | return ErrPending |
| 88 | } |
| 89 | permits, _ := ctx.Value(appMutationPermitKey{}).([]*AppMutationPermit) |
| 90 | var admission *AppAdmission |
| 91 | var readErr error |
| 92 | for _, p := range permits { |
| 93 | if p == nil || p.sqlOnly || !p.live.Load() || p.account != account || p.app != app || p.appID != appID { |
| 94 | continue |
| 95 | } |
| 96 | if admission == nil && readErr == nil { |
| 97 | a, err := ReadAdmission(ctx, s, account, app, appID) |
| 98 | if err != nil { |
| 99 | readErr = err |
| 100 | } else { |
| 101 | admission = &a.Value |
| 102 | } |
| 103 | } |
| 104 | if readErr != nil { |
| 105 | // A failed read cannot validate any permit for this app. |
| 106 | return readErr |
| 107 | } |
| 108 | if p.valid(*admission) && p.live.Load() { |
| 109 | return nil |
| 110 | } |
| 111 | } |
| 112 | return ErrPending |
| 113 | } |
| 114 | |
| 115 | // LivePermits returns every permit in ctx that is valid now, each validated |
| 116 | // against a fresh admission read. The SQL boundary loads exactly this list. |
| 117 | func LivePermits(ctx context.Context, s release.Store) ([]PermitScope, error) { |
| 118 | if err := ctx.Err(); err != nil { |
| 119 | return nil, err |
| 120 | } |
| 121 | permits, _ := ctx.Value(appMutationPermitKey{}).([]*AppMutationPermit) |
| 122 | type appKey struct{ account, app, appID string } |
| 123 | cache := map[appKey]*AppAdmission{} |
| 124 | var out []PermitScope |
| 125 | for _, p := range permits { |
| 126 | if p == nil || p.sqlOnly || !p.live.Load() { |
| 127 | continue |
| 128 | } |
| 129 | k := appKey{p.account, p.app, p.appID} |
| 130 | a, ok := cache[k] |
| 131 | if !ok { |
| 132 | obs, err := ReadAdmission(ctx, s, p.account, p.app, p.appID) |
| 133 | if err != nil { |
| 134 | return nil, err |
| 135 | } |
| 136 | a = &obs.Value |
| 137 | cache[k] = a |
| 138 | } |
| 139 | if p.valid(*a) { |
| 140 | out = append(out, PermitScope{Account: p.account, App: p.app, AppID: p.appID, Epoch: p.epoch, Operation: p.operation, Effect: p.effect}) |
| 141 | } |
| 142 | } |
| 143 | return out, nil |
| 144 | } |
| 145 | |
| 146 | // PreparedSQLPermit contains only process-local authority. Preparing it does |
| 147 | // the last remote read, before SQLite is acquired. Its deadline uses Go's |
| 148 | // monotonic time component; changing wall clock cannot extend it. |
| 149 | type PreparedSQLPermit struct { |
| 150 | settle func(string, error) error |
| 151 | permit *AppMutationPermit |
| 152 | ctx context.Context |
| 153 | deadline time.Time |
| 154 | scope PermitScope |
| 155 | receipt SQLReceiptScope |
| 156 | } |
| 157 | |
| 158 | func PrepareSQLPermit(ctx context.Context, s release.Store, account, app, appID string) (*PreparedSQLPermit, error) { |
| 159 | if err := ctx.Err(); err != nil { |
| 160 | return nil, err |
| 161 | } |
| 162 | a, err := ReadAdmission(ctx, s, account, app, appID) |
| 163 | if err != nil { |
| 164 | return nil, err |
| 165 | } |
| 166 | permits, _ := ctx.Value(appMutationPermitKey{}).([]*AppMutationPermit) |
| 167 | for _, p := range permits { |
| 168 | if p == nil || p.account != account || p.app != app || p.appID != appID || !p.valid(a.Value) { |
| 169 | continue |
| 170 | } |
| 171 | sc := PermitScope{Account: account, App: app, AppID: appID, Epoch: p.epoch, Operation: p.operation, Effect: p.effect} |
| 172 | q := &PreparedSQLPermit{permit: p, ctx: ctx, deadline: time.Now().Add(SQLTransactionLifetime), scope: sc} |
| 173 | e := a.Value.Effects[p.effect] |
| 174 | if e.Action == sqlIngestAction { |
| 175 | q.receipt = SQLReceiptScope{Database: e.Inputs["sql_database"], Account: account, App: app, AppID: appID, Epoch: p.epoch, Operation: p.operation, Effect: p.effect, Owner: p.owner} |
| 176 | } |
| 177 | return q, nil |
| 178 | } |
| 179 | return nil, ErrPending |
| 180 | } |
| 181 | func (p *PreparedSQLPermit) Check(ctx context.Context) error { |
| 182 | if p == nil || p.permit == nil || !p.permit.live.Load() { |
| 183 | return ErrPending |
| 184 | } |
| 185 | if err := p.ctx.Err(); err != nil { |
| 186 | return err |
| 187 | } |
| 188 | if err := ctx.Err(); err != nil { |
| 189 | return err |
| 190 | } |
| 191 | if !time.Now().Before(p.deadline) { |
| 192 | return context.DeadlineExceeded |
| 193 | } |
| 194 | return nil |
| 195 | } |
| 196 | func (p *PreparedSQLPermit) Receipt() SQLReceiptScope { return p.receipt } |
| 197 | |
| 198 | // BindTransaction gives SQL running under a lifecycle/mail effect its own |
| 199 | // receipt-bound child. Specialized recovery cannot complete the parent |
| 200 | // operation while that child is issued. Call before taking any SQL handle. |
| 201 | func (p *PreparedSQLPermit) BindTransaction(database string) error { |
| 202 | if err := p.Check(p.ctx); err != nil { |
| 203 | return err |
| 204 | } |
| 205 | if p.receipt.Database != "" { |
| 206 | if p.receipt.Database != database { |
| 207 | return ErrIdentity |
| 208 | } |
| 209 | return nil |
| 210 | } |
| 211 | if p.permit.writer == nil || database == "" { |
| 212 | return ErrContract |
| 213 | } |
| 214 | var nonce [24]byte |
| 215 | if _, err := rand.Read(nonce[:]); err != nil { |
| 216 | return err |
| 217 | } |
| 218 | id := hex.EncodeToString(nonce[:]) |
| 219 | old := p.permit |
| 220 | runner := EffectRunner{w: old.writer, account: old.account, app: old.app, appID: old.appID, operation: old.operation, owner: old.owner, now: old.now, exclusive: old.exclusive} |
| 221 | ctx, settle, err := runner.Issue(p.ctx, Effect{ID: id, Resource: "app-sql:" + id, Action: sqlIngestAction, Inputs: map[string]string{"sql_database": database, "sql_parent": old.effect, "sql_recover_after": old.now().Add(SQLTransactionLifetime + RegistrationSkew).UTC().Format(time.RFC3339Nano)}}) |
| 222 | if err != nil { |
| 223 | return err |
| 224 | } |
| 225 | permits, _ := ctx.Value(appMutationPermitKey{}).([]*AppMutationPermit) |
| 226 | p.permit = permits[len(permits)-1] |
| 227 | p.ctx = ctx |
| 228 | p.settle = settle |
| 229 | p.receipt = SQLReceiptScope{Database: database, Account: old.account, App: old.app, AppID: old.appID, Operation: old.operation, Effect: id, Epoch: old.epoch, Owner: old.owner} |
| 230 | return nil |
| 231 | } |
| 232 | |
| 233 | // FinishTransaction is called only after SQL handles are revoked and joined. |
| 234 | func (p *PreparedSQLPermit) FinishTransaction(err error) error { |
| 235 | if err == nil { |
| 236 | p.permit.sqlJoined.Store(true) |
| 237 | } |
| 238 | if p.settle == nil { |
| 239 | return nil |
| 240 | } |
| 241 | settle := p.settle |
| 242 | p.settle = nil |
| 243 | return settle("sql-joined", err) |
| 244 | } |
| 245 | |
amadan