| 1 | package registry |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "reflect" |
| 11 | "strings" |
| 12 | ) |
| 13 | |
| 14 | // The import journal (seeded-instances spec §5.7): one row per host this |
| 15 | // box has swapped for an import, holding the complete original route, so |
| 16 | // the box always knows what it replaced, whatever happened to the record. |
| 17 | const ( |
| 18 | JournalSwapped = "swapped" |
| 19 | JournalRollingBack = "rolling-back" |
| 20 | JournalCleanupPending = "cleanup-pending" |
| 21 | ) |
| 22 | |
| 23 | // ImportGateClosed is the import gate a rollback installs: it matches no |
| 24 | // token's hash (seeds.ClosedGate holds the same value). |
| 25 | const ImportGateClosed = "closed" |
| 26 | |
| 27 | // ErrImportJournal is a journal transaction whose precondition did not |
| 28 | // hold: no row or another seed's, a row already there, a changed route. |
| 29 | var ErrImportJournal = errors.New("registry: the import journal does not allow this") |
| 30 | |
| 31 | // What the host's own env file (<host>.env, at the nested path the swap |
| 32 | // gives the import's row) held when the swap ran, before any launch of the |
| 33 | // import: ImportJournal.EnvBefore. The platform mints the instance key into |
| 34 | // that file at the import's first launch, create-only, and a key there |
| 35 | // outranks the app's configuration, so a rollback has to know whether the |
| 36 | // key it finds is the import's own (it retires it) or was there before (it |
| 37 | // never touches it). "" is a journal row written before this was recorded: |
| 38 | // nothing is known, so nothing is retired. |
| 39 | const ( |
| 40 | EnvBeforeAbsent = "absent" // no env file: every byte in it later is the import's |
| 41 | EnvBeforeNoKey = "no-key" // an env file with no instance key: a key in it later is the import's |
| 42 | EnvBeforeKey = "key" // an env file that already carried a key: the import minted none |
| 43 | ) |
| 44 | |
| 45 | // ImportJournal is one import_journal row. Original is the route the swap |
| 46 | // replaced, exactly as it was stored. EnvBefore is one of the EnvBefore |
| 47 | // values, or "" for unknown. |
| 48 | type ImportJournal struct { |
| 49 | Host string |
| 50 | SeedID string |
| 51 | Original Route |
| 52 | RecordKey string |
| 53 | State string |
| 54 | FenceGen uint64 |
| 55 | At int64 |
| 56 | EnvBefore string |
| 57 | } |
| 58 | |
| 59 | func (r *Registry) migrateImportJournal() error { |
| 60 | if _, err := r.sql.Exec(`CREATE TABLE IF NOT EXISTS import_journal ( |
| 61 | host TEXT PRIMARY KEY, |
| 62 | seed_id TEXT NOT NULL, |
| 63 | original TEXT NOT NULL, |
| 64 | record_key TEXT NOT NULL DEFAULT '', |
| 65 | state TEXT NOT NULL, |
| 66 | fence_gen INTEGER NOT NULL DEFAULT 0, |
| 67 | at INTEGER NOT NULL DEFAULT 0, |
| 68 | env_before TEXT NOT NULL DEFAULT '' |
| 69 | );`); err != nil { |
| 70 | return err |
| 71 | } |
| 72 | // A journal table from before env_before gains it; its rows read "". |
| 73 | if _, err := r.sql.Exec(`ALTER TABLE import_journal ADD COLUMN env_before TEXT NOT NULL DEFAULT ''`); err != nil && |
| 74 | !strings.Contains(err.Error(), "duplicate column") { |
| 75 | return err |
| 76 | } |
| 77 | return nil |
| 78 | } |
| 79 | |
| 80 | type queryer interface { |
| 81 | QueryRow(query string, args ...any) *sql.Row |
| 82 | } |
| 83 | |
| 84 | type txer interface { |
| 85 | queryer |
| 86 | Exec(query string, args ...any) (sql.Result, error) |
| 87 | } |
| 88 | |
| 89 | const journalCols = `host, seed_id, original, record_key, state, fence_gen, at, env_before` |
| 90 | |
| 91 | func scanJournalRow(scan func(...any) error) (ImportJournal, error) { |
| 92 | var j ImportJournal |
| 93 | var orig string |
| 94 | if err := scan(&j.Host, &j.SeedID, &orig, &j.RecordKey, &j.State, &j.FenceGen, &j.At, &j.EnvBefore); err != nil { |
| 95 | return j, err |
| 96 | } |
| 97 | if err := json.Unmarshal([]byte(orig), &j.Original); err != nil { |
| 98 | return j, fmt.Errorf("registry: import journal %s: original row: %w", j.Host, err) |
| 99 | } |
| 100 | return j, nil |
| 101 | } |
| 102 | |
| 103 | func scanJournal(q queryer, host string) (*ImportJournal, error) { |
| 104 | j, err := scanJournalRow(q.QueryRow(`SELECT `+journalCols+` FROM import_journal WHERE host = ?`, host).Scan) |
| 105 | if errors.Is(err, sql.ErrNoRows) { |
| 106 | return nil, nil |
| 107 | } |
| 108 | if err != nil { |
| 109 | return nil, err |
| 110 | } |
| 111 | return &j, nil |
| 112 | } |
| 113 | |
| 114 | // ImportJournal returns host's journal row, or nil when there is none. |
| 115 | func (r *Registry) ImportJournal(host string) (*ImportJournal, error) { |
| 116 | if r.remote != nil { |
| 117 | return nil, errRemoteMethod |
| 118 | } |
| 119 | return scanJournal(r.sql, host) |
| 120 | } |
| 121 | |
| 122 | // ImportJournals returns every journal row, ordered by host. |
| 123 | func (r *Registry) ImportJournals() ([]ImportJournal, error) { |
| 124 | if r.remote != nil { |
| 125 | return nil, errRemoteMethod |
| 126 | } |
| 127 | rows, err := r.sql.Query(`SELECT ` + journalCols + ` FROM import_journal ORDER BY host`) |
| 128 | if err != nil { |
| 129 | return nil, err |
| 130 | } |
| 131 | defer rows.Close() |
| 132 | var out []ImportJournal |
| 133 | for rows.Next() { |
| 134 | j, err := scanJournalRow(rows.Scan) |
| 135 | if err != nil { |
| 136 | return nil, err |
| 137 | } |
| 138 | out = append(out, j) |
| 139 | } |
| 140 | return out, rows.Err() |
| 141 | } |
| 142 | |
| 143 | // ImportJournalHosts returns the hosts the journal holds a row for, in any |
| 144 | // state, ordered. It is what a box says about its imports in the claim |
| 145 | // activation census (hostops.CollectCensusSeeds). Unlike the journal's |
| 146 | // other methods it answers on a handle that proxies to root's registry |
| 147 | // child too, so a collector that runs as root can ask. |
| 148 | func (r *Registry) ImportJournalHosts() ([]string, error) { |
| 149 | if r.remote != nil { |
| 150 | out, err := r.remote.call(nil, remoteRequest{Method: "ImportJournalHosts"}, nil) |
| 151 | if err != nil { |
| 152 | return nil, err |
| 153 | } |
| 154 | // The transport drops an empty list; "none" is not "did not say". |
| 155 | return append([]string{}, out.Hosts...), nil |
| 156 | } |
| 157 | rows, err := r.sql.Query(`SELECT host FROM import_journal ORDER BY host`) |
| 158 | if err != nil { |
| 159 | return nil, err |
| 160 | } |
| 161 | defer rows.Close() |
| 162 | out := []string{} |
| 163 | for rows.Next() { |
| 164 | var h string |
| 165 | if err := rows.Scan(&h); err != nil { |
| 166 | return nil, err |
| 167 | } |
| 168 | out = append(out, h) |
| 169 | } |
| 170 | return out, rows.Err() |
| 171 | } |
| 172 | |
| 173 | // hashImportJournal folds the journal into SnapshotFingerprint's digest: |
| 174 | // the journal is box-local state nothing in the bucket can rebuild, and a |
| 175 | // snapshot that kept a deleted cleanup-pending row would, restored cold, |
| 176 | // resume a finished rollback's bucket cleanup over a later instance on the |
| 177 | // same host. Every column, in host order, under its own section marker. An |
| 178 | // empty journal writes nothing, so a box with no import in flight digests |
| 179 | // exactly as it did before the journal existed. |
| 180 | func hashImportJournal(db *sql.DB, h io.Writer) error { |
| 181 | rows, err := db.Query(`SELECT ` + journalCols + ` FROM import_journal ORDER BY host`) |
| 182 | if err != nil { |
| 183 | return err |
| 184 | } |
| 185 | defer rows.Close() |
| 186 | first := true |
| 187 | for rows.Next() { |
| 188 | var host, seedID, orig, recordKey, state, envBefore string |
| 189 | var fenceGen uint64 |
| 190 | var at int64 |
| 191 | if err := rows.Scan(&host, &seedID, &orig, &recordKey, &state, &fenceGen, &at, &envBefore); err != nil { |
| 192 | return err |
| 193 | } |
| 194 | if first { |
| 195 | fmt.Fprint(h, "\x1dimport_journal\x1e") |
| 196 | first = false |
| 197 | } |
| 198 | fmt.Fprintf(h, "%q\x1f%q\x1f%q\x1f%q\x1f%q\x1f%d\x1f%d", host, seedID, orig, recordKey, state, fenceGen, at) |
| 199 | // env_before is hashed only when set, tagged, so a row written |
| 200 | // before the column existed digests as it did. |
| 201 | if envBefore != "" { |
| 202 | fmt.Fprintf(h, "\x1fenv_before=%q", envBefore) |
| 203 | } |
| 204 | fmt.Fprint(h, "\x1e") |
| 205 | } |
| 206 | return rows.Err() |
| 207 | } |
| 208 | |
| 209 | // inTx runs fn in one BEGIN IMMEDIATE transaction at synchronous=FULL (the |
| 210 | // claim transactions' form). IMMEDIATE takes the write lock before the first |
| 211 | // read, so another process's write waits on busy_timeout instead of failing |
| 212 | // the upgrade; FULL makes the commit durable before the caller acts on it |
| 213 | // (the swap's record CAS follows it). The registry's pool holds one |
| 214 | // connection (Open), so in-process writers serialise on it. |
| 215 | // |
| 216 | // The import journal, the fence generation source and the seed column are |
| 217 | // the agent's own: the agent opens the registry itself (Open), and no root |
| 218 | // pass reads or writes them. A handle that proxies to root's registry child |
| 219 | // (OpenRemote) has no dispatch for them, so each refuses there |
| 220 | // (errRemoteMethod) rather than touch a database it does not hold. |
| 221 | func (r *Registry) inTx(fn func(tx txer) error) error { |
| 222 | if r.remote != nil { |
| 223 | return errRemoteMethod |
| 224 | } |
| 225 | tx, err := r.beginClaimTransaction(context.Background(), true) |
| 226 | if err != nil { |
| 227 | return err |
| 228 | } |
| 229 | defer tx.Rollback() |
| 230 | if err := fn(tx); err != nil { |
| 231 | return err |
| 232 | } |
| 233 | return tx.Commit() |
| 234 | } |
| 235 | |
| 236 | func getRouteTx(tx txer, host string) (*Route, error) { |
| 237 | t, err := scanRoute(tx.QueryRow(`SELECT `+routeCols+` FROM routes WHERE host = ?`, host).Scan) |
| 238 | if errors.Is(err, sql.ErrNoRows) { |
| 239 | return nil, nil |
| 240 | } |
| 241 | if err != nil { |
| 242 | return nil, err |
| 243 | } |
| 244 | return &t, nil |
| 245 | } |
| 246 | |
| 247 | // insertRouteTx inserts t exactly as given (no Upsert defaults). |
| 248 | func insertRouteTx(tx txer, t Route) error { |
| 249 | _, err := tx.Exec(`INSERT INTO routes (`+routeCols+`) VALUES (`+placeholders(routeColCount)+`)`, routeArgs(t)...) |
| 250 | return err |
| 251 | } |
| 252 | |
| 253 | // IsImportRow reports whether rt is the row j's swap put in place: the |
| 254 | // journal's host, an instance row a record provisioned, still gated (by |
| 255 | // the verify hash, or the closed sentinel once a rollback began), and |
| 256 | // whose seed column names j's seed. The rollback acts only on this row |
| 257 | // (spec §5.6): any other row at the host belongs to someone else, such as |
| 258 | // a later instance when a stale journal row came back from a snapshot. |
| 259 | func IsImportRow(rt Route, j ImportJournal) bool { |
| 260 | return j.SeedID != "" && rt.Host == j.Host && rt.Kind == "instance" && rt.Provisioned && |
| 261 | rt.ImportGate != "" && seedColumnID(rt.SeedSource) == j.SeedID |
| 262 | } |
| 263 | |
| 264 | // seedColumnID is the seed id a seed_source column names, "" for none or |
| 265 | // one that does not parse. The column is seeds.RowSeed's JSON; the |
| 266 | // registry reads only its id, since seeds imports this package. |
| 267 | func seedColumnID(col string) string { |
| 268 | if col == "" { |
| 269 | return "" |
| 270 | } |
| 271 | var v struct { |
| 272 | ID string `json:"id"` |
| 273 | } |
| 274 | if err := json.Unmarshal([]byte(col), &v); err != nil { |
| 275 | return "" |
| 276 | } |
| 277 | return v.ID |
| 278 | } |
| 279 | |
| 280 | // SwapForImport is the swap's one transaction (spec §5.4, codex 14): with |
| 281 | // no journal row for the host and the stored route still exactly |
| 282 | // j.Original, it replaces the route with next (Upsert's defaults applied) |
| 283 | // and writes the journal row at swapped. next must be the import's row as |
| 284 | // IsImportRow reads it (the same host, gated, naming the journal's seed), |
| 285 | // and the journal must name its record: a journal row with no record key |
| 286 | // can never be decided, so it is never written. The original must be one |
| 287 | // the journal can put back: not claim-owned (claims are off for imports, |
| 288 | // spec §3.4), not held by an operation, not itself gated, and unchanged by |
| 289 | // its JSON round trip. j.EnvBefore is stored as given: one of the EnvBefore |
| 290 | // values, or "" when the caller could not say. |
| 291 | func (r *Registry) SwapForImport(j ImportJournal, next Route) error { |
| 292 | if j.Host == "" || j.SeedID == "" || j.RecordKey == "" || j.Original.Host != j.Host || !IsImportRow(normalizeRoute(next), j) { |
| 293 | return fmt.Errorf("%w: the swap for %s is not well formed", ErrImportJournal, j.Host) |
| 294 | } |
| 295 | switch j.EnvBefore { |
| 296 | case "", EnvBeforeAbsent, EnvBeforeNoKey, EnvBeforeKey: |
| 297 | default: |
| 298 | return fmt.Errorf("%w: the swap for %s is not well formed", ErrImportJournal, j.Host) |
| 299 | } |
| 300 | if j.Original.ClaimOwnership != (ClaimOwnership{}) || j.Original.HoldOp != "" || j.Original.ImportGate != "" { |
| 301 | return fmt.Errorf("%w: the route at %s is not one an import replaces", ErrImportJournal, j.Host) |
| 302 | } |
| 303 | orig, err := json.Marshal(j.Original) |
| 304 | if err != nil { |
| 305 | return err |
| 306 | } |
| 307 | var back Route |
| 308 | if err := json.Unmarshal(orig, &back); err != nil || !reflect.DeepEqual(back, j.Original) { |
| 309 | return fmt.Errorf("%w: the route at %s would not restore exactly", ErrImportJournal, j.Host) |
| 310 | } |
| 311 | return r.inTx(func(tx txer) error { |
| 312 | if cur, err := scanJournal(tx, j.Host); err != nil { |
| 313 | return err |
| 314 | } else if cur != nil { |
| 315 | return fmt.Errorf("%w: %s already has a journal row for seed %s", ErrImportJournal, j.Host, cur.SeedID) |
| 316 | } |
| 317 | stored, err := getRouteTx(tx, j.Host) |
| 318 | if err != nil { |
| 319 | return err |
| 320 | } |
| 321 | if stored == nil || !reflect.DeepEqual(*stored, j.Original) { |
| 322 | return fmt.Errorf("%w: the route at %s changed since it was checked", ErrImportJournal, j.Host) |
| 323 | } |
| 324 | if _, err := tx.Exec(`DELETE FROM routes WHERE host = ?`, j.Host); err != nil { |
| 325 | return err |
| 326 | } |
| 327 | if err := insertRouteTx(tx, normalizeRoute(next)); err != nil { |
| 328 | return err |
| 329 | } |
| 330 | _, err = tx.Exec(`INSERT INTO import_journal (`+journalCols+`) VALUES (?,?,?,?,?,0,?,?)`, |
| 331 | j.Host, j.SeedID, string(orig), j.RecordKey, JournalSwapped, j.At, j.EnvBefore) |
| 332 | return err |
| 333 | }) |
| 334 | } |
| 335 | |
| 336 | // journalFor is host's journal row, which must name seedID and be in one of |
| 337 | // states. |
| 338 | func journalFor(tx txer, host, seedID string, states ...string) (*ImportJournal, error) { |
| 339 | j, err := scanJournal(tx, host) |
| 340 | if err != nil { |
| 341 | return nil, err |
| 342 | } |
| 343 | if j == nil || j.SeedID != seedID { |
| 344 | return nil, fmt.Errorf("%w: no journal row for %s seed %s", ErrImportJournal, host, seedID) |
| 345 | } |
| 346 | for _, s := range states { |
| 347 | if j.State == s { |
| 348 | return j, nil |
| 349 | } |
| 350 | } |
| 351 | return nil, fmt.Errorf("%w: %s's journal is %s", ErrImportJournal, host, j.State) |
| 352 | } |
| 353 | |
| 354 | // importRowTx is host's row when it is j's import row, nil when the host |
| 355 | // has no row, and ErrImportJournal when the row there is anyone else's. |
| 356 | func importRowTx(tx txer, j *ImportJournal) (*Route, error) { |
| 357 | cur, err := getRouteTx(tx, j.Host) |
| 358 | if err != nil { |
| 359 | return nil, err |
| 360 | } |
| 361 | if cur != nil && !IsImportRow(*cur, *j) { |
| 362 | return nil, fmt.Errorf("%w: the route at %s is not import %s's", ErrImportJournal, j.Host, j.SeedID) |
| 363 | } |
| 364 | return cur, nil |
| 365 | } |
| 366 | |
| 367 | // CloseImportGate is the rollback's step 1 (spec §5.6): the journal says |
| 368 | // rolling-back with the fence generation, and the gate refuses even verify |
| 369 | // requests, in one transaction. Run again on a row already rolling-back, it |
| 370 | // re-closes the gate and keeps the stored generation (fenceGen is ignored): |
| 371 | // step 2 recognises this rollback's own fence by that generation, so the |
| 372 | // caller reads FenceGen back from the journal rather than trusting its own. |
| 373 | // |
| 374 | // It refuses when the host's row is not the import's (IsImportRow), so it |
| 375 | // never gates a row the import does not own. A host with no row is closed |
| 376 | // all the same: the journal moves on, and there is no gate to set. |
| 377 | func (r *Registry) CloseImportGate(host, seedID string, fenceGen uint64) error { |
| 378 | return r.inTx(func(tx txer) error { |
| 379 | j, err := journalFor(tx, host, seedID, JournalSwapped, JournalRollingBack) |
| 380 | if err != nil { |
| 381 | return err |
| 382 | } |
| 383 | if _, err := importRowTx(tx, j); err != nil { |
| 384 | return err |
| 385 | } |
| 386 | if j.State == JournalSwapped { |
| 387 | if _, err := tx.Exec(`UPDATE import_journal SET state = ?, fence_gen = ? WHERE host = ?`, JournalRollingBack, fenceGen, host); err != nil { |
| 388 | return err |
| 389 | } |
| 390 | } |
| 391 | _, err = tx.Exec(`UPDATE routes SET import_gate = ? WHERE host = ?`, ImportGateClosed, host) |
| 392 | return err |
| 393 | }) |
| 394 | } |
| 395 | |
| 396 | // RestoreImportRoute is the rollback's step 4 (spec §5.6, codex R3-1): the |
| 397 | // original route back exactly as it was stored, and the journal at |
| 398 | // cleanup-pending. |
| 399 | // |
| 400 | // Only the import's own row is replaced (IsImportRow). Any other row at |
| 401 | // the host is refused, and nothing changes: it is not the import's to |
| 402 | // remove. A host with no row at all (something removed the import's row) |
| 403 | // gets no route back, since whatever removed it decided the host should |
| 404 | // not be served; the journal still moves to cleanup-pending, so the |
| 405 | // rollback carries on with its cleanup. |
| 406 | func (r *Registry) RestoreImportRoute(host, seedID string) error { |
| 407 | return r.inTx(func(tx txer) error { |
| 408 | j, err := journalFor(tx, host, seedID, JournalRollingBack) |
| 409 | if err != nil { |
| 410 | return err |
| 411 | } |
| 412 | cur, err := importRowTx(tx, j) |
| 413 | if err != nil { |
| 414 | return err |
| 415 | } |
| 416 | if cur != nil { |
| 417 | if _, err := tx.Exec(`DELETE FROM routes WHERE host = ?`, host); err != nil { |
| 418 | return err |
| 419 | } |
| 420 | if err := insertRouteTx(tx, j.Original); err != nil { |
| 421 | return err |
| 422 | } |
| 423 | } |
| 424 | _, err = tx.Exec(`UPDATE import_journal SET state = ? WHERE host = ?`, JournalCleanupPending, host) |
| 425 | return err |
| 426 | }) |
| 427 | } |
| 428 | |
| 429 | // CommitImport clears the gate and retires the journal in one transaction |
| 430 | // (spec §5.4 "Commit"). |
| 431 | func (r *Registry) CommitImport(host, seedID string) error { |
| 432 | return r.inTx(func(tx txer) error { |
| 433 | if _, err := journalFor(tx, host, seedID, JournalSwapped); err != nil { |
| 434 | return err |
| 435 | } |
| 436 | if _, err := tx.Exec(`UPDATE routes SET import_gate = '' WHERE host = ?`, host); err != nil { |
| 437 | return err |
| 438 | } |
| 439 | _, err := tx.Exec(`DELETE FROM import_journal WHERE host = ?`, host) |
| 440 | return err |
| 441 | }) |
| 442 | } |
| 443 | |
| 444 | // DeleteImportJournal is the rollback's step 7. |
| 445 | func (r *Registry) DeleteImportJournal(host, seedID string) error { |
| 446 | return r.inTx(func(tx txer) error { |
| 447 | if _, err := journalFor(tx, host, seedID, JournalCleanupPending); err != nil { |
| 448 | return err |
| 449 | } |
| 450 | _, err := tx.Exec(`DELETE FROM import_journal WHERE host = ?`, host) |
| 451 | return err |
| 452 | }) |
| 453 | } |
| 454 | |
amadan