| 1 | // The import door (seeded-instances spec §5): `carlos instances import` |
| 2 | // moves an instance the platform does not host yet onto it, from an |
| 3 | // uploaded snapshot. Begin is the create door with "import" in its body: |
| 4 | // handleAPIInstanceCreate runs every ordinary check, then hands over here. |
| 5 | // Begin writes the record awaiting-upload, provisions an empty store for a |
| 6 | // private-storage app, and answers 201 with the verify token, once. |
| 7 | |
| 8 | package console |
| 9 | |
| 10 | import ( |
| 11 | "context" |
| 12 | "crypto/sha256" |
| 13 | "encoding/hex" |
| 14 | "errors" |
| 15 | "fmt" |
| 16 | "io" |
| 17 | "log" |
| 18 | "net" |
| 19 | "net/http" |
| 20 | "os" |
| 21 | "path/filepath" |
| 22 | "regexp" |
| 23 | "runtime/debug" |
| 24 | "strconv" |
| 25 | "time" |
| 26 | |
| 27 | "github.com/carlosframework/platform/internal/lease" |
| 28 | "github.com/carlosframework/platform/internal/provision" |
| 29 | "github.com/carlosframework/platform/internal/release" |
| 30 | "github.com/carlosframework/platform/internal/seeds" |
| 31 | ) |
| 32 | |
| 33 | // seedImportReq is the create body's "import" (spec §5.1): the snapshot's |
| 34 | // length and hash, which the upload must match, and the address of the |
| 35 | // addr-backed row the box swaps out. |
| 36 | type seedImportReq struct { |
| 37 | SHA256 string `json:"sha256"` |
| 38 | Bytes int64 `json:"bytes"` |
| 39 | Replaces string `json:"replaces"` |
| 40 | } |
| 41 | |
| 42 | // importFailed is the import door's retry sentence (SD-N67). |
| 43 | const importFailed = "could not store the upload. Try again." |
| 44 | |
| 45 | // importStepFailure is the retry sentence of every import call that is not |
| 46 | // storing the upload: the record read at the gate, and the stopped, commit |
| 47 | // and abort steps (SD-N120). |
| 48 | func importStepFailure(host string) string { |
| 49 | return fmt.Sprintf("could not update the import of %s. Try again.", host) |
| 50 | } |
| 51 | |
| 52 | // importSizeRule is SD-N48, answered 400 below the range and 413 above it. |
| 53 | const importSizeRule = "bytes must be more than 0 and at most 4 GiB" |
| 54 | |
| 55 | var ( |
| 56 | sha256Hex = regexp.MustCompile(`^[0-9a-f]{64}$`) |
| 57 | portShape = regexp.MustCompile(`^[0-9]{1,5}$`) |
| 58 | ) |
| 59 | |
| 60 | // seedOneDoorRefusal is the create door's rule that a body takes at most |
| 61 | // one seed door: "from" and "import" together are refused. |
| 62 | func seedOneDoorRefusal(from, imp bool) *doorRefusal { |
| 63 | if from && imp { |
| 64 | return &doorRefusal{http.StatusBadRequest, "a create takes from or import, not both."} |
| 65 | } |
| 66 | return nil |
| 67 | } |
| 68 | |
| 69 | // importShapeRefusal checks begin's own fields: the size against the |
| 70 | // limit, the hash's spelling, and replaces as host:port with a port in |
| 71 | // 1 to 65535. |
| 72 | func importShapeRefusal(imp seedImportReq) *doorRefusal { |
| 73 | switch { |
| 74 | case imp.Bytes <= 0: |
| 75 | return &doorRefusal{http.StatusBadRequest, importSizeRule} |
| 76 | case imp.Bytes > seedMaxBytes: |
| 77 | return &doorRefusal{http.StatusRequestEntityTooLarge, importSizeRule} |
| 78 | case !sha256Hex.MatchString(imp.SHA256): |
| 79 | return &doorRefusal{http.StatusBadRequest, "sha256 must be 64 lowercase hex characters"} |
| 80 | case !validReplaces(imp.Replaces): |
| 81 | return &doorRefusal{http.StatusBadRequest, "replaces must name the address the import takes over, as host:port"} |
| 82 | } |
| 83 | return nil |
| 84 | } |
| 85 | |
| 86 | func validReplaces(addr string) bool { |
| 87 | host, port, err := net.SplitHostPort(addr) |
| 88 | if err != nil || host == "" || !portShape.MatchString(port) { |
| 89 | return false |
| 90 | } |
| 91 | n, err := strconv.Atoi(port) |
| 92 | return err == nil && n > 0 && n < 65536 |
| 93 | } |
| 94 | |
| 95 | // createImport is the import door's begin (spec §5.1). Its answer is held |
| 96 | // until the door is done, so that a begin which fails after writing its |
| 97 | // record has rolled that record back and finished it (freeing the host) |
| 98 | // before the refusal reaches the caller (preflight M2). |
| 99 | func (s *Server) createImport(w http.ResponseWriter, r *http.Request, c seedCreate, imp seedImportReq) { |
| 100 | held := &heldResponse{header: w.Header()} |
| 101 | s.beginImport(held, r, c, imp) |
| 102 | held.flush(w, importFailed) |
| 103 | } |
| 104 | |
| 105 | // beginImport runs begin's checks after the create door's own, then mints |
| 106 | // the seed id and the verify token, writes the record awaiting-upload with |
| 107 | // only the token's hash, provisions the seed's store, and answers 201 with |
| 108 | // the token. The token is in that one body and nowhere else: never logged, |
| 109 | // audited or stored. |
| 110 | func (s *Server) beginImport(w http.ResponseWriter, r *http.Request, c seedCreate, imp seedImportReq) { |
| 111 | ctx := r.Context() |
| 112 | acct, app, host := c.acct, c.app, c.rec.Host |
| 113 | if ref := seedPlacementRefusal(c); ref != nil { |
| 114 | jsonError(w, ref.Status, ref.Msg) |
| 115 | return |
| 116 | } |
| 117 | if s.cfg.SeedData == nil || s.cfg.HibernateBucket == "" { |
| 118 | ref := seedNoDataRefusal() |
| 119 | jsonError(w, ref.Status, ref.Msg) |
| 120 | return |
| 121 | } |
| 122 | if ref := importShapeRefusal(imp); ref != nil { |
| 123 | jsonError(w, ref.Status, ref.Msg) |
| 124 | return |
| 125 | } |
| 126 | // The snapshot goes to this console's data bucket, and the import's |
| 127 | // rollback reads it there: the record is homed in this region. |
| 128 | if c.rec.Region != s.cfg.Region { |
| 129 | jsonError(w, http.StatusConflict, fmt.Sprintf("an import runs in this console's region, %s. To run %s elsewhere, import it here, commit it, and then run carlos move.", |
| 130 | s.dbRegionName(s.cfg.Region), host)) |
| 131 | return |
| 132 | } |
| 133 | checks, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 134 | defer cancel() |
| 135 | if ref := s.seedRegionReady(checks, c.rec.Region, true); ref != nil { |
| 136 | jsonError(w, ref.Status, ref.Msg) |
| 137 | return |
| 138 | } |
| 139 | id, err := seeds.NewID() |
| 140 | if err != nil { |
| 141 | log.Printf("console: import %s: seed id: %v", host, err) |
| 142 | jsonError(w, http.StatusBadGateway, importFailed) |
| 143 | return |
| 144 | } |
| 145 | token, hash, err := seeds.NewVerifyToken() |
| 146 | if err != nil { |
| 147 | log.Printf("console: import %s: verify token: %v", host, err) |
| 148 | jsonError(w, http.StatusBadGateway, importFailed) |
| 149 | return |
| 150 | } |
| 151 | if c.m.PrivateStorage { |
| 152 | // The store gates that write nothing, before the record (ruling |
| 153 | // R3): their refusals leave nothing holding the host. |
| 154 | if _, ok := s.instanceStorePreflight(ctx, w, acct, app); !ok { |
| 155 | return |
| 156 | } |
| 157 | storeCtx, storeCancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 158 | ref := s.seedStoreRefusal(storeCtx, c, id, importFailed) |
| 159 | storeCancel() |
| 160 | if ref != nil { |
| 161 | jsonError(w, ref.Status, ref.Msg) |
| 162 | return |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | rec := c.rec |
| 167 | rec.Seed = &provision.SeedSource{ID: id, Kind: seeds.KindImport, SHA256: imp.SHA256, Bytes: imp.Bytes, |
| 168 | State: seeds.StateAwaitingUpload, By: c.principal, At: time.Now().UTC(), Replaces: imp.Replaces, VerifyHash: hash} |
| 169 | // Not the request's cancellation: a client leaving now must not turn a |
| 170 | // write that landed into an error. |
| 171 | writeCtx, writeCancel := context.WithTimeout(context.WithoutCancel(ctx), instanceStoreTimeout) |
| 172 | err = provision.CreateRecord(writeCtx, s.cfg.Releases, rec) |
| 173 | writeCancel() |
| 174 | if err != nil { |
| 175 | if errors.Is(err, release.ErrCASConflict) { |
| 176 | jsonError(w, http.StatusConflict, "instance "+host+" already exists") |
| 177 | return |
| 178 | } |
| 179 | // The write may have landed all the same: undo it by its seed id, |
| 180 | // which touches nothing if it did not. |
| 181 | log.Printf("console: import %s: write the record: %v", host, err) |
| 182 | s.unbeginImport(ctx, c, id, "a failed record write") |
| 183 | jsonError(w, http.StatusBadGateway, importFailed) |
| 184 | return |
| 185 | } |
| 186 | |
| 187 | // From here every failure rolls the import back and finishes it |
| 188 | // before the answer goes out. |
| 189 | privateBucket := "" |
| 190 | if c.m.PrivateStorage { |
| 191 | // The request's own context, as the ordinary create's walk: the |
| 192 | // walk takes storeGrantTimeout, and a client that leaves cancels a |
| 193 | // begin whose token it can no longer receive. A refusal is held on |
| 194 | // w already, except errSeedStoreTaken and errSeedEnvironmentUnread, |
| 195 | // which are spoken here. |
| 196 | storeRec, ok, err := s.instanceStoreProvisionFor(ctx, w, acct, app, rec.Environment, c.principal, id) |
| 197 | if !ok { |
| 198 | s.unbeginImport(ctx, c, id, "store provisioning") |
| 199 | switch { |
| 200 | case errors.Is(err, errSeedStoreTaken): |
| 201 | ref := seedStoreTakenRefusal(rec.Environment, app, storeRec) |
| 202 | jsonError(w, ref.Status, ref.Msg) |
| 203 | case errors.Is(err, errSeedEnvironmentUnread): |
| 204 | jsonError(w, http.StatusBadGateway, importFailed) |
| 205 | } |
| 206 | return |
| 207 | } |
| 208 | privateBucket = storeBucketName(storeRec) |
| 209 | } |
| 210 | // The create door's re-check after its write (apiinstances.go): an app |
| 211 | // created under the host's label since the door asked owns its apex. |
| 212 | if c.siblingLabel != "" { |
| 213 | if live, trashed, err := s.store.AppNameState(acct.ID, c.siblingLabel); err == nil && (live || trashed) { |
| 214 | s.unbeginImport(ctx, c, id, "a sibling app took the label") |
| 215 | jsonError(w, http.StatusConflict, seedSiblingApexReason(host, c.siblingLabel)) |
| 216 | return |
| 217 | } |
| 218 | } |
| 219 | |
| 220 | s.auditActivity(acct, app, c.principal, Activity{ |
| 221 | Action: "instance-seed", TargetType: "instance", Target: host, Message: c.message, |
| 222 | Detail: fmt.Sprintf("began an import of %s over %s (%s, %d bytes)", host, imp.Replaces, imp.SHA256[:12], imp.Bytes), |
| 223 | }) |
| 224 | viewCtx, viewCancel := context.WithTimeout(context.WithoutCancel(ctx), instanceStoreTimeout) |
| 225 | defer viewCancel() |
| 226 | out := struct { |
| 227 | apiRecordView |
| 228 | VerifyToken string `json:"verify_token"` |
| 229 | }{recordView(rec), token} |
| 230 | out.PrivateStoreBucket = privateBucket |
| 231 | out.RegionFrom = c.regionFrom |
| 232 | out.RegionName = s.placeOf(rec.Region, s.candsOrHome(viewCtx)) // never fleet-placed: refused above |
| 233 | writeJSON(w, http.StatusCreated, out) |
| 234 | } |
| 235 | |
| 236 | // unbeginImport undoes a begin that failed after its record was written: |
| 237 | // the pre-swap rollback, then the console's finish, which tears down the |
| 238 | // store this seed made and deletes the record (spec §5.6's pre-swap path, |
| 239 | // preflight M2). Neither is cut by the client leaving. A failure is |
| 240 | // logged, and the record is left rolled-back for the sweep to finish. |
| 241 | func (s *Server) unbeginImport(ctx context.Context, c seedCreate, id, why string) { |
| 242 | sqid, app, host := c.acct.Sqid, c.app, c.rec.Host |
| 243 | bg := context.WithoutCancel(ctx) |
| 244 | rbCtx, rbCancel := context.WithTimeout(bg, instanceStoreTimeout) |
| 245 | _, err := provision.SetRecordSeed(rbCtx, s.cfg.Releases, sqid, app, host, provision.SeedRolledBackBeforeSwap(id)) |
| 246 | rbCancel() |
| 247 | if errors.Is(err, os.ErrNotExist) { |
| 248 | return // the record was never written |
| 249 | } |
| 250 | if err != nil { |
| 251 | log.Printf("console: import %s: roll back seed %s after %s: %v", host, id, why, err) |
| 252 | } |
| 253 | finCtx, finCancel := context.WithTimeout(bg, storeTeardownTimeout+instanceStoreTimeout) |
| 254 | defer finCancel() |
| 255 | if err := s.finishSeed(finCtx, sqid, app, host, id); err != nil { |
| 256 | log.Printf("console: import %s: finish seed %s after %s: %v", host, id, why, err) |
| 257 | } |
| 258 | } |
| 259 | |
| 260 | // Seams for the tests. |
| 261 | var ( |
| 262 | // checkSeedDBConsole is the upload's integrity check (spec §5.2 step 3). |
| 263 | checkSeedDBConsole = lease.CheckDBContext |
| 264 | // importCheckDeadline bounds that check. |
| 265 | importCheckDeadline = 30 * time.Minute |
| 266 | // importSpoolHook, when set, sees the spool once the body is read and |
| 267 | // before it is checked or removed. |
| 268 | importSpoolHook func(path string) |
| 269 | // importReserveHook, when set, runs after the upload's first read of |
| 270 | // the record and before its reservation. |
| 271 | importReserveHook func(host string) |
| 272 | ) |
| 273 | |
| 274 | // importUploadReadDeadline is how long the upload's body may take to |
| 275 | // arrive: up to 4 GiB, past the server's ReadTimeout. It covers the body |
| 276 | // read only; the check and the put run after the answer. |
| 277 | const importUploadReadDeadline = 60 * time.Minute |
| 278 | |
| 279 | // importGate is every import call's gate: the instance gate (cliRoute has |
| 280 | // already applied the instance role), the claim console's refusal, the |
| 281 | // host, the record, that the record is an import (SD-N59), and that it is |
| 282 | // the import the caller names (SD-N60). |
| 283 | func (s *Server) importGate(w http.ResponseWriter, r *http.Request) (principal string, acct *Account, app, host string, rec *provision.Record, ok bool) { |
| 284 | principal, acct, app, ok = s.instanceGate(w, r) |
| 285 | if !ok { |
| 286 | return "", nil, "", "", nil, false |
| 287 | } |
| 288 | // A claim console begins no import (seedRegionReady), and its claim |
| 289 | // protocol has no place for one yet (#250): every import door refuses |
| 290 | // there before it reads the record. |
| 291 | if s.cfg.HostClaim { |
| 292 | ref := seedClaimConsoleRefusal() |
| 293 | jsonError(w, ref.Status, ref.Msg) |
| 294 | return "", nil, "", "", nil, false |
| 295 | } |
| 296 | host = r.PathValue("host") |
| 297 | notInstance := notAnInstanceOf(host, app) |
| 298 | if !provision.ValidHostShape(host) { |
| 299 | jsonError(w, http.StatusNotFound, notInstance) |
| 300 | return "", nil, "", "", nil, false |
| 301 | } |
| 302 | ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout) |
| 303 | defer cancel() |
| 304 | rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host) |
| 305 | if errors.Is(err, os.ErrNotExist) { |
| 306 | jsonError(w, http.StatusNotFound, notInstance) |
| 307 | return "", nil, "", "", nil, false |
| 308 | } |
| 309 | if err != nil { |
| 310 | log.Printf("console: import %s/%s/%s: read the record: %v", acct.Sqid, app, host, err) |
| 311 | jsonError(w, http.StatusBadGateway, importStepFailure(host)) |
| 312 | return "", nil, "", "", nil, false |
| 313 | } |
| 314 | if !rec.Seed.Import() { |
| 315 | ref := notAnImportRefusal(host) |
| 316 | jsonError(w, ref.Status, ref.Msg) |
| 317 | return "", nil, "", "", nil, false |
| 318 | } |
| 319 | // The caller names the import it means (final review F1). A delayed |
| 320 | // stopped, commit, abort or upload sent for one import must never land |
| 321 | // on a later import of the same host: the record's id is what the |
| 322 | // transitions below are guarded on, so it has to be the caller's id |
| 323 | // too. A request that names another seed, or none, changed nothing: |
| 324 | // SD-N60, and the caller reads the state. |
| 325 | if r.URL.Query().Get(importSeedParam) != rec.Seed.ID { |
| 326 | s.importTransitionFailure(w, app, host, provision.ErrSeedConflict) |
| 327 | return "", nil, "", "", nil, false |
| 328 | } |
| 329 | return principal, acct, app, host, rec, true |
| 330 | } |
| 331 | |
| 332 | // importSeedParam is the query parameter every import mutation names its |
| 333 | // seed id in: stopped, commit, abort and the upload. |
| 334 | const importSeedParam = "seed" |
| 335 | |
| 336 | // importUpload is an upload whose spool matched: everything |
| 337 | // runImportUpload needs once the request has been answered. It owns the |
| 338 | // spool's directory and the scratch reservation from the hand-over. |
| 339 | type importUpload struct { |
| 340 | ctx context.Context // the request's values, never its cancellation |
| 341 | acct *Account |
| 342 | app, host, by string |
| 343 | id string |
| 344 | dir, spool string |
| 345 | release func() |
| 346 | } |
| 347 | |
| 348 | // handleAPIImportUpload is spec §5.2, in two halves (preflight M6, a |
| 349 | // deviation from the spec's synchronous 422 and 504 for Paul, following |
| 350 | // Plan 5 ruling R2). The request half reads the body into a private spool |
| 351 | // and checks its length and sha256 against begin's: a mismatch answers 422 |
| 352 | // and leaves the import awaiting-upload, so the right file can be sent. |
| 353 | // Once the spool matches it answers 202 {"state":"checking"} and the rest |
| 354 | // runs detached (runImportUpload): an edge backhaul's 30-second |
| 355 | // response-header timeout and idle NATs would cut a response that waited |
| 356 | // for a 30-minute check and a 4 GiB put. The caller follows the record |
| 357 | // until it is uploaded, or aborted with the reason in Seed.Reason. |
| 358 | func (s *Server) handleAPIImportUpload(w http.ResponseWriter, r *http.Request) { |
| 359 | principal, acct, app, host, rec, ok := s.importGate(w, r) |
| 360 | if !ok { |
| 361 | return |
| 362 | } |
| 363 | ctx := r.Context() |
| 364 | seed := rec.Seed |
| 365 | if ref := staleUploadRefusal(host, rec, seed.ID); ref != nil { |
| 366 | jsonError(w, ref.Status, ref.Msg) |
| 367 | return |
| 368 | } |
| 369 | // The snapshot key is built from the record (preflight M8): an id that |
| 370 | // cannot name a key must never reach the put. |
| 371 | if err := seeds.Check(host, seed.ID); err != nil { |
| 372 | log.Printf("console: import %s: seed %q: %v", host, seed.ID, err) |
| 373 | jsonError(w, http.StatusBadGateway, importFailed) |
| 374 | return |
| 375 | } |
| 376 | if s.cfg.SeedData == nil || s.cfg.HibernateBucket == "" { |
| 377 | ref := seedNoDataRefusal() |
| 378 | jsonError(w, ref.Status, ref.Msg) |
| 379 | return |
| 380 | } |
| 381 | checks, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 382 | ref := s.seedRegionReady(checks, rec.Region, true) |
| 383 | cancel() |
| 384 | if ref != nil { |
| 385 | jsonError(w, ref.Status, ref.Msg) |
| 386 | return |
| 387 | } |
| 388 | if importReserveHook != nil { |
| 389 | importReserveHook(host) |
| 390 | } |
| 391 | releaseScratch, ref := s.seedScratch.reserve(acct.Sqid+"/"+app, seed.Bytes) |
| 392 | if ref != nil { |
| 393 | jsonError(w, ref.Status, ref.Msg) |
| 394 | return |
| 395 | } |
| 396 | // The read above came before the reservation: another upload of this |
| 397 | // import may have landed in between (fix round 1). Under the |
| 398 | // reservation, the record must still wait for this seed's upload. |
| 399 | fresh, err := s.readImportRecord(ctx, acct.Sqid, app, host) |
| 400 | switch { |
| 401 | case errors.Is(err, os.ErrNotExist): |
| 402 | releaseScratch() |
| 403 | jsonError(w, http.StatusNotFound, notAnInstanceOf(host, app)) |
| 404 | return |
| 405 | case err != nil: |
| 406 | releaseScratch() |
| 407 | log.Printf("console: import %s/%s/%s: re-read the record: %v", acct.Sqid, app, host, err) |
| 408 | jsonError(w, http.StatusBadGateway, importFailed) |
| 409 | return |
| 410 | } |
| 411 | if ref := staleUploadRefusal(host, fresh, seed.ID); ref != nil { |
| 412 | releaseScratch() |
| 413 | jsonError(w, ref.Status, ref.Msg) |
| 414 | return |
| 415 | } |
| 416 | dir, err := os.MkdirTemp(s.seedScratch.dir(), seedScratchPrefix) |
| 417 | if err != nil { |
| 418 | releaseScratch() |
| 419 | log.Printf("console: import %s: scratch: %v", host, err) |
| 420 | ref := seedScratchFull() |
| 421 | jsonError(w, ref.Status, ref.Msg) |
| 422 | return |
| 423 | } |
| 424 | handedOver := false |
| 425 | defer func() { |
| 426 | if !handedOver { |
| 427 | _ = os.RemoveAll(dir) // with any -shm or -wal beside the spool |
| 428 | releaseScratch() |
| 429 | } |
| 430 | }() |
| 431 | spool := filepath.Join(dir, host+".db") |
| 432 | if !s.spoolImport(w, r, spool, seed) { |
| 433 | return |
| 434 | } |
| 435 | writeJSON(w, http.StatusAccepted, map[string]string{"state": "checking"}) |
| 436 | handedOver = true |
| 437 | s.seedCopies.Add(1) |
| 438 | go s.runImportUpload(&importUpload{ctx: context.WithoutCancel(ctx), acct: acct, app: app, host: host, by: principal, |
| 439 | id: seed.ID, dir: dir, spool: spool, release: releaseScratch}) |
| 440 | } |
| 441 | |
| 442 | // readImportRecord reads host's record under instanceStoreTimeout. |
| 443 | func (s *Server) readImportRecord(ctx context.Context, sqid, app, host string) (*provision.Record, error) { |
| 444 | ctx, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 445 | defer cancel() |
| 446 | return provision.GetRecord(ctx, s.cfg.Releases, sqid, app, host) |
| 447 | } |
| 448 | |
| 449 | // awaitingUpload reports whether rec's seed is the import id, still |
| 450 | // waiting for its upload. |
| 451 | func awaitingUpload(rec *provision.Record, id string) bool { |
| 452 | return rec.Seed.Import() && rec.Seed.ID == id && rec.Seed.State == seeds.StateAwaitingUpload |
| 453 | } |
| 454 | |
| 455 | // notAnImportRefusal is SD-N59. |
| 456 | func notAnImportRefusal(host string) *doorRefusal { |
| 457 | return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s is not an import.", host)} |
| 458 | } |
| 459 | |
| 460 | // staleUploadRefusal is the upload's answer to a record that no longer |
| 461 | // waits for seed id's upload: SD-N53 naming its state, or SD-N59 when it |
| 462 | // is no longer an import at all. |
| 463 | func staleUploadRefusal(host string, rec *provision.Record, id string) *doorRefusal { |
| 464 | switch { |
| 465 | case awaitingUpload(rec, id): |
| 466 | return nil |
| 467 | case !rec.Seed.Import(): |
| 468 | return notAnImportRefusal(host) |
| 469 | } |
| 470 | return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s is %s. It takes an upload only while it waits for one.", host, rec.Seed.State)} |
| 471 | } |
| 472 | |
| 473 | // spoolImport reads the body into spool, never more than the declared |
| 474 | // length and one byte, hashing as it writes, and reports whether it |
| 475 | // matches begin's length and sha256. It has answered w when it reports |
| 476 | // false. |
| 477 | func (s *Server) spoolImport(w http.ResponseWriter, r *http.Request, spool string, seed *provision.SeedSource) bool { |
| 478 | host := r.PathValue("host") |
| 479 | // The server's ReadTimeout would cut a large body; this read gets an |
| 480 | // hour. Go drops the deadline once the body has been read. |
| 481 | if err := http.NewResponseController(w).SetReadDeadline(time.Now().Add(importUploadReadDeadline)); err != nil { |
| 482 | log.Printf("console: import %s: extend the read deadline: %v", host, err) |
| 483 | } |
| 484 | f, err := os.OpenFile(spool, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) |
| 485 | if err != nil { |
| 486 | log.Printf("console: import %s: spool: %v", host, err) |
| 487 | ref := seedScratchFull() |
| 488 | jsonError(w, ref.Status, ref.Msg) |
| 489 | return false |
| 490 | } |
| 491 | h := sha256.New() |
| 492 | n, cerr := io.Copy(io.MultiWriter(f, h), http.MaxBytesReader(w, r.Body, seed.Bytes+1)) |
| 493 | ferr := f.Close() |
| 494 | if importSpoolHook != nil { |
| 495 | importSpoolHook(spool) |
| 496 | } |
| 497 | var tooLong *http.MaxBytesError |
| 498 | switch { |
| 499 | case cerr != nil && !errors.As(cerr, &tooLong): |
| 500 | // The caller went away, the body broke off, or the spool's disk |
| 501 | // failed: nothing to keep, and the import still waits. |
| 502 | log.Printf("console: import %s: read the upload: %v", host, cerr) |
| 503 | jsonError(w, http.StatusBadGateway, importFailed) |
| 504 | return false |
| 505 | case ferr != nil: |
| 506 | log.Printf("console: import %s: spool: %v", host, ferr) |
| 507 | jsonError(w, http.StatusBadGateway, importFailed) |
| 508 | return false |
| 509 | case tooLong != nil || n != seed.Bytes || hex.EncodeToString(h.Sum(nil)) != seed.SHA256: |
| 510 | jsonError(w, http.StatusUnprocessableEntity, "the upload does not match the length and sha256 this import began with, so it was not kept.") |
| 511 | return false |
| 512 | } |
| 513 | return true |
| 514 | } |
| 515 | |
| 516 | // runImportUpload is the detached half: the integrity check, the put to |
| 517 | // the seed's snapshot key, and the move to uploaded. It never depends on |
| 518 | // the request, which has been answered. A failure moves the record to |
| 519 | // aborted with the sentence the door would have answered in Seed.Reason |
| 520 | // (SD-N52, SD-N61 or SD-N67), but only while the record still waits for |
| 521 | // this upload (SeedAbortUploadBecause): an import that has moved on, by |
| 522 | // another upload, an abort or a delete, is never touched, and neither is |
| 523 | // a snapshot already at its key (fix round 1). On every exit it removes |
| 524 | // the spool's directory and then releases the reservation, as runFrom |
| 525 | // does. |
| 526 | // |
| 527 | // The console has no shutdown hook to cancel a running check: a restart |
| 528 | // leaves the record awaiting-upload, its spool is cleared at start, and |
| 529 | // the sweep aborts it (Plan 8). |
| 530 | func (s *Server) runImportUpload(j *importUpload) { |
| 531 | defer s.seedCopies.Done() |
| 532 | defer j.release() |
| 533 | ctx := j.ctx |
| 534 | sqid, app, host := j.acct.Sqid, j.app, j.host |
| 535 | abort := func(reason, why string) { |
| 536 | bg, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 537 | defer cancel() |
| 538 | _, err := provision.SetRecordSeed(bg, s.cfg.Releases, sqid, app, host, provision.SeedAbortUploadBecause(j.id, reason)) |
| 539 | switch { |
| 540 | case errors.Is(err, provision.ErrSeedConflict), errors.Is(err, os.ErrNotExist): |
| 541 | log.Printf("console: import %s: seed %s moved on before its upload's %s was recorded; not aborted: %v", host, j.id, why, err) |
| 542 | return |
| 543 | case err != nil: |
| 544 | log.Printf("console: import %s: abort seed %s after %s: %v", host, j.id, why, err) |
| 545 | } |
| 546 | s.bumpSerial(bg) |
| 547 | } |
| 548 | // A panic in the check or the put would end the whole console: it |
| 549 | // aborts this seed instead. Registered after release and Done, so it |
| 550 | // runs before them; the spool's removal, registered later, has |
| 551 | // already run. |
| 552 | defer func() { |
| 553 | if p := recover(); p != nil { |
| 554 | log.Printf("console: import %s: panic in the upload of seed %s: %v\n%s", host, j.id, p, debug.Stack()) |
| 555 | abort(importFailed, "a panic") |
| 556 | } |
| 557 | }() |
| 558 | defer os.RemoveAll(j.dir) // with any -shm or -wal the check left |
| 559 | |
| 560 | cctx, ccancel := context.WithTimeout(ctx, importCheckDeadline) |
| 561 | err := checkSeedDBConsole(cctx, j.spool) |
| 562 | timedOut := errors.Is(cctx.Err(), context.DeadlineExceeded) |
| 563 | ccancel() |
| 564 | if err != nil { |
| 565 | if timedOut { |
| 566 | abort("checking the upload took longer than 30 minutes, so it was not kept.", "the check's deadline") |
| 567 | return |
| 568 | } |
| 569 | log.Printf("console: import %s: check: %v", host, err) |
| 570 | abort("the upload is not a database that passes an integrity check, so it was not kept.", "the check") |
| 571 | return |
| 572 | } |
| 573 | // The check can take half an hour: put nothing for an import that has |
| 574 | // moved on meanwhile (finished, aborted, or landed by another upload). |
| 575 | rec, err := s.readImportRecord(ctx, sqid, app, host) |
| 576 | switch { |
| 577 | case errors.Is(err, os.ErrNotExist): |
| 578 | log.Printf("console: import %s: the record went while seed %s was checked; nothing put", host, j.id) |
| 579 | return |
| 580 | case err != nil: |
| 581 | log.Printf("console: import %s: re-read before the put: %v", host, err) |
| 582 | abort(importFailed, "the re-read") |
| 583 | return |
| 584 | case !awaitingUpload(rec, j.id): |
| 585 | log.Printf("console: import %s: seed %s moved on while it was checked (%+v); nothing put", host, j.id, rec.Seed) |
| 586 | return |
| 587 | } |
| 588 | uctx, ucancel := context.WithTimeout(ctx, seedUploadDeadline) |
| 589 | err = putSnapshot(uctx, s.cfg.SeedData, j.spool, seeds.SnapshotKey(host, j.id)) |
| 590 | ucancel() |
| 591 | if errors.Is(err, release.ErrCASConflict) { |
| 592 | // Another writer put this seed's snapshot first, and owns the |
| 593 | // outcome. Nothing is overwritten and nothing aborted. |
| 594 | log.Printf("console: import %s: the snapshot of seed %s is already there; left as it is", host, j.id) |
| 595 | return |
| 596 | } |
| 597 | if err != nil { |
| 598 | log.Printf("console: import %s: put the snapshot: %v", host, err) |
| 599 | abort(importFailed, "the put") |
| 600 | return |
| 601 | } |
| 602 | pctx, pcancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 603 | defer pcancel() |
| 604 | _, err = provision.SetRecordSeed(pctx, s.cfg.Releases, sqid, app, host, provision.SeedUploaded(j.id)) |
| 605 | switch { |
| 606 | case errors.Is(err, provision.ErrSeedConflict), errors.Is(err, os.ErrNotExist): |
| 607 | log.Printf("console: import %s: seed %s moved on before it was marked uploaded; not aborted: %v", host, j.id, err) |
| 608 | return |
| 609 | case err != nil: |
| 610 | log.Printf("console: import %s: record uploaded: %v", host, err) |
| 611 | abort(importFailed, "the record write") |
| 612 | return |
| 613 | } |
| 614 | s.auditActivity(j.acct, app, j.by, Activity{Action: "import-uploaded", TargetType: "instance", Target: host, |
| 615 | Detail: fmt.Sprintf("uploaded the snapshot for %s", host)}) |
| 616 | } |
| 617 | |
| 618 | // putSnapshot writes path to key with one create-only PutFrom (spec §5.2 |
| 619 | // step 4). It never replaces an object already there: that answers |
| 620 | // release.ErrCASConflict. |
| 621 | func putSnapshot(ctx context.Context, data release.Store, path, key string) error { |
| 622 | f, err := os.Open(path) |
| 623 | if err != nil { |
| 624 | return err |
| 625 | } |
| 626 | defer f.Close() |
| 627 | st, err := f.Stat() |
| 628 | if err != nil { |
| 629 | return err |
| 630 | } |
| 631 | _, err = release.PutObjectFrom(ctx, data, key, f, st.Size(), "") |
| 632 | return err |
| 633 | } |
| 634 | |
| 635 | // notAnInstanceOf is the 404 sentence for a record that is not there. |
| 636 | func notAnInstanceOf(host, app string) string { |
| 637 | return fmt.Sprintf("%s is not an instance of %s", host, app) |
| 638 | } |
| 639 | |
| 640 | // lostTransition reports whether err is a transition that found the |
| 641 | // import elsewhere, or a record write that lost its race twice. |
| 642 | func lostTransition(err error) bool { |
| 643 | return errors.Is(err, provision.ErrSeedConflict) || errors.Is(err, release.ErrCASConflict) |
| 644 | } |
| 645 | |
| 646 | // importNow re-reads host's record once, after a transition lost, and |
| 647 | // returns its seed when it is still seed id. Nil means the import is gone, |
| 648 | // is another seed, or could not be read: the caller answers SD-N60. |
| 649 | func (s *Server) importNow(ctx context.Context, sqid, app, host, id string) *provision.SeedSource { |
| 650 | rec, err := s.readImportRecord(ctx, sqid, app, host) |
| 651 | if err != nil || rec.Seed == nil || rec.Seed.ID != id { |
| 652 | return nil |
| 653 | } |
| 654 | return rec.Seed |
| 655 | } |
| 656 | |
| 657 | // importTransitionFailure answers a synchronous transition of the import's |
| 658 | // record that did not land. A guarded transition that finds the import |
| 659 | // somewhere else (provision.ErrSeedConflict), or a record write that lost |
| 660 | // its race twice, means another request got there first: SD-N60, and the |
| 661 | // caller reads the state and decides. A record that is gone is the 404. |
| 662 | // Anything else is the retry sentence, with the cause logged. |
| 663 | func (s *Server) importTransitionFailure(w http.ResponseWriter, app, host string, err error) { |
| 664 | switch { |
| 665 | case lostTransition(err): |
| 666 | jsonError(w, http.StatusConflict, fmt.Sprintf("%s changed while this request was made. Read its state with carlos instances list and try again.", host)) |
| 667 | case errors.Is(err, os.ErrNotExist): |
| 668 | jsonError(w, http.StatusNotFound, notAnInstanceOf(host, app)) |
| 669 | default: |
| 670 | log.Printf("console: import %s: write the record: %v", host, err) |
| 671 | jsonError(w, http.StatusBadGateway, importStepFailure(host)) |
| 672 | } |
| 673 | } |
| 674 | |
| 675 | // handleAPIImportStopped is the app saying its own process is down (spec |
| 676 | // §5.3): uploaded to ready, and a serial bump so the box swaps. The |
| 677 | // transition is guarded on the seed id and on uploaded, so a stopped that |
| 678 | // races an abort, a delete or a second stopped changes nothing but its own |
| 679 | // answer; the check below only picks the sentence. |
| 680 | func (s *Server) handleAPIImportStopped(w http.ResponseWriter, r *http.Request) { |
| 681 | principal, acct, app, host, rec, ok := s.importGate(w, r) |
| 682 | if !ok { |
| 683 | return |
| 684 | } |
| 685 | if rec.Seed.State != seeds.StateUploaded { |
| 686 | jsonError(w, http.StatusConflict, fmt.Sprintf("%s is %s. Mark it stopped once, after its upload.", host, rec.Seed.State)) |
| 687 | return |
| 688 | } |
| 689 | checks, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout) |
| 690 | ref := s.seedRegionReady(checks, rec.Region, true) |
| 691 | cancel() |
| 692 | if ref != nil { |
| 693 | jsonError(w, ref.Status, ref.Msg) |
| 694 | return |
| 695 | } |
| 696 | // Not the request's cancellation: a client leaving now must not turn a |
| 697 | // transition that landed into a missing bump. |
| 698 | wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout) |
| 699 | defer wcancel() |
| 700 | if _, err := provision.SetRecordSeed(wctx, s.cfg.Releases, acct.Sqid, app, host, provision.SeedStopped(rec.Seed.ID)); err != nil { |
| 701 | // An overlapping stopped may have won: this import is past stopped |
| 702 | // now, and a sequential second stopped is told so (SD-N54). That |
| 703 | // is not success: the caller reads 409 naming ready or later as |
| 704 | // the stop having landed. Any other loss is SD-N60. |
| 705 | if lostTransition(err) { |
| 706 | if now := s.importNow(wctx, acct.Sqid, app, host, rec.Seed.ID); now != nil && pastStopped(now.State) { |
| 707 | jsonError(w, http.StatusConflict, fmt.Sprintf("%s is %s. Mark it stopped once, after its upload.", host, now.State)) |
| 708 | return |
| 709 | } |
| 710 | } |
| 711 | s.importTransitionFailure(w, app, host, err) |
| 712 | return |
| 713 | } |
| 714 | s.auditActivity(acct, app, principal, Activity{Action: "import-stopped", TargetType: "instance", Target: host, |
| 715 | Detail: fmt.Sprintf("marked the old process of %s stopped", host)}) |
| 716 | s.bumpSerial(wctx) |
| 717 | writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateReady}) |
| 718 | } |
| 719 | |
| 720 | // handleAPIImportCommit is the app's verdict that the instance serves |
| 721 | // (spec §5.4): placed to committed, then the box clears the gate. A commit |
| 722 | // is a record transition and nothing else: the verify token is what the |
| 723 | // edge checks on the app's own requests (spec §5.5), and this door never |
| 724 | // sees it. Committed already answers 200, so a retried commit is safe. |
| 725 | func (s *Server) handleAPIImportCommit(w http.ResponseWriter, r *http.Request) { |
| 726 | principal, acct, app, host, rec, ok := s.importGate(w, r) |
| 727 | if !ok { |
| 728 | return |
| 729 | } |
| 730 | switch rec.Seed.State { |
| 731 | case seeds.StateCommitted: |
| 732 | writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateCommitted}) |
| 733 | return |
| 734 | case seeds.StatePlaced: |
| 735 | case seeds.StateAwaitingUpload, seeds.StateUploaded, seeds.StateReady, seeds.StateSwapped: |
| 736 | jsonError(w, http.StatusConflict, "nothing has been verified yet. Send a request with the verify token first, then commit.") |
| 737 | return |
| 738 | default: |
| 739 | jsonError(w, http.StatusConflict, fmt.Sprintf("%s is %s, so there is nothing to commit.", host, rec.Seed.State)) |
| 740 | return |
| 741 | } |
| 742 | wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout) |
| 743 | defer wcancel() |
| 744 | if _, err := provision.SetRecordSeed(wctx, s.cfg.Releases, acct.Sqid, app, host, provision.SeedCommit(rec.Seed.ID)); err != nil { |
| 745 | // An overlapping commit may have won, between the gate's read and |
| 746 | // this write: the import is committed, and this is a retry like any |
| 747 | // other. The winner audited and bumped, so this writes nothing. |
| 748 | if lostTransition(err) { |
| 749 | if now := s.importNow(wctx, acct.Sqid, app, host, rec.Seed.ID); now != nil && now.State == seeds.StateCommitted { |
| 750 | writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateCommitted}) |
| 751 | return |
| 752 | } |
| 753 | } |
| 754 | s.importTransitionFailure(w, app, host, err) |
| 755 | return |
| 756 | } |
| 757 | s.auditActivity(acct, app, principal, Activity{Action: "import-committed", TargetType: "instance", Target: host, |
| 758 | Detail: fmt.Sprintf("committed the import of %s", host)}) |
| 759 | s.bumpSerial(wctx) |
| 760 | writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateCommitted}) |
| 761 | } |
| 762 | |
| 763 | // pastStopped reports whether an import's state is one it only reaches |
| 764 | // after a stopped: the box has the stop, or has gone further. |
| 765 | func pastStopped(state string) bool { |
| 766 | switch state { |
| 767 | case seeds.StateReady, seeds.StateSwapped, seeds.StatePlaced, seeds.StateCommitted: |
| 768 | return true |
| 769 | } |
| 770 | return false |
| 771 | } |
| 772 | |
| 773 | // importDeleteFailed is the delete door's retry sentence when its import |
| 774 | // rule cannot read the record or finish the import (preflight M15). Main's |
| 775 | // own delete sentence (apiinstances.go) is unchanged. |
| 776 | const importDeleteFailed = "could not delete the instance. Try again." |
| 777 | |
| 778 | // importFinishTimeout bounds a finish run inline for a request: the |
| 779 | // store's teardown has its own storeTeardownTimeout, and the reads, the |
| 780 | // snapshot delete and the record delete around it need a little more. |
| 781 | const importFinishTimeout = storeTeardownTimeout + instanceStoreTimeout |
| 782 | |
| 783 | // committedRefusal is abort after commit (SD-N56). |
| 784 | func committedRefusal(host string) string { |
| 785 | return fmt.Sprintf("%s is committed. Delete it with carlos instances delete instead.", host) |
| 786 | } |
| 787 | |
| 788 | // finishImport finishes a rolled-back import inline for a request: the |
| 789 | // request's values, never its cancellation, so a client that leaves cannot |
| 790 | // cut a teardown midway (preflight M15). |
| 791 | func (s *Server) finishImport(r *http.Request, sqid, app, host, id string) error { |
| 792 | ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), importFinishTimeout) |
| 793 | defer cancel() |
| 794 | return s.finishSeed(ctx, sqid, app, host, id) |
| 795 | } |
| 796 | |
| 797 | // readDeletedRecord is the delete door's read of host's record for its |
| 798 | // import rule, with the etag the ordinary delete is then bound to. Only |
| 799 | // the store read can fail it (preflight M15, review I1): a body that will |
| 800 | // not decode, or that disagrees with its own path, is no import this |
| 801 | // console can finish, so it answers nil and the ordinary delete removes |
| 802 | // it as it always has. |
| 803 | func (s *Server) readDeletedRecord(ctx context.Context, sqid, app, host string) (*provision.Record, string, error) { |
| 804 | ctx, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 805 | defer cancel() |
| 806 | raw, etag, err := s.cfg.Releases.Get(ctx, provision.RecordKey(sqid, app, host)) |
| 807 | if err != nil { |
| 808 | return nil, "", err |
| 809 | } |
| 810 | rec, err := provision.DecodeRecord(raw) |
| 811 | if err == nil && (rec.Account != sqid || rec.App != app || rec.Host != host) { |
| 812 | err = errors.New("the record disagrees with its own path") |
| 813 | } |
| 814 | if err != nil { |
| 815 | log.Printf("console: api instances delete %s/%s/%s: the record could not be read as an import (%v), so it is deleted the ordinary way", sqid, app, host, err) |
| 816 | return nil, etag, nil |
| 817 | } |
| 818 | return &rec, etag, nil |
| 819 | } |
| 820 | |
| 821 | // handleAPIImportAbort abandons an import before it is committed (spec |
| 822 | // §5.6): a record CAS to aborted, then the box rolls it back. The |
| 823 | // transition is provision.SeedAbort, which allows every state before |
| 824 | // committed and refuses while a run is open, in the same CAS (preflight |
| 825 | // M14: one rule, in provision). The no-reason form keeps a refused |
| 826 | // import's reason. |
| 827 | // |
| 828 | // It is safe to repeat. An import already aborted answers 200 aborted; one |
| 829 | // the box has rolled back is finished here (its store, its snapshot, then |
| 830 | // its record) and answers 200 rolled-back, even when the finish fails and |
| 831 | // is left to the next abort or the sweep. An abort that loses its race |
| 832 | // re-reads once and answers by what won. |
| 833 | func (s *Server) handleAPIImportAbort(w http.ResponseWriter, r *http.Request) { |
| 834 | principal, acct, app, host, rec, ok := s.importGate(w, r) |
| 835 | if !ok { |
| 836 | return |
| 837 | } |
| 838 | id := rec.Seed.ID |
| 839 | if s.abortAnswered(w, r, acct.Sqid, app, host, rec.Seed) { |
| 840 | return |
| 841 | } |
| 842 | wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout) |
| 843 | defer wcancel() |
| 844 | _, err := provision.SetRecordSeed(wctx, s.cfg.Releases, acct.Sqid, app, host, provision.SeedAbort(id)) |
| 845 | switch { |
| 846 | case errors.Is(err, provision.ErrRunOpen): |
| 847 | jsonError(w, http.StatusConflict, fmt.Sprintf("carlos run is busy on %s. Wait for it to finish, then abort.", host)) |
| 848 | return |
| 849 | case lostTransition(err): |
| 850 | // Another request got there first. Answer as a sequential abort |
| 851 | // would be answered after it; anything but this import is SD-N60. |
| 852 | if now := s.importNow(wctx, acct.Sqid, app, host, id); now != nil && s.abortAnswered(w, r, acct.Sqid, app, host, now) { |
| 853 | return |
| 854 | } |
| 855 | s.importTransitionFailure(w, app, host, err) |
| 856 | return |
| 857 | case err != nil: |
| 858 | s.importTransitionFailure(w, app, host, err) |
| 859 | return |
| 860 | } |
| 861 | s.auditActivity(acct, app, principal, Activity{Action: "import-aborted", TargetType: "instance", Target: host, |
| 862 | Detail: fmt.Sprintf("aborted the import of %s", host)}) |
| 863 | s.bumpSerial(wctx) |
| 864 | writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateAborted}) |
| 865 | } |
| 866 | |
| 867 | // abortAnswered answers an abort of seed that needs no transition, and |
| 868 | // reports whether it did: committed is refused (SD-N56), aborted is done |
| 869 | // already, and rolled-back is finished here. A finish that fails is |
| 870 | // logged and still answers 200: the abort itself has landed, and the next |
| 871 | // abort or the sweep finishes it. |
| 872 | func (s *Server) abortAnswered(w http.ResponseWriter, r *http.Request, sqid, app, host string, seed *provision.SeedSource) bool { |
| 873 | switch seed.State { |
| 874 | case seeds.StateCommitted: |
| 875 | jsonError(w, http.StatusConflict, committedRefusal(host)) |
| 876 | case seeds.StateAborted: |
| 877 | writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateAborted}) |
| 878 | case seeds.StateRolledBack: |
| 879 | if err := s.finishImport(r, sqid, app, host, seed.ID); err != nil { |
| 880 | log.Printf("console: import %s: finish seed %s on abort: %v", host, seed.ID, err) |
| 881 | } |
| 882 | writeJSON(w, http.StatusOK, map[string]string{"state": seeds.StateRolledBack}) |
| 883 | default: |
| 884 | return false |
| 885 | } |
| 886 | return true |
| 887 | } |
| 888 | |
| 889 | // refuseImportDelete is the delete door's import rule (spec §5.6, codex |
| 890 | // R2-7), run before the ordinary delete. answered: the delete must not go |
| 891 | // the ordinary way, and either this function has answered it (a refusal or |
| 892 | // a failure), or, with finished, the delete was done here (the import |
| 893 | // finished, or the ordinary delete bound to this read) and the caller |
| 894 | // writes the ordinary delete's audit row and answer. (false, false) is |
| 895 | // left only to a claim console, whose ordinary delete is the claim's. |
| 896 | // |
| 897 | // - awaiting-upload, uploaded, refused: no box has swapped, so the record |
| 898 | // moves straight to rolled-back and is finished here (its store, its |
| 899 | // snapshot under seeds/<host>/<id>/, then the record). |
| 900 | // - rolled-back: finished here. |
| 901 | // - any other uncommitted state: 409, abort the import instead (SD-N57). |
| 902 | // - committed, a --from seed, an unseeded record, or a record that will |
| 903 | // not decode: the ordinary delete, conditional on the etag read here |
| 904 | // (deleteRead), answered as finished. |
| 905 | // - no record at all: finished, with nothing deleted (final review F7), |
| 906 | // except on a claim console, where the claim's retirement answers. |
| 907 | // |
| 908 | // It fails closed (preflight M15): a store read that fails, a seed id that |
| 909 | // cannot name a key (M8) and a finish that fails answer main's delete |
| 910 | // sentence, so a read blip never lets the ordinary delete drop an |
| 911 | // uncommitted import's record and leak what its seed made. A body that |
| 912 | // was read but will not decode, or disagrees with its path, is no import |
| 913 | // anything can finish, so the ordinary delete removes it as before |
| 914 | // (review I1). On a claim |
| 915 | // console, where the finish runs nowhere, an uncommitted import is refused |
| 916 | // as every import door refuses there. |
| 917 | func (s *Server) refuseImportDelete(w http.ResponseWriter, r *http.Request, acct *Account, app, host string) (answered, finished bool) { |
| 918 | return s.importDeleteRule(w, r, acct, app, host, true) |
| 919 | } |
| 920 | |
| 921 | // importDeleteRule is refuseImportDelete; again is whether a record that |
| 922 | // changed under the ordinary delete may be read and decided once more. |
| 923 | // |
| 924 | // The ordinary delete is bound to what the rule read (final review F7): |
| 925 | // absence is the delete's own idempotent answer, so no delete runs that |
| 926 | // could remove an import begun since; and otherwise the record is deleted |
| 927 | // only while it is the one the rule read (deleteRead). |
| 928 | func (s *Server) importDeleteRule(w http.ResponseWriter, r *http.Request, acct *Account, app, host string, again bool) (answered, finished bool) { |
| 929 | sqid := acct.Sqid |
| 930 | rec, etag, err := s.readDeletedRecord(r.Context(), sqid, app, host) |
| 931 | if errors.Is(err, os.ErrNotExist) { |
| 932 | // With host claims on, a record can be gone while its retirement |
| 933 | // is not (parking, with obligations left): the claim's own flow |
| 934 | // answers that, pending or done (re-review finding 10). |
| 935 | if s.cfg.HostClaim { |
| 936 | return false, false |
| 937 | } |
| 938 | return true, true |
| 939 | } |
| 940 | if err != nil { |
| 941 | log.Printf("console: api instances delete %s/%s/%s: read the record: %v", sqid, app, host, err) |
| 942 | jsonError(w, http.StatusBadGateway, importDeleteFailed) |
| 943 | return true, false |
| 944 | } |
| 945 | if rec == nil || !rec.Seed.Import() || rec.Seed.State == seeds.StateCommitted { |
| 946 | return s.deleteRead(w, r, acct, app, host, etag, again) |
| 947 | } |
| 948 | if s.cfg.HostClaim { |
| 949 | ref := seedClaimConsoleRefusal() |
| 950 | jsonError(w, ref.Status, ref.Msg) |
| 951 | return true, false |
| 952 | } |
| 953 | id := rec.Seed.ID |
| 954 | if err := seeds.Check(host, id); err != nil { |
| 955 | log.Printf("console: api instances delete %s/%s/%s: import seed %q: %v", sqid, app, host, id, err) |
| 956 | jsonError(w, http.StatusBadGateway, importDeleteFailed) |
| 957 | return true, false |
| 958 | } |
| 959 | switch rec.Seed.State { |
| 960 | case seeds.StateAwaitingUpload, seeds.StateUploaded, seeds.StateRefused: |
| 961 | wctx, wcancel := context.WithTimeout(context.WithoutCancel(r.Context()), instanceStoreTimeout) |
| 962 | _, err := provision.SetRecordSeed(wctx, s.cfg.Releases, sqid, app, host, provision.SeedRolledBackBeforeSwap(id)) |
| 963 | var now *provision.Record |
| 964 | var nowErr error |
| 965 | if lostTransition(err) { |
| 966 | now, nowErr = s.readImportRecord(wctx, sqid, app, host) |
| 967 | } |
| 968 | wcancel() |
| 969 | switch { |
| 970 | case err == nil: |
| 971 | case errors.Is(err, os.ErrNotExist), errors.Is(nowErr, os.ErrNotExist): |
| 972 | // Gone already, by this CAS's read or by another delete that |
| 973 | // finished first: the delete's idempotent answer, with no |
| 974 | // ordinary delete run after it (final review F7). |
| 975 | return true, true |
| 976 | case now != nil && now.Seed != nil && now.Seed.ID == id && now.Seed.State == seeds.StateRolledBack: |
| 977 | // Another delete rolled it back first: finish it all the same. |
| 978 | case lostTransition(err): |
| 979 | s.importTransitionFailure(w, app, host, err) // SD-N60 |
| 980 | return true, false |
| 981 | default: |
| 982 | log.Printf("console: api instances delete %s/%s/%s: roll back import seed %s: %v", sqid, app, host, id, err) |
| 983 | jsonError(w, http.StatusBadGateway, importDeleteFailed) |
| 984 | return true, false |
| 985 | } |
| 986 | case seeds.StateRolledBack: |
| 987 | case seeds.StateAborted: |
| 988 | // Aborting is done: the box puts the old route back, and then a |
| 989 | // delete finishes the import. |
| 990 | jsonError(w, http.StatusConflict, fmt.Sprintf("the import of %s is already aborted. Once the box puts the old route back on its next pass, delete %s again.", host, host)) |
| 991 | return true, false |
| 992 | default: |
| 993 | jsonError(w, http.StatusConflict, fmt.Sprintf("%s is an import that is not committed. Abort the import instead.", host)) |
| 994 | return true, false |
| 995 | } |
| 996 | if err := s.finishImport(r, sqid, app, host, id); err != nil { |
| 997 | log.Printf("console: api instances delete %s/%s/%s: finish import seed %s: %v", sqid, app, host, id, err) |
| 998 | jsonError(w, http.StatusBadGateway, importDeleteFailed) |
| 999 | return true, false |
| 1000 | } |
| 1001 | return true, true |
| 1002 | } |
| 1003 | |
| 1004 | // deleteRead is the ordinary delete of host's record, conditional on the |
| 1005 | // etag the import rule read it under (final review F7). A record gone |
| 1006 | // since is deleted already. A record that changed since may now be an |
| 1007 | // import begun after the rule's read: it is read and decided again, once, |
| 1008 | // and a second change answers main's retry sentence. |
| 1009 | // |
| 1010 | // A claim console retires records through the claim |
| 1011 | // (retireClaimHostRecord) and begins no import, so there the ordinary |
| 1012 | // delete runs as before: (false, false). |
| 1013 | func (s *Server) deleteRead(w http.ResponseWriter, r *http.Request, acct *Account, app, host, etag string, again bool) (answered, finished bool) { |
| 1014 | if s.cfg.HostClaim { |
| 1015 | return false, false |
| 1016 | } |
| 1017 | ctx, cancel := context.WithTimeout(r.Context(), instanceStoreTimeout) |
| 1018 | defer cancel() |
| 1019 | err := release.DeleteIfMatch(ctx, s.cfg.Releases, provision.RecordKey(acct.Sqid, app, host), etag) |
| 1020 | switch { |
| 1021 | case err == nil: |
| 1022 | return true, true |
| 1023 | case errors.Is(err, release.ErrCASConflict) && again: |
| 1024 | return s.importDeleteRule(w, r, acct, app, host, false) |
| 1025 | } |
| 1026 | log.Printf("console: api instances delete %s/%s/%s: %v", acct.Sqid, app, host, err) |
| 1027 | jsonError(w, http.StatusBadGateway, importDeleteFailed) |
| 1028 | return true, false |
| 1029 | } |
| 1030 | |
amadan