| 1 | // The --from seed door (seeded-instances spec §4): `carlos instances create |
| 2 | // --from <host>` creates an instance whose database starts as a copy of a |
| 3 | // sibling's replica. handleAPIInstanceCreate runs every ordinary check, |
| 4 | // then hands over here. The door answers 202 once the record is reserved, |
| 5 | // and the copy finishes on its own (createFrom). |
| 6 | |
| 7 | package console |
| 8 | |
| 9 | import ( |
| 10 | "bytes" |
| 11 | "context" |
| 12 | "crypto/sha256" |
| 13 | "encoding/hex" |
| 14 | "encoding/json" |
| 15 | "errors" |
| 16 | "fmt" |
| 17 | "io" |
| 18 | "io/fs" |
| 19 | "log" |
| 20 | "math" |
| 21 | "net/http" |
| 22 | "os" |
| 23 | "path/filepath" |
| 24 | "runtime/debug" |
| 25 | "strconv" |
| 26 | "strings" |
| 27 | "sync/atomic" |
| 28 | "time" |
| 29 | |
| 30 | "github.com/carlosframework/platform/internal/dbdoor" |
| 31 | "github.com/carlosframework/platform/internal/objstore" |
| 32 | "github.com/carlosframework/platform/internal/provision" |
| 33 | "github.com/carlosframework/platform/internal/registry" |
| 34 | "github.com/carlosframework/platform/internal/release" |
| 35 | "github.com/carlosframework/platform/internal/seeds" |
| 36 | ) |
| 37 | |
| 38 | // Seams for the tests. |
| 39 | var ( |
| 40 | seedMaxBytes = seeds.MaxBytes |
| 41 | planRestore = dbdoor.PlanRestore |
| 42 | restoreToTarget = dbdoor.RestoreToTarget |
| 43 | // seedRestoreDeadline bounds the restore and its check (spec §4 step 5). |
| 44 | seedRestoreDeadline = 30 * time.Minute |
| 45 | // restoreWatchEvery is how often the size watcher looks (spec §4 step 5). |
| 46 | restoreWatchEvery = time.Second |
| 47 | ) |
| 48 | |
| 49 | // seedUploadDeadline bounds the snapshot's one PUT, after the restore's own |
| 50 | // deadline: up to 4 GiB to the region's own bucket. The record is reserving |
| 51 | // for the whole restore, store walk and upload, and the sweep aborts a |
| 52 | // --from that has been reserving for an hour (spec §3.1), so the three |
| 53 | // together stay well under that. |
| 54 | const seedUploadDeadline = 20 * time.Minute |
| 55 | |
| 56 | // seedFromReq is the create body's "from" (spec §4). TXID 0 is the latest |
| 57 | // copy; the CLI refuses a non-positive --at. |
| 58 | type seedFromReq struct { |
| 59 | Host string `json:"host"` |
| 60 | TXID uint64 `json:"txid,omitempty"` |
| 61 | } |
| 62 | |
| 63 | // seedCreate is what handleAPIInstanceCreate has decided by the time it |
| 64 | // hands a seed door over: every ordinary check has passed. siblingLabel is |
| 65 | // the create door's own (set when the host sits at another label of the |
| 66 | // account's platform namespace), for its re-check before publishing. |
| 67 | type seedCreate struct { |
| 68 | acct *Account |
| 69 | app, principal, message string |
| 70 | m *provision.Manifest |
| 71 | rec provision.Record |
| 72 | placed, regionFrom string |
| 73 | siblingLabel string |
| 74 | } |
| 75 | |
| 76 | // seedCommonRefusal is both seed doors' shape rule: an ordinary instance. |
| 77 | func seedCommonRefusal(kind string, addrs []string, role string, replicas []string) *doorRefusal { |
| 78 | if (kind != "" && kind != "instance") || len(addrs) > 0 || role != "" || len(replicas) > 0 { |
| 79 | return seedOrdinaryRefusal() |
| 80 | } |
| 81 | return nil |
| 82 | } |
| 83 | |
| 84 | // seedOrdinaryRefusal is SD-N36, the shape rule's answer. |
| 85 | func seedOrdinaryRefusal() *doorRefusal { |
| 86 | return &doorRefusal{http.StatusBadRequest, "a seeded instance is an ordinary instance. Drop --kind, --addr and --role."} |
| 87 | } |
| 88 | |
| 89 | // seedNoDataRefusal is both doors' answer on a console with no data bucket |
| 90 | // for seeds. |
| 91 | func seedNoDataRefusal() *doorRefusal { |
| 92 | return &doorRefusal{http.StatusServiceUnavailable, "this console has no data bucket set up, so it cannot create a seeded instance."} |
| 93 | } |
| 94 | |
| 95 | // seedPlacementRefusal is both seed doors' placement rule: the platform |
| 96 | // keeps a replica only of an instance it hosts itself. |
| 97 | func seedPlacementRefusal(c seedCreate) *doorRefusal { |
| 98 | if c.placed != "" || c.m.Pool != "" { |
| 99 | return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s places its instances in a pool or on a fleet, so it cannot take a seeded instance.", c.app)} |
| 100 | } |
| 101 | return nil |
| 102 | } |
| 103 | |
| 104 | // seedOwnEnvironment is SD-N33, the rule a seeded instance's environment |
| 105 | // keeps (spec §4 Storage). |
| 106 | const seedOwnEnvironment = "a seeded instance gets its own environment and an empty store" |
| 107 | |
| 108 | // refuseSeededEnvironment is both seed doors' answer to an explicit |
| 109 | // --environment on an app with private storage (spec §4 Storage). |
| 110 | func (s *Server) refuseSeededEnvironment(w http.ResponseWriter) { |
| 111 | jsonError(w, http.StatusConflict, seedOwnEnvironment) |
| 112 | } |
| 113 | |
| 114 | // seedStoreTakenRefusal is the refusal of a seeded create whose environment |
| 115 | // is not new: SD-N34 naming the store's bucket, or, when the environment |
| 116 | // has no store with a bucket and is only named by another instance (final |
| 117 | // review F6), the sentence that says so and what to do. |
| 118 | func seedStoreTakenRefusal(env, app string, store *objstore.Record) *doorRefusal { |
| 119 | if bucket := storeBucketName(store); bucket != "" { |
| 120 | return &doorRefusal{http.StatusConflict, fmt.Sprintf("environment %s already has a store (%s). A seeded instance needs a new, empty one.", env, bucket)} |
| 121 | } |
| 122 | return &doorRefusal{http.StatusConflict, fmt.Sprintf("environment %s is already used by another instance of %s. A seeded instance gets a new environment named after its host, so choose another host.", env, app)} |
| 123 | } |
| 124 | |
| 125 | // seedSiblingApexReason is the sentence both seed doors speak when an app |
| 126 | // created under the host's label while the instance was being written has |
| 127 | // made the host its platform apex. Main's create door keeps its own. |
| 128 | func seedSiblingApexReason(host, label string) string { |
| 129 | return fmt.Sprintf("%s is app %s's platform apex. That app was created while this instance was being written, so the instance was not kept.", host, label) |
| 130 | } |
| 131 | |
| 132 | // errSeedEnvironmentUnread is instanceStoreProvisionFor's answer when it |
| 133 | // cannot tell whether another instance names a seed's environment. It |
| 134 | // answers nothing on w: the seed door speaks its own retry sentence. |
| 135 | var errSeedEnvironmentUnread = errors.New("console: the instances naming the environment could not be read") |
| 136 | |
| 137 | // seedEnvironmentUser names an instance of app, other than seed seedID's |
| 138 | // own, whose record names env, or "" when none does (final review F6). A |
| 139 | // seeded instance's derived environment must be unused: its store is new |
| 140 | // and empty and its seed's finish may tear it down, which another |
| 141 | // instance's environment must never be. It reads strictly |
| 142 | // (provision.ListRecordsStrict): a listing it cannot complete, too many |
| 143 | // records, or a record that will not decode is an error, since any of them |
| 144 | // may name env. |
| 145 | func (s *Server) seedEnvironmentUser(ctx context.Context, sqid, app, env, seedID string) (string, error) { |
| 146 | recs, unreadable, err := provision.ListRecordsStrict(ctx, s.cfg.Releases, sqid, app, instanceListMax) |
| 147 | if err != nil { |
| 148 | return "", err |
| 149 | } |
| 150 | if len(unreadable) > 0 { |
| 151 | return "", fmt.Errorf("%d records of %s/%s could not be read, so one may name environment %q", len(unreadable), sqid, app, env) |
| 152 | } |
| 153 | for _, r := range recs { |
| 154 | if r.Environment == env && (r.Seed == nil || r.Seed.ID != seedID) { |
| 155 | return r.Host, nil |
| 156 | } |
| 157 | } |
| 158 | return "", nil |
| 159 | } |
| 160 | |
| 161 | // seedStoreRefusal is spec §4 Storage's existing-store rule for both doors, |
| 162 | // checked before the record is reserved. instanceStoreProvisionFor checks |
| 163 | // it again under the store lock. failed is the door's own retry sentence, |
| 164 | // answered with a 502 when the store record cannot be read. |
| 165 | func (s *Server) seedStoreRefusal(ctx context.Context, c seedCreate, seedID, failed string) *doorRefusal { |
| 166 | if !c.m.PrivateStorage { |
| 167 | return nil |
| 168 | } |
| 169 | rec, _, err := objstore.GetIn(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.rec.Environment) |
| 170 | if err != nil { |
| 171 | log.Printf("console: seed store check %s/%s/%s: %v", c.acct.Sqid, c.app, c.rec.Environment, err) |
| 172 | return &doorRefusal{http.StatusBadGateway, failed} |
| 173 | } |
| 174 | if !seedOwnsStore(rec, seedID) { |
| 175 | return seedStoreTakenRefusal(c.rec.Environment, c.app, rec) |
| 176 | } |
| 177 | user, err := s.seedEnvironmentUser(ctx, c.acct.Sqid, c.app, c.rec.Environment, seedID) |
| 178 | if err != nil { |
| 179 | log.Printf("console: seed store check %s/%s/%s: who names the environment: %v", c.acct.Sqid, c.app, c.rec.Environment, err) |
| 180 | return &doorRefusal{http.StatusBadGateway, failed} |
| 181 | } |
| 182 | if user != "" { |
| 183 | log.Printf("console: seed store check %s/%s/%s: instance %s already names the environment", c.acct.Sqid, c.app, c.rec.Environment, user) |
| 184 | return seedStoreTakenRefusal(c.rec.Environment, c.app, rec) |
| 185 | } |
| 186 | return nil |
| 187 | } |
| 188 | |
| 189 | // seedGiB is SD-N26's size, rounded up to a tenth so a size over the limit |
| 190 | // never prints as the limit itself. |
| 191 | func seedGiB(n int64) string { |
| 192 | return fmt.Sprintf("%.1f GiB", math.Ceil(float64(n)/(1<<30)*10)/10) |
| 193 | } |
| 194 | |
| 195 | // fromCopy is a --from copy whose record is reserved: everything runFrom |
| 196 | // needs once the request has been answered. It owns the scratch |
| 197 | // reservation (release) from the moment beginFrom hands it over. |
| 198 | type fromCopy struct { |
| 199 | ctx context.Context // the request's values, never its cancellation |
| 200 | c seedCreate |
| 201 | from seedFromReq |
| 202 | source string |
| 203 | id string |
| 204 | plan dbdoor.RestorePlan |
| 205 | client *dbdoor.Client |
| 206 | release func() |
| 207 | } |
| 208 | |
| 209 | // createFrom is --from (spec §4). Every refusal up to the record's |
| 210 | // reservation is answered here, synchronously, after the scratch |
| 211 | // reservation is released. Once the record is reserved the door answers |
| 212 | // 202 with the reserving record and the copy runs on, detached from the |
| 213 | // request (runFrom): a copy can take most of an hour, and the edge's |
| 214 | // backhaul, the CLI's client and idle NATs all cut a response that keeps |
| 215 | // its headers longer than 30 seconds (Plan 5 ruling R2, a deviation from |
| 216 | // spec §4 for Paul). The caller follows the record until its seed is ready |
| 217 | // or aborted, with the reason in Seed.Reason. |
| 218 | func (s *Server) createFrom(w http.ResponseWriter, r *http.Request, c seedCreate, from seedFromReq) { |
| 219 | held := &heldResponse{header: w.Header()} |
| 220 | job := s.beginFrom(held, r, c, from) |
| 221 | if job != nil { |
| 222 | s.seedCopies.Add(1) |
| 223 | go s.runFrom(job) |
| 224 | } |
| 225 | held.flush(w, fmt.Sprintf("could not copy %s. Nothing was created. Try again.", strings.TrimSpace(from.Host))) |
| 226 | } |
| 227 | |
| 228 | // beginFrom is the synchronous half: every check, the plan and its |
| 229 | // admission, the store gates that write nothing, and the record's |
| 230 | // reservation. It returns the copy to run, or nil when it has answered a |
| 231 | // refusal on w. |
| 232 | func (s *Server) beginFrom(w http.ResponseWriter, r *http.Request, c seedCreate, from seedFromReq) *fromCopy { |
| 233 | ctx := r.Context() |
| 234 | acct, app, host := c.acct, c.app, c.rec.Host |
| 235 | source := strings.TrimSpace(from.Host) |
| 236 | failed := fmt.Sprintf("could not copy %s. Nothing was created. Try again.", source) |
| 237 | if ref := seedPlacementRefusal(c); ref != nil { |
| 238 | jsonError(w, ref.Status, ref.Msg) |
| 239 | return nil |
| 240 | } |
| 241 | if s.cfg.SeedData == nil || s.cfg.HibernateBucket == "" { |
| 242 | ref := seedNoDataRefusal() |
| 243 | jsonError(w, ref.Status, ref.Msg) |
| 244 | return nil |
| 245 | } |
| 246 | checks, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 247 | defer cancel() |
| 248 | if ref := s.seedRegionReady(checks, c.rec.Region, false); ref != nil { |
| 249 | jsonError(w, ref.Status, ref.Msg) |
| 250 | return nil |
| 251 | } |
| 252 | src, err := s.resolveDBSource(checks, acct, c.principal, app, source) |
| 253 | if ref := s.fromSourceRefusal(src, source, app, err); ref != nil { |
| 254 | jsonError(w, ref.Status, ref.Msg) |
| 255 | return nil |
| 256 | } |
| 257 | if src.rec.Region != c.rec.Region { |
| 258 | jsonError(w, http.StatusConflict, "a copy stays in its source's region. To run it elsewhere, create it here and then run carlos move.") |
| 259 | return nil |
| 260 | } |
| 261 | rd, err := s.dbReplica(ctx, src.prefix) |
| 262 | if err != nil { |
| 263 | log.Printf("console: from %s: replica client: %v", source, err) |
| 264 | jsonError(w, http.StatusBadGateway, failed) |
| 265 | return nil |
| 266 | } |
| 267 | client := dbdoor.NewClient(replicaDoer(rd)) |
| 268 | planCtx, planCancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 269 | plan, err := planRestore(planCtx, client, from.TXID) |
| 270 | planCancel() |
| 271 | switch { |
| 272 | case from.TXID != 0 && (errors.Is(err, dbdoor.ErrHistoryGone) || errors.Is(err, dbdoor.ErrNothingReplicated)): |
| 273 | jsonError(w, http.StatusConflict, fmt.Sprintf("history before %d is no longer kept", from.TXID)) |
| 274 | return nil |
| 275 | case errors.Is(err, dbdoor.ErrNothingReplicated): |
| 276 | jsonError(w, http.StatusConflict, fmt.Sprintf("%s has not replicated anything yet. Wake it once, then try again.", source)) |
| 277 | return nil |
| 278 | case err != nil: |
| 279 | // A file that vanished under the plan, a header that does not |
| 280 | // describe its file, a bucket read: a retry plans again. |
| 281 | log.Printf("console: from %s: plan: %v", source, err) |
| 282 | jsonError(w, http.StatusBadGateway, failed) |
| 283 | return nil |
| 284 | case from.TXID > plan.TXID: |
| 285 | // The plan clamps an --at past the latest transaction to the |
| 286 | // latest copy. A caller who named a point later than anything the |
| 287 | // source has replicated is told so, and nothing is reserved. |
| 288 | jsonError(w, http.StatusConflict, fmt.Sprintf("%s has replicated up to txid %d, so there is no txid %d to copy yet. Leave out --at to copy the latest.", source, plan.TXID, from.TXID)) |
| 289 | return nil |
| 290 | } |
| 291 | if plan.RestoredBytes > seedMaxBytes { |
| 292 | jsonError(w, http.StatusRequestEntityTooLarge, fmt.Sprintf("a copy of %s would be %s, over the 4 GiB limit.", source, seedGiB(plan.RestoredBytes))) |
| 293 | return nil |
| 294 | } |
| 295 | releaseScratch, ref := s.seedScratch.reserve(acct.Sqid+"/"+app, plan.RestoredBytes+plan.LTXBytes) |
| 296 | if ref != nil { |
| 297 | jsonError(w, ref.Status, ref.Msg) |
| 298 | return nil |
| 299 | } |
| 300 | handedOver := false |
| 301 | defer func() { |
| 302 | if !handedOver { |
| 303 | releaseScratch() |
| 304 | } |
| 305 | }() |
| 306 | id, err := seeds.NewID() |
| 307 | if err != nil { |
| 308 | log.Printf("console: from %s: seed id: %v", source, err) |
| 309 | jsonError(w, http.StatusBadGateway, failed) |
| 310 | return nil |
| 311 | } |
| 312 | if c.m.PrivateStorage { |
| 313 | // The store gates that write nothing run before the reservation, |
| 314 | // so their refusals leave no aborted record holding the host |
| 315 | // (ruling R3). The walk runs them again, after it. |
| 316 | if _, ok := s.instanceStorePreflight(ctx, w, acct, app); !ok { |
| 317 | return nil |
| 318 | } |
| 319 | storeCtx, storeCancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 320 | ref = s.seedStoreRefusal(storeCtx, c, id, failed) |
| 321 | storeCancel() |
| 322 | if ref != nil { |
| 323 | jsonError(w, ref.Status, ref.Msg) |
| 324 | return nil |
| 325 | } |
| 326 | } |
| 327 | |
| 328 | // Reserve the record (spec §4 step 7.1). It is reserved here, after the |
| 329 | // size, scratch and store checks rather than straight after resolving |
| 330 | // the source: a refusal before this point leaves nothing behind, where |
| 331 | // an earlier record would turn each into an aborted record holding the |
| 332 | // host until it is finished (a recorded deviation, Plan 5 M10). |
| 333 | rec := c.rec |
| 334 | rec.Seed = &provision.SeedSource{ID: id, Kind: seeds.KindReplica, From: source, State: seeds.StateReserving, |
| 335 | By: c.principal, At: time.Now().UTC()} |
| 336 | // Not the request's cancellation: a client leaving now must not turn a |
| 337 | // write that landed into an error, which would answer 502 and start no |
| 338 | // copy while the reserving record held the host. |
| 339 | writeCtx, writeCancel := context.WithTimeout(context.WithoutCancel(ctx), instanceStoreTimeout) |
| 340 | defer writeCancel() |
| 341 | if err := provision.CreateRecord(writeCtx, s.cfg.Releases, rec); err != nil { |
| 342 | if errors.Is(err, release.ErrCASConflict) { |
| 343 | jsonError(w, http.StatusConflict, "instance "+host+" already exists") |
| 344 | return nil |
| 345 | } |
| 346 | log.Printf("console: from %s: reserve %s: %v", source, host, err) |
| 347 | jsonError(w, http.StatusBadGateway, failed) |
| 348 | return nil |
| 349 | } |
| 350 | out := recordView(rec) |
| 351 | out.RegionFrom = c.regionFrom |
| 352 | out.RegionName = s.placeOf(rec.Region, s.candsOrHome(writeCtx)) // never fleet-placed: refused above |
| 353 | writeJSON(w, http.StatusAccepted, out) |
| 354 | handedOver = true |
| 355 | return &fromCopy{ctx: context.WithoutCancel(ctx), c: c, from: from, source: source, |
| 356 | id: id, plan: plan, client: client, release: releaseScratch} |
| 357 | } |
| 358 | |
| 359 | // runFrom is the detached half: the seed's store, the restore, the upload |
| 360 | // and the publish. It never depends on the request, which has been |
| 361 | // answered. Every failure moves the record to aborted with the sentence |
| 362 | // the synchronous door would have answered in Seed.Reason, for the box |
| 363 | // and the console's finish to clean up (spec §4 step 7). On every exit it |
| 364 | // removes its scratch and then releases the reservation. |
| 365 | // |
| 366 | // The console has no shutdown hook to cancel a running copy: a restart |
| 367 | // leaves the record reserving, its scratch is cleared at start, and the |
| 368 | // sweep aborts it (Plan 8). |
| 369 | func (s *Server) runFrom(j *fromCopy) { |
| 370 | defer s.seedCopies.Done() |
| 371 | defer j.release() |
| 372 | ctx := j.ctx |
| 373 | c := j.c |
| 374 | acct, app, host, source := c.acct, c.app, c.rec.Host, j.source |
| 375 | // The reason a copy that failed after the answer leaves on its record. |
| 376 | // The aborted record holds the host until the box and the next sweep |
| 377 | // finish it, so a retry before then finds the host taken. |
| 378 | failed := fmt.Sprintf("could not copy %s, so no instance was created. %s is free again within about 5 minutes. Try again then.", source, host) |
| 379 | abort := func(reason, why string) { |
| 380 | bg, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 381 | defer cancel() |
| 382 | if _, err := provision.SetRecordSeed(bg, s.cfg.Releases, acct.Sqid, app, host, provision.SeedAbortBecause(j.id, reason)); err != nil { |
| 383 | log.Printf("console: from %s: abort %s (seed %s) after %s: %v", source, host, j.id, why, err) |
| 384 | } |
| 385 | s.bumpSerial(bg) |
| 386 | } |
| 387 | // Nothing above this copy contains a panic any more: net/http's |
| 388 | // handler recover covered the synchronous door, and a panic here (in |
| 389 | // the restore's decode of replica data, the check, the store walk, the |
| 390 | // upload) would end the whole console. It aborts this seed instead. |
| 391 | // Registered after release and Done, so it runs before them; scratch |
| 392 | // removal, registered later, has already run. |
| 393 | defer func() { |
| 394 | if p := recover(); p != nil { |
| 395 | log.Printf("console: from %s: panic in the copy of %s (seed %s): %v\n%s", source, host, j.id, p, debug.Stack()) |
| 396 | abort(failed, "a panic") |
| 397 | } |
| 398 | }() |
| 399 | |
| 400 | if c.m.PrivateStorage { |
| 401 | // Its refusals are written to a held answer nobody reads but this: |
| 402 | // the sentence becomes the seed's reason. |
| 403 | held := &heldResponse{header: http.Header{}} |
| 404 | storeRec, ok, err := s.instanceStoreProvisionFor(ctx, held, acct, app, c.rec.Environment, c.principal, j.id) |
| 405 | if !ok { |
| 406 | reason := held.message(failed) |
| 407 | switch { |
| 408 | case errors.Is(err, errSeedStoreTaken): |
| 409 | reason = seedStoreTakenRefusal(c.rec.Environment, c.app, storeRec).Msg |
| 410 | case errors.Is(err, errSeedEnvironmentUnread): |
| 411 | reason = failed |
| 412 | } |
| 413 | abort(reason, "store provisioning") |
| 414 | return |
| 415 | } |
| 416 | } |
| 417 | |
| 418 | dir, err := os.MkdirTemp(s.seedScratch.dir(), seedScratchPrefix) |
| 419 | if err != nil { |
| 420 | log.Printf("console: from %s: scratch: %v", source, err) |
| 421 | abort(seedScratchFull().Msg, "scratch") |
| 422 | return |
| 423 | } |
| 424 | defer os.RemoveAll(dir) |
| 425 | // Pinned at the planned TXID, never "latest" again: only that restore |
| 426 | // produces the size that was admitted, however far the source has moved |
| 427 | // on since it was planned. |
| 428 | plan := j.plan |
| 429 | rctx, rcancel := context.WithTimeout(ctx, seedRestoreDeadline) |
| 430 | defer rcancel() |
| 431 | grew := restoreWatch(rctx, rcancel, dir, plan.RestoredBytes, plan.RestoredBytes+plan.LTXBytes) |
| 432 | restored, err := restoreToTarget(rctx, j.client, dir, plan.TXID) |
| 433 | timedOut := errors.Is(rctx.Err(), context.DeadlineExceeded) |
| 434 | rcancel() // stops the watcher |
| 435 | overgrown := grew() |
| 436 | if err == nil && !overgrown { |
| 437 | // The watcher's tick can come after a fast restore has finished. |
| 438 | st, serr := os.Stat(restored.Path) |
| 439 | switch { |
| 440 | case serr != nil: |
| 441 | err = serr |
| 442 | case st.Size() > plan.RestoredBytes: |
| 443 | overgrown = true |
| 444 | } |
| 445 | } |
| 446 | if err != nil || overgrown { |
| 447 | reason := failed |
| 448 | switch { |
| 449 | case overgrown: |
| 450 | reason = fmt.Sprintf("the copy of %s grew past its planned size, so it was stopped and no instance was created.", source) |
| 451 | case timedOut: |
| 452 | reason = fmt.Sprintf("copying %s took longer than 30 minutes, so no instance was created.", source) |
| 453 | case errors.Is(err, dbdoor.ErrCheckFailed): |
| 454 | reason = fmt.Sprintf("the copy of %s failed its integrity check, so no instance was created.", source) |
| 455 | case errors.Is(err, dbdoor.ErrHistoryGone) && plan.Target != 0: |
| 456 | reason = fmt.Sprintf("history before %d is no longer kept", j.from.TXID) |
| 457 | default: |
| 458 | // Including history lost under a latest copy between its plan |
| 459 | // and its restore: a retry plans again. |
| 460 | log.Printf("console: from %s: restore: %v", source, err) |
| 461 | } |
| 462 | abort(reason, "restore") |
| 463 | return |
| 464 | } |
| 465 | |
| 466 | uctx, ucancel := context.WithTimeout(ctx, seedUploadDeadline) |
| 467 | sha, size, err := uploadSnapshot(uctx, s.cfg.SeedData, restored.Path, seeds.SnapshotKey(host, j.id)) |
| 468 | ucancel() |
| 469 | if err != nil { |
| 470 | log.Printf("console: from %s: upload %s: %v", source, host, err) |
| 471 | abort(failed, "upload") |
| 472 | return |
| 473 | } |
| 474 | // The create door's re-check after its write (apiinstances.go), made |
| 475 | // here before the publish: a reserving record mints nothing, so the |
| 476 | // publish is the point of no return. |
| 477 | if c.siblingLabel != "" { |
| 478 | if live, trashed, err := s.store.AppNameState(acct.ID, c.siblingLabel); err == nil && (live || trashed) { |
| 479 | abort(seedSiblingApexReason(host, c.siblingLabel), "a sibling app took the label") |
| 480 | return |
| 481 | } |
| 482 | } |
| 483 | pctx, pcancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| 484 | defer pcancel() |
| 485 | if _, err := provision.SetRecordSeed(pctx, s.cfg.Releases, acct.Sqid, app, host, |
| 486 | provision.SeedPublishReplica(j.id, strconv.FormatUint(restored.TXID, 10), sha, size)); err != nil { |
| 487 | log.Printf("console: from %s: publish %s: %v", source, host, err) |
| 488 | abort(failed, "publish") |
| 489 | return |
| 490 | } |
| 491 | s.auditActivity(acct, app, c.principal, Activity{ |
| 492 | Action: "instance-seed", TargetType: "instance", Target: host, Message: c.message, |
| 493 | Detail: fmt.Sprintf("seeded %s from %s at txid %d", host, source, restored.TXID), |
| 494 | }) |
| 495 | s.bumpSerial(pctx) |
| 496 | } |
| 497 | |
| 498 | // fromSourceRefusal maps resolveDBSource's answer to --from's sentences. |
| 499 | func (s *Server) fromSourceRefusal(src dbSource, source, app string, err error) *doorRefusal { |
| 500 | switch { |
| 501 | case err == nil: |
| 502 | return nil |
| 503 | case errors.Is(err, os.ErrNotExist), errors.Is(err, ErrBadHost): |
| 504 | return &doorRefusal{http.StatusConflict, fmt.Sprintf("no instance %s in %s. A copy can only come from an instance of the same app in the same account.", source, app)} |
| 505 | case errors.Is(err, errSourceClass): |
| 506 | return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s is not an instance the platform keeps a replica of, so it cannot be copied.", source)} |
| 507 | case errors.Is(err, ErrPooled): |
| 508 | return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s runs in the pool %s, so the platform keeps no replica of it to copy.", source, src.rec.Pool)} |
| 509 | case errors.Is(err, ErrResidency): |
| 510 | if fleet, ok := registry.FleetPlacement(src.rec.Region); ok { |
| 511 | return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s runs on your fleet %s, so the platform keeps no replica of it to copy.", source, fleet)} |
| 512 | } |
| 513 | return &doorRefusal{http.StatusConflict, s.dbResidencyRefusal(src.rec)} |
| 514 | case errors.Is(err, errSourceClaim): |
| 515 | // Unreachable while seedRegionReady refuses a claim console first. |
| 516 | return seedClaimConsoleRefusal() |
| 517 | } |
| 518 | log.Printf("console: from %s: resolve: %v", source, err) |
| 519 | return &doorRefusal{http.StatusBadGateway, fmt.Sprintf("could not copy %s. Nothing was created. Try again.", source)} |
| 520 | } |
| 521 | |
| 522 | // uploadSnapshot hashes path and writes it to key with one create-only |
| 523 | // PutFrom (spec §4 step 7.2: seekable, under S3's 5 GB single-PUT limit). |
| 524 | func uploadSnapshot(ctx context.Context, data release.Store, path, key string) (sha string, size int64, err error) { |
| 525 | f, err := os.Open(path) |
| 526 | if err != nil { |
| 527 | return "", 0, err |
| 528 | } |
| 529 | defer f.Close() |
| 530 | h := sha256.New() |
| 531 | if size, err = io.Copy(h, f); err != nil { |
| 532 | return "", 0, err |
| 533 | } |
| 534 | if _, err := f.Seek(0, io.SeekStart); err != nil { |
| 535 | return "", 0, err |
| 536 | } |
| 537 | if _, err := release.PutObjectFrom(ctx, data, key, f, size, ""); err != nil { |
| 538 | return "", 0, err |
| 539 | } |
| 540 | return hex.EncodeToString(h.Sum(nil)), size, nil |
| 541 | } |
| 542 | |
| 543 | // restoreWatch is the size watcher (spec §4 step 5). Every |
| 544 | // restoreWatchEvery it measures the restore under dir and cancels it when |
| 545 | // the database (litestream's replica.db.tmp while it decodes, replica.db |
| 546 | // after its rename) grows past dbLimit, or everything the restore holds |
| 547 | // (the database and its downloaded LTX files) grows past totalLimit, the |
| 548 | // reservation. grew reports whether it fired. |
| 549 | func restoreWatch(ctx context.Context, cancel context.CancelFunc, dir string, dbLimit, totalLimit int64) (grew func() bool) { |
| 550 | var fired atomic.Bool |
| 551 | every := restoreWatchEvery |
| 552 | go func() { |
| 553 | t := time.NewTicker(every) |
| 554 | defer t.Stop() |
| 555 | for { |
| 556 | select { |
| 557 | case <-ctx.Done(): |
| 558 | return |
| 559 | case <-t.C: |
| 560 | } |
| 561 | if db, total := restoreScratchUse(dir); db > dbLimit || total > totalLimit { |
| 562 | fired.Store(true) |
| 563 | cancel() |
| 564 | return |
| 565 | } |
| 566 | } |
| 567 | }() |
| 568 | return fired.Load |
| 569 | } |
| 570 | |
| 571 | // restoreScratchUse measures a restore under dir: the largest database file |
| 572 | // (replica.db or replica.db.tmp, dbdoor's names) and every regular file |
| 573 | // together. SQLite's shared-memory index, which the read-only check opens |
| 574 | // beside the copy, is not the restore's and is not counted. |
| 575 | func restoreScratchUse(dir string) (db, total int64) { |
| 576 | _ = filepath.WalkDir(dir, func(_ string, d fs.DirEntry, err error) error { |
| 577 | if err != nil || !d.Type().IsRegular() || strings.HasSuffix(d.Name(), "-shm") { |
| 578 | return nil // a file the restore removed between listing and stat is not there |
| 579 | } |
| 580 | info, err := d.Info() |
| 581 | if err != nil { |
| 582 | return nil |
| 583 | } |
| 584 | total += info.Size() |
| 585 | if n := d.Name(); n == "replica.db" || n == "replica.db.tmp" { |
| 586 | db = max(db, info.Size()) |
| 587 | } |
| 588 | return nil |
| 589 | }) |
| 590 | return db, total |
| 591 | } |
| 592 | |
| 593 | // heldResponse holds a door's answer until its cleanup has run. |
| 594 | type heldResponse struct { |
| 595 | header http.Header |
| 596 | status int |
| 597 | body bytes.Buffer |
| 598 | } |
| 599 | |
| 600 | func (h *heldResponse) Header() http.Header { return h.header } |
| 601 | |
| 602 | func (h *heldResponse) WriteHeader(code int) { |
| 603 | if h.status == 0 { |
| 604 | h.status = code |
| 605 | } |
| 606 | } |
| 607 | |
| 608 | func (h *heldResponse) Write(b []byte) (int, error) { |
| 609 | h.WriteHeader(http.StatusOK) |
| 610 | return h.body.Write(b) |
| 611 | } |
| 612 | |
| 613 | // flush writes what was held. A door that held nothing answers failed |
| 614 | // with a 502, never an empty 200. |
| 615 | func (h *heldResponse) flush(w http.ResponseWriter, failed string) { |
| 616 | if h.status == 0 { |
| 617 | jsonError(w, http.StatusBadGateway, failed) |
| 618 | return |
| 619 | } |
| 620 | w.WriteHeader(h.status) |
| 621 | _, _ = w.Write(h.body.Bytes()) |
| 622 | } |
| 623 | |
| 624 | // message is the held refusal's sentence, or fallback when nothing usable |
| 625 | // was held. |
| 626 | func (h *heldResponse) message(fallback string) string { |
| 627 | var out struct { |
| 628 | Error string `json:"error"` |
| 629 | } |
| 630 | if err := json.Unmarshal(h.body.Bytes(), &out); err != nil || out.Error == "" { |
| 631 | return fallback |
| 632 | } |
| 633 | return out.Error |
| 634 | } |
| 635 | |
amadan