| 1 | package provision |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "regexp" |
| 9 | "strings" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/registry" |
| 13 | "github.com/carlosframework/platform/internal/release" |
| 14 | "github.com/carlosframework/platform/internal/seeds" |
| 15 | ) |
| 16 | |
| 17 | // importShape is the legacy row's shape check (spec §5.4, codex 10, R1-10): |
| 18 | // "" when the row is an addr-backed instance or service this import may |
| 19 | // take over, else what makes it not one (the column's name). |
| 20 | // |
| 21 | // Addr-backed means the edge serves the row from its addr or addrs and |
| 22 | // from nothing else. The edge's dispatch (edge.Reload) takes a tunnel, |
| 23 | // then a fleet placement, then a socket, before it reaches the addrs, and |
| 24 | // an instance flagged hibernate goes through the activator, which starts |
| 25 | // a process of its own. So each of those is refused here, beside the |
| 26 | // columns shapeMismatch refuses adoption over and a claim's ownership, |
| 27 | // which the import journal cannot put back (registry.SwapForImport). |
| 28 | func importShape(rt registry.Route) string { |
| 29 | switch { |
| 30 | case rt.Kind != "instance" && rt.Kind != "service": |
| 31 | return "kind " + rt.Kind |
| 32 | case rt.Socket != "": |
| 33 | return "socket" |
| 34 | case rt.Addr == "" && rt.Addrs == "": |
| 35 | return "no address" |
| 36 | case rt.Provisioned: |
| 37 | return "provisioned" |
| 38 | case rt.Backing != "": |
| 39 | return "backing" |
| 40 | case rt.Tunnel != "": |
| 41 | return "tunnel" |
| 42 | case rt.Caps != "": |
| 43 | return "caps" |
| 44 | case rt.Platform: |
| 45 | return "platform" |
| 46 | case rt.Canary: |
| 47 | return "canary" |
| 48 | case rt.FleetBox != "": |
| 49 | return "fleet_box" |
| 50 | case rt.Kind == "instance" && rt.Hibernate: |
| 51 | return "hibernate" |
| 52 | case rt.ClaimOwnership != (registry.ClaimOwnership{}): |
| 53 | return "claim_owner" |
| 54 | } |
| 55 | if _, fleet := registry.FleetPlacement(rt.Region); fleet { |
| 56 | return "region" |
| 57 | } |
| 58 | return "" |
| 59 | } |
| 60 | |
| 61 | // importAddr is the address the row serves, as the edge reads it |
| 62 | // (registry.Route.AddrList): addrs when set, else addr. A row whose addr |
| 63 | // is Replaces but whose addrs names something else serves the addrs, so |
| 64 | // it is not the route the import recorded. |
| 65 | func importAddr(rt registry.Route) string { |
| 66 | return strings.Join(rt.AddrList(), ",") |
| 67 | } |
| 68 | |
| 69 | // importRefusal is §5.4's predicates in order: "" to swap, else the |
| 70 | // refusal reason the record keeps. owners is consulted only for a row |
| 71 | // filed under the operator account: release.RouteAccount folds a legacy |
| 72 | // row's "" (and any account that is not a sqid) into release.OpsAccount, |
| 73 | // so such a row is the operator's too (preflight M5). |
| 74 | func importRefusal(rec Record, existing *registry.Route, cfg ReconcileConfig, owners *seeds.LegacyOwners, nested bool) string { |
| 75 | switch { |
| 76 | case cfg.ClaimEnabled: |
| 77 | return "claim control is on for this box" |
| 78 | case !nested: |
| 79 | return "this box does not run instances nested" |
| 80 | case existing == nil: |
| 81 | return fmt.Sprintf("there is no route at %s to replace", rec.Host) |
| 82 | } |
| 83 | if what := importShape(*existing); what != "" { |
| 84 | return fmt.Sprintf("the route is not an address route this import can take over (%s)", what) |
| 85 | } |
| 86 | if addr := importAddr(*existing); addr != rec.Seed.Replaces { |
| 87 | return fmt.Sprintf("the route's address is %s, not %s", addr, rec.Seed.Replaces) |
| 88 | } |
| 89 | if existing.App != rec.App { |
| 90 | return fmt.Sprintf("the route belongs to app %s", existing.App) |
| 91 | } |
| 92 | switch { |
| 93 | case release.RouteAccount(*existing) == rec.Account: |
| 94 | return "" |
| 95 | case operatorFiled(rec, existing): |
| 96 | if o, ok := owners.Lookup(existing.App); ok && o.Account == rec.Account && o.App == rec.App { |
| 97 | return "" |
| 98 | } |
| 99 | return fmt.Sprintf("the route is filed under the operator account, and no assignment maps %s to this app", existing.App) |
| 100 | } |
| 101 | return "the route is filed under another account" |
| 102 | } |
| 103 | |
| 104 | // operatorFiled reports a row the swap may take only through the operator |
| 105 | // assignment: filed under the operator account, for a record that is not |
| 106 | // the operator's own. importRefusal's account rule and swapImport's |
| 107 | // decision to read the assignment both ask this one question. |
| 108 | func operatorFiled(rec Record, rt *registry.Route) bool { |
| 109 | if rt == nil { |
| 110 | return false |
| 111 | } |
| 112 | acct := release.RouteAccount(*rt) |
| 113 | return acct != rec.Account && acct == release.OpsAccount |
| 114 | } |
| 115 | |
| 116 | var verifyHashShape = regexp.MustCompile(`^[0-9a-f]{64}$`) |
| 117 | |
| 118 | // importUngateable is why the swap must not run for this record at all |
| 119 | // (preflight M3), or "": the gate needs a hash (an empty one would swap in |
| 120 | // an ungated row), the host and seed id name the snapshot and must pass |
| 121 | // seeds.Check, and the new row must be one the activator manages, nested |
| 122 | // and hibernating, since only that row places the snapshot before it |
| 123 | // serves. These are not refusals: the record stays ready, logged, until |
| 124 | // it is aborted. |
| 125 | func importUngateable(rec Record, next registry.Route, dataDir string) string { |
| 126 | if err := seeds.Check(rec.Host, rec.Seed.ID); err != nil { |
| 127 | return fmt.Sprintf("its host or seed id is malformed (%v)", err) |
| 128 | } |
| 129 | switch { |
| 130 | case !verifyHashShape.MatchString(rec.Seed.VerifyHash): |
| 131 | return "its verify hash is not a sha256" |
| 132 | case !rec.DerivesPaths() || !next.Hibernate || !NestedPaths(dataDir, next): |
| 133 | return "its record does not describe a nested, hibernating instance" |
| 134 | } |
| 135 | return "" |
| 136 | } |
| 137 | |
| 138 | // importRow is the exec row the swap installs (spec §5.4): the record's |
| 139 | // route at the nested paths, explicitly, never NestedPaths(existing), |
| 140 | // which reads the addr row as flat (codex 12); no seed command; filed |
| 141 | // under the record's account; gated by the verify hash. Its seed column |
| 142 | // already says swapped, the state the record's CAS moves it to. policy is |
| 143 | // the app's update policy, as for any mint (RouteFor): the swap's row is |
| 144 | // born with the record's effective update settings. |
| 145 | func importRow(rec Record, dataDir string, policy *UpdatePolicy) registry.Route { |
| 146 | swapped := *rec.Seed |
| 147 | swapped.State = seeds.StateSwapped |
| 148 | r := rec |
| 149 | r.Seed = &swapped |
| 150 | next := RouteFor(r, dataDir, "", true, policy) |
| 151 | next.ImportGate = rec.Seed.VerifyHash |
| 152 | return next |
| 153 | } |
| 154 | |
| 155 | // swapImport is the swap (spec §5.4): the journal first, then the |
| 156 | // predicates, then one registry transaction, then the record's CAS to |
| 157 | // swapped. Transient failures defer to the next pass and write nothing; a |
| 158 | // failed predicate is a durable refusal. policy is the app's update policy, |
| 159 | // known (Reconcile defers the swap, as it defers a mint, while it cannot be |
| 160 | // read). |
| 161 | func swapImport(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, rec Record, existing *registry.Route, nested func() (bool, error), policy *UpdatePolicy) { |
| 162 | if cfg.DryRun { |
| 163 | log.Printf("provision: dry-run: would consider swapping %s for import %s", rec.Host, rec.Seed.ID) |
| 164 | return |
| 165 | } |
| 166 | // The journal comes first (codex R2-3): a row there means this box has |
| 167 | // already swapped, so the legacy-row predicates never judge the row it |
| 168 | // made. This seed's own row at swapped is a crash between the swap's |
| 169 | // transaction and the record's CAS: only the CAS is redone. |
| 170 | // |
| 171 | // Reconcile's journalPass decides every journaled host before the swap |
| 172 | // is reached, so in a pass this check finds nothing. It is kept as a |
| 173 | // backstop: without it, a journal row this function did not expect |
| 174 | // (another caller, or a row written after the pass read the journal) |
| 175 | // would be judged as a legacy row and refused, terminally. |
| 176 | j, err := reg.ImportJournal(rec.Host) |
| 177 | if err != nil { |
| 178 | log.Printf("provision: %s: import deferred: the import journal could not be read: %v", rec.Host, err) |
| 179 | return |
| 180 | } |
| 181 | if j != nil { |
| 182 | if j.SeedID != rec.Seed.ID || j.State != registry.JournalSwapped || j.RecordKey != RecordKey(rec.Account, rec.App, rec.Host) { |
| 183 | log.Printf("provision: %s: import %s left alone: the import journal holds seed %s at %s", rec.Host, rec.Seed.ID, j.SeedID, j.State) |
| 184 | return |
| 185 | } |
| 186 | markSwapped(ctx, s, rec) |
| 187 | return |
| 188 | } |
| 189 | |
| 190 | next := importRow(rec, cfg.DataDir, policy) |
| 191 | if why := importUngateable(rec, next, cfg.DataDir); why != "" { |
| 192 | log.Printf("provision: %s: not swapping for import %s: %s", rec.Host, rec.Seed.ID, why) |
| 193 | return |
| 194 | } |
| 195 | isNested := false |
| 196 | if !cfg.ClaimEnabled { |
| 197 | ok, err := nested() |
| 198 | if err != nil { |
| 199 | log.Printf("provision: %s: import deferred: the box's layout intent could not be read: %v", rec.Host, err) |
| 200 | return |
| 201 | } |
| 202 | // Only the intent refuses (SD-N97). The live probe can fail for a |
| 203 | // moment, and a refusal is terminal: an error waits for the next |
| 204 | // pass (preflight M4). |
| 205 | if ok && cfg.Seeds.NestedOK != nil { |
| 206 | if err := cfg.Seeds.NestedOK(); err != nil { |
| 207 | log.Printf("provision: %s: import deferred: the box's nested probe did not answer: %v", rec.Host, err) |
| 208 | return |
| 209 | } |
| 210 | } |
| 211 | isNested = ok |
| 212 | } |
| 213 | if existing != nil && existing.HoldOp != "" { |
| 214 | log.Printf("provision: %s: import deferred: the route is held by operation %s", rec.Host, existing.HoldOp) |
| 215 | return |
| 216 | } |
| 217 | // The assignment is read only when it can decide: claims off (SD-N98 |
| 218 | // comes first) and the box nested (SD-N97 next). |
| 219 | var owners *seeds.LegacyOwners |
| 220 | if !cfg.ClaimEnabled && isNested && operatorFiled(rec, existing) { |
| 221 | if cfg.Seeds.LegacyOwners == nil { |
| 222 | log.Printf("provision: %s: import deferred: this box cannot read the legacy route assignments", rec.Host) |
| 223 | return |
| 224 | } |
| 225 | o, err := cfg.Seeds.LegacyOwners(ctx) |
| 226 | if err != nil { |
| 227 | log.Printf("provision: %s: import deferred: the legacy route assignments could not be read: %v", rec.Host, err) |
| 228 | return |
| 229 | } |
| 230 | owners = o |
| 231 | } |
| 232 | if reason := importRefusal(rec, existing, cfg, owners, isNested); reason != "" { |
| 233 | if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedRefuse(rec.Seed.ID, reason)); err != nil { |
| 234 | log.Printf("provision: %s: record the import's refusal (%s): %v", rec.Host, reason, err) |
| 235 | return |
| 236 | } |
| 237 | log.Printf("provision: %s: import %s refused: %s", rec.Host, rec.Seed.ID, reason) |
| 238 | return |
| 239 | } |
| 240 | if cfg.ConfigGate != nil { |
| 241 | if err := cfg.ConfigGate(ctx, next); err != nil { |
| 242 | return // the gate logs its own refusal; retried every pass |
| 243 | } |
| 244 | } |
| 245 | // What the host's own env file holds before the import's row exists, |
| 246 | // for the journal: the rollback retires only the instance key the |
| 247 | // import's launches created (merge review S1). Read last, so nothing |
| 248 | // can launch the import between this read and the transaction. |
| 249 | if cfg.Seeds.EnvBefore == nil { |
| 250 | log.Printf("provision: %s: import deferred: this box cannot read the instance's env file", rec.Host) |
| 251 | return |
| 252 | } |
| 253 | envBefore, err := cfg.Seeds.EnvBefore(next.DBPath, rec.Host) |
| 254 | if err != nil { |
| 255 | log.Printf("provision: %s: import deferred: the instance's env file could not be read: %v", rec.Host, err) |
| 256 | return |
| 257 | } |
| 258 | now := time.Now |
| 259 | if cfg.Now != nil { |
| 260 | now = cfg.Now |
| 261 | } |
| 262 | journal := registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *existing, |
| 263 | RecordKey: RecordKey(rec.Account, rec.App, rec.Host), At: now().Unix(), EnvBefore: envBefore} |
| 264 | if err := reg.SwapForImport(journal, next); err != nil { |
| 265 | log.Printf("provision: %s: import swap deferred: %v", rec.Host, err) |
| 266 | return |
| 267 | } |
| 268 | log.Printf("provision: %s: swapped for import %s (the original route is journaled)", rec.Host, rec.Seed.ID) |
| 269 | markSwapped(ctx, s, rec) |
| 270 | } |
| 271 | |
| 272 | // markSwapped is the swap's record CAS, ready to swapped. It runs only |
| 273 | // after the registry transaction committed; a failure leaves the journal |
| 274 | // at swapped and the record at ready, which the next pass resumes. |
| 275 | func markSwapped(ctx context.Context, s release.Store, rec Record) { |
| 276 | if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedSwapped(rec.Seed.ID)); err != nil { |
| 277 | if errors.Is(err, ErrSeedConflict) { |
| 278 | log.Printf("provision: %s: import %s swapped, but the record moved on: %v", rec.Host, rec.Seed.ID, err) |
| 279 | return |
| 280 | } |
| 281 | log.Printf("provision: %s: import %s swapped, record CAS deferred to the next pass: %v", rec.Host, rec.Seed.ID, err) |
| 282 | } |
| 283 | } |
| 284 | |
amadan