| 1 | package hostops |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "encoding/hex" |
| 7 | "errors" |
| 8 | "time" |
| 9 | ) |
| 10 | |
| 11 | const RegistrationIngest = "ingest" |
| 12 | const SQLTransactionLifetime = 30 * time.Second |
| 13 | const sqlIngestAction = "sql-ingest" |
| 14 | |
| 15 | // IngestHandle has no lifecycle or generic runner door. One registration |
| 16 | // issues exactly one SQL batch; the effect remains a drain obligation forever |
| 17 | // until its issuer joins it or its own SQL database durably fences it. |
| 18 | type IngestHandle struct{ start *StartHandle } |
| 19 | |
| 20 | func (w *ControlWriter) RegisterIngest(ctx context.Context, account, app, appID, id string, owner RegistrationOwner) (*IngestHandle, error) { |
| 21 | if owner.Console == "" || !validRegistrationOwner(owner) { |
| 22 | return nil, ErrIdentity |
| 23 | } |
| 24 | if err := w.purgeAbsent(ctx, account, app); err != nil { |
| 25 | return nil, err |
| 26 | } |
| 27 | a, err := w.read(ctx, account, app, appID) |
| 28 | if err != nil { |
| 29 | return nil, err |
| 30 | } |
| 31 | reg := Registration{Kind: RegistrationIngest, Owner: owner, AppID: appID, Epoch: a.Value.Epoch} |
| 32 | _, err = w.retryAdmission(ctx, account, app, appID, a, func(v *AppAdmission) error { return registerStart(v, id, reg, w.now()) }) |
| 33 | if err != nil { |
| 34 | return nil, err |
| 35 | } |
| 36 | return &IngestHandle{&StartHandle{w: w, account: account, app: app, appID: appID, id: id, owner: owner, epoch: reg.Epoch, kind: RegistrationIngest}}, nil |
| 37 | } |
| 38 | |
| 39 | // SQLIngestRunner is implemented only by the configured mutation boundary. |
| 40 | // Database identifies the persistent local receipt namespace, never a path. |
| 41 | type SQLIngestRunner interface { |
| 42 | RunSQLIngest(context.Context, string, string, string, string, func(context.Context) error) error |
| 43 | } |
| 44 | |
| 45 | func (b *mutationBoundary) RunSQLIngest(ctx context.Context, account, app, action, database string, f func(context.Context) error) error { |
| 46 | if b.control == nil || b.localAppID == nil || f == nil || database == "" || action == "" { |
| 47 | return ErrContract |
| 48 | } |
| 49 | appID, err := b.localAppID(ctx, account, app) |
| 50 | if err != nil { |
| 51 | return err |
| 52 | } |
| 53 | if appID == "" { |
| 54 | return ErrIdentity |
| 55 | } |
| 56 | // Existing lifecycle and mail operations retain their own authority. |
| 57 | if RequireAppMutation(ctx, b.raw, account, app, appID) == nil { |
| 58 | return f(ctx) |
| 59 | } |
| 60 | inherited, _ := ctx.Value(appMutationPermitKey{}).([]*AppMutationPermit) |
| 61 | for _, p := range inherited { |
| 62 | if p != nil && p.account == account && p.app == app { |
| 63 | return ErrPending |
| 64 | } |
| 65 | } |
| 66 | var nonce [24]byte |
| 67 | if _, err = rand.Read(nonce[:]); err != nil { |
| 68 | return err |
| 69 | } |
| 70 | id := hex.EncodeToString(nonce[:]) |
| 71 | owner := b.owner |
| 72 | owner.Controller = id |
| 73 | h, err := b.control.RegisterIngest(ctx, account, app, appID, id, owner) |
| 74 | if err != nil { |
| 75 | return err |
| 76 | } |
| 77 | pctx, settle, err := h.start.Runner(b.control.now).Issue(ctx, Effect{ID: id, Resource: "app-sql:" + id, Action: sqlIngestAction, Inputs: map[string]string{"sql_database": database, "sql_action": action}}) |
| 78 | if err != nil { |
| 79 | return err |
| 80 | } |
| 81 | // Panic is uncertain too. A process crash leaves issued in the bucket. |
| 82 | var callErr error |
| 83 | func() { |
| 84 | defer func() { |
| 85 | if p := recover(); p != nil { |
| 86 | _ = settle("", errors.New("SQL callback panicked")) |
| 87 | panic(p) |
| 88 | } |
| 89 | }() |
| 90 | callErr = f(pctx) |
| 91 | }() |
| 92 | settleErr := callErr |
| 93 | permits, _ := pctx.Value(appMutationPermitKey{}).([]*AppMutationPermit) |
| 94 | for _, p := range permits { |
| 95 | if p.effect == id && p.sqlJoined.Load() { |
| 96 | settleErr = nil |
| 97 | } |
| 98 | } |
| 99 | if err = settle("sql-joined", settleErr); err != nil { |
| 100 | return err |
| 101 | } |
| 102 | return errors.Join(callErr, b.control.finishIngest(context.WithoutCancel(ctx), h.start)) |
| 103 | } |
| 104 | |
| 105 | func (w *ControlWriter) finishIngest(ctx context.Context, h *StartHandle) error { |
| 106 | _, err := w.retryAdmission(ctx, h.account, h.app, h.appID, AdmissionObserved{}, func(v *AppAdmission) error { |
| 107 | reg, ok := v.Registrations[h.id] |
| 108 | if !ok { |
| 109 | return nil |
| 110 | } |
| 111 | if reg.Kind != RegistrationIngest || reg.Owner != h.owner || !effectsSettled(v, h.id) { |
| 112 | return ErrPending |
| 113 | } |
| 114 | delete(v.Registrations, h.id) |
| 115 | return nil |
| 116 | }) |
| 117 | return err |
| 118 | } |
| 119 | |
| 120 | // SQLRecovery is a trusted console-local adapter. FenceSQL must commit a |
| 121 | // durable fence in the exact original receipt database, joining SQLite's |
| 122 | // writer lock before returning. A missing database is never an empty receipt. |
| 123 | type SQLRecovery interface { |
| 124 | FenceSQL(context.Context, SQLReceiptScope) error |
| 125 | } |
| 126 | |
| 127 | // SQLOutcomeRecovery is a SQLRecovery that also reports whether the fence |
| 128 | // found the original transaction's committed receipt. Recovery then records |
| 129 | // sql-committed instead of sql-fenced. |
| 130 | type SQLOutcomeRecovery interface { |
| 131 | SQLRecovery |
| 132 | FenceSQLOutcome(context.Context, SQLReceiptScope) (committed bool, err error) |
| 133 | } |
| 134 | |
| 135 | type SQLReceiptScope struct { |
| 136 | Database string `json:"database"` |
| 137 | Account string `json:"account"` |
| 138 | App string `json:"app"` |
| 139 | AppID string `json:"app_id"` |
| 140 | Operation string `json:"operation"` |
| 141 | Effect string `json:"effect"` |
| 142 | Epoch uint64 `json:"epoch"` |
| 143 | Owner RegistrationOwner `json:"owner"` |
| 144 | } |
| 145 | |
| 146 | // RecoverIngest joins/fences SQL first, releases the connection, then uses |
| 147 | // the runner's one settlement implementation. It never infers death or |
| 148 | // not-applied from lease age, a missing response, or a missing receipt. |
| 149 | func (w *ControlWriter) RecoverIngest(ctx context.Context, account, app, appID, id string, local SQLRecovery) error { |
| 150 | if local == nil { |
| 151 | return ErrContract |
| 152 | } |
| 153 | a, err := w.read(ctx, account, app, appID) |
| 154 | if err != nil { |
| 155 | return err |
| 156 | } |
| 157 | reg, ok := a.Value.Registrations[id] |
| 158 | if !ok { |
| 159 | return nil |
| 160 | } |
| 161 | e, ok := a.Value.Effects[id] |
| 162 | if reg.Kind != RegistrationIngest || !ok || e.OperationID != id || e.AppID != appID || e.Owner != reg.Owner || e.Action != sqlIngestAction || e.Inputs["sql_database"] == "" { |
| 163 | return ErrIdentity |
| 164 | } |
| 165 | h := &StartHandle{w: w, account: account, app: app, appID: appID, id: id, owner: reg.Owner, epoch: reg.Epoch, kind: RegistrationIngest} |
| 166 | if e.State == "complete" { |
| 167 | return w.finishIngest(ctx, h) |
| 168 | } |
| 169 | if e.State != "prepared" && e.State != "issued" && e.State != "uncertain" { |
| 170 | return ErrPending |
| 171 | } |
| 172 | if err = w.RecoverSQLTransaction(ctx, account, app, appID, e.ID, local); err != nil { |
| 173 | return err |
| 174 | } |
| 175 | return w.finishIngest(ctx, h) |
| 176 | } |
| 177 | |
| 178 | // RecoverSQLTransaction proves the one SQL child's terminal outcome without |
| 179 | // settling its parent (restore, placement, purge or non-expiring mail). |
| 180 | func (w *ControlWriter) RecoverSQLTransaction(ctx context.Context, account, app, appID, id string, local SQLRecovery) error { |
| 181 | if local == nil { |
| 182 | return ErrContract |
| 183 | } |
| 184 | a, err := w.read(ctx, account, app, appID) |
| 185 | if err != nil { |
| 186 | return err |
| 187 | } |
| 188 | e, ok := a.Value.Effects[id] |
| 189 | if !ok || e.AppID != appID || e.Action != sqlIngestAction || e.Resource != "app-sql:"+id || e.Inputs["sql_database"] == "" || !validRegistrationOwner(e.Owner) { |
| 190 | return ErrIdentity |
| 191 | } |
| 192 | if e.State == "complete" { |
| 193 | return nil |
| 194 | } |
| 195 | exclusive := a.Value.Op != nil && a.Value.Op.ID == e.OperationID |
| 196 | if !exclusive { |
| 197 | reg, ok := a.Value.Registrations[e.OperationID] |
| 198 | if !ok || reg.Owner != e.Owner || reg.AppID != appID || reg.Epoch != a.Value.Epoch { |
| 199 | return ErrPending |
| 200 | } |
| 201 | } |
| 202 | if e.State != "prepared" && e.State != "issued" && e.State != "uncertain" { |
| 203 | return ErrPending |
| 204 | } |
| 205 | sc := SQLReceiptScope{Database: e.Inputs["sql_database"], Account: account, App: app, AppID: appID, Operation: e.OperationID, Effect: e.ID, Epoch: a.Value.Epoch, Owner: e.Owner} |
| 206 | result := "sql-fenced" |
| 207 | if o, ok := local.(SQLOutcomeRecovery); ok { |
| 208 | committed, ferr := o.FenceSQLOutcome(ctx, sc) |
| 209 | if ferr != nil { |
| 210 | return ferr |
| 211 | } |
| 212 | if committed { |
| 213 | result = "sql-committed" |
| 214 | } |
| 215 | } else if err = local.FenceSQL(ctx, sc); err != nil { |
| 216 | return err |
| 217 | } |
| 218 | r := EffectRunner{w: w, account: account, app: app, appID: appID, operation: e.OperationID, owner: e.Owner, now: w.now, exclusive: exclusive} |
| 219 | _, err = r.saveSettlement(ctx, e.ID, "", "complete", result) |
| 220 | return err |
| 221 | } |
| 222 | |
| 223 | func (w *ControlWriter) sqlChildrenSettled(ctx context.Context, account, app, appID, operation string) error { |
| 224 | a, err := w.read(ctx, account, app, appID) |
| 225 | if err != nil { |
| 226 | return err |
| 227 | } |
| 228 | for _, e := range a.Value.Effects { |
| 229 | if e.OperationID == operation && e.Action == sqlIngestAction && e.State != "complete" { |
| 230 | return ErrPending |
| 231 | } |
| 232 | } |
| 233 | return nil |
| 234 | } |
| 235 | |
| 236 | func validIngestRegistration(id string, r Registration, appID string, epoch uint64) bool { |
| 237 | return id != "" && r.Kind == RegistrationIngest && r.Host == "" && r.Claim == (SavedClaim{}) && r.AppID == appID && r.Epoch == epoch && validRegistrationOwner(r.Owner) && r.Owner.Console != "" && !r.Expires.IsZero() && r.Incarnation == "" && r.CreatedAt.IsZero() |
| 238 | } |
| 239 | |
amadan