amadan
console: --from answers 202 and copies detached, failures record their reason (Plan 5 Task 5 review)
Once the record is reserved the door answers 202 with the reserving record. The store walk, restore, upload and publish then run detached from the request, which a backhauled edge, the CLI's client or an idle NAT would cut after 30 seconds. Every failure aborts the seed through the new provision.SeedAbortBecause, with the sentence the synchronous door would have answered as Seed.Reason. Scratch and the reservation are released on every exit. The store plane's write-free gates now run before the reservation (instanceStorePreflight), so their refusals leave no aborted record. The read-deadline extension is gone, and a door that held no answer replies 502, never an empty 200. SD-N26's size rounds up.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
6 files changed,
+470
−268
internal/console/apiinstanceseed.go+149 −104internal/console/apiinstanceseed_test.go+253 −141internal/console/instanceprivatestore.go+29 −19internal/console/server.go+5 −2internal/provision/seed.go+11 −2internal/provision/seed_test.go+23 −0
diff --git a/internal/console/apiinstanceseed.go b/internal/console/apiinstanceseed.go| index a01e92d1d..35cb5f974 100644 |
| --- a/internal/console/apiinstanceseed.go |
| +++ b/internal/console/apiinstanceseed.go |
| @@ -1,7 +1,8 @@ |
| // The --from seed door (seeded-instances spec §4): `carlos instances create |
| // --from <host>` creates an instance whose database starts as a copy of a |
| // sibling's replica. handleAPIInstanceCreate runs every ordinary check, |
| -// then hands over here. |
| +// then hands over here. The door answers 202 once the record is reserved, |
| +// and the copy finishes on its own (createFrom). |
| package console |
| @@ -10,11 +11,13 @@ import ( |
| "context" |
| "crypto/sha256" |
| "encoding/hex" |
| + "encoding/json" |
| "errors" |
| "fmt" |
| "io" |
| "io/fs" |
| "log" |
| + "math" |
| "net/http" |
| "os" |
| "path/filepath" |
| @@ -49,10 +52,6 @@ var ( |
| // together stay well under that. |
| const seedUploadDeadline = 20 * time.Minute |
| -// seedReadSlack is how far past a step's own deadline the connection's read |
| -// deadline is set, so the step's deadline is always the one that fires. |
| -const seedReadSlack = 5 * time.Minute |
| - |
| // seedFromReq is the create body's "from" (spec §4). TXID 0 is the latest |
| // copy; the CLI refuses a non-positive --at. |
| type seedFromReq struct { |
| @@ -120,65 +119,83 @@ func (s *Server) seedStoreRefusal(ctx context.Context, c seedCreate, seedID, fai |
| return nil |
| } |
| -func seedGiB(n int64) string { return fmt.Sprintf("%.1f GiB", float64(n)/(1<<30)) } |
| +// seedGiB is SD-N26's size, rounded up to a tenth so a size over the limit |
| +// never prints as the limit itself. |
| +func seedGiB(n int64) string { |
| + return fmt.Sprintf("%.1f GiB", math.Ceil(float64(n)/(1<<30)*10)/10) |
| +} |
| -// createFrom is --from (spec §4). It answers once, after its scratch is |
| -// removed and its reservation released, so a caller that retries at once |
| -// never meets its own leftovers. |
| -// |
| -// The request runs as long as the copy does. The connection's read |
| -// deadline is set past each step's own deadline (seedReadSlack), so the |
| -// server's ReadTimeout is never the one that ends a copy. Go 1.26's |
| -// net/http already clears that deadline once the body is read; this keeps |
| -// the door's deadlines in charge whatever the server does. A client that |
| -// goes away still cancels the request's context. |
| +// fromCopy is a --from copy whose record is reserved: everything runFrom |
| +// needs once the request has been answered. It owns the scratch |
| +// reservation (release) from the moment beginFrom hands it over. |
| +type fromCopy struct { |
| + ctx context.Context // the request's values, never its cancellation |
| + c seedCreate |
| + from seedFromReq |
| + source string |
| + failed string // SD-N39 for this source |
| + id string |
| + plan dbdoor.RestorePlan |
| + client *dbdoor.Client |
| + release func() |
| +} |
| + |
| +// createFrom is --from (spec §4). Every refusal up to the record's |
| +// reservation is answered here, synchronously, after the scratch |
| +// reservation is released. Once the record is reserved the door answers |
| +// 202 with the reserving record and the copy runs on, detached from the |
| +// request (runFrom): a copy can take most of an hour, and the edge's |
| +// backhaul, the CLI's client and idle NATs all cut a response that keeps |
| +// its headers longer than 30 seconds (Plan 5 ruling R2, a deviation from |
| +// spec §4 for Paul). The caller follows the record until its seed is ready |
| +// or aborted, with the reason in Seed.Reason. |
| func (s *Server) createFrom(w http.ResponseWriter, r *http.Request, c seedCreate, from seedFromReq) { |
| - rc := http.NewResponseController(w) |
| - extend := func(d time.Duration) { |
| - if err := rc.SetReadDeadline(time.Now().Add(d + seedReadSlack)); err != nil && !errors.Is(err, http.ErrNotSupported) { |
| - log.Printf("console: from %s: extend the read deadline: %v", c.rec.Host, err) |
| - } |
| - } |
| - extend(seedRestoreDeadline) |
| - defer func() { _ = rc.SetReadDeadline(time.Time{}) }() |
| held := &heldResponse{header: w.Header()} |
| - s.copyFrom(held, r, c, from, func() { extend(seedUploadDeadline) }) |
| - held.flush(w) |
| + job := s.beginFrom(held, r, c, from) |
| + if job != nil { |
| + s.seedCopies.Add(1) |
| + go s.runFrom(job) |
| + } |
| + held.flush(w, fmt.Sprintf("could not copy %s. Nothing was created. Try again.", strings.TrimSpace(from.Host))) |
| } |
| -func (s *Server) copyFrom(w http.ResponseWriter, r *http.Request, c seedCreate, from seedFromReq, beforeUpload func()) { |
| +// beginFrom is the synchronous half: every check, the plan and its |
| +// admission, the store gates that write nothing, and the record's |
| +// reservation. It returns the copy to run, or nil when it has answered a |
| +// refusal on w. |
| +func (s *Server) beginFrom(w http.ResponseWriter, r *http.Request, c seedCreate, from seedFromReq) *fromCopy { |
| ctx := r.Context() |
| acct, app, host := c.acct, c.app, c.rec.Host |
| source := strings.TrimSpace(from.Host) |
| failed := fmt.Sprintf("could not copy %s. Nothing was created. Try again.", source) |
| if ref := seedPlacementRefusal(c); ref != nil { |
| jsonError(w, ref.Status, ref.Msg) |
| - return |
| + return nil |
| } |
| if s.cfg.SeedData == nil || s.cfg.HibernateBucket == "" { |
| jsonError(w, http.StatusServiceUnavailable, "this console has no data bucket set up, so it cannot create a seeded instance.") |
| - return |
| + return nil |
| } |
| checks, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| defer cancel() |
| if ref := s.seedRegionReady(checks, c.rec.Region, false); ref != nil { |
| jsonError(w, ref.Status, ref.Msg) |
| - return |
| + return nil |
| } |
| src, err := s.resolveDBSource(checks, acct, c.principal, app, source) |
| if ref := s.fromSourceRefusal(src, source, app, err); ref != nil { |
| jsonError(w, ref.Status, ref.Msg) |
| - return |
| + return nil |
| } |
| if src.rec.Region != c.rec.Region { |
| jsonError(w, http.StatusConflict, "a copy stays in its source's region. To run it elsewhere, create it here and then run carlos move.") |
| - return |
| + return nil |
| } |
| rd, err := s.dbReplica(ctx, src.prefix) |
| if err != nil { |
| log.Printf("console: from %s: replica client: %v", source, err) |
| jsonError(w, http.StatusBadGateway, failed) |
| - return |
| + return nil |
| } |
| client := dbdoor.NewClient(replicaDoer(rd)) |
| planCtx, planCancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| @@ -187,39 +204,52 @@ func (s *Server) copyFrom(w http.ResponseWriter, r *http.Request, c seedCreate, |
| switch { |
| case from.TXID != 0 && (errors.Is(err, dbdoor.ErrHistoryGone) || errors.Is(err, dbdoor.ErrNothingReplicated)): |
| jsonError(w, http.StatusConflict, fmt.Sprintf("history before %d is no longer kept", from.TXID)) |
| - return |
| + return nil |
| case errors.Is(err, dbdoor.ErrNothingReplicated): |
| jsonError(w, http.StatusConflict, fmt.Sprintf("%s has not replicated anything yet. Wake it once, then try again.", source)) |
| - return |
| + return nil |
| case err != nil: |
| // A file that vanished under the plan, a header that does not |
| // describe its file, a bucket read: a retry plans again. |
| log.Printf("console: from %s: plan: %v", source, err) |
| jsonError(w, http.StatusBadGateway, failed) |
| - return |
| + return nil |
| } |
| if plan.RestoredBytes > seedMaxBytes { |
| jsonError(w, http.StatusRequestEntityTooLarge, fmt.Sprintf("a copy of %s would be %s, over the 4 GiB limit.", source, seedGiB(plan.RestoredBytes))) |
| - return |
| + return nil |
| } |
| releaseScratch, ref := s.seedScratch.reserve(acct.Sqid+"/"+app, plan.RestoredBytes+plan.LTXBytes) |
| if ref != nil { |
| jsonError(w, ref.Status, ref.Msg) |
| - return |
| + return nil |
| } |
| - defer releaseScratch() |
| + handedOver := false |
| + defer func() { |
| + if !handedOver { |
| + releaseScratch() |
| + } |
| + }() |
| id, err := seeds.NewID() |
| if err != nil { |
| log.Printf("console: from %s: seed id: %v", source, err) |
| jsonError(w, http.StatusBadGateway, failed) |
| - return |
| + return nil |
| } |
| - storeCtx, storeCancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| - ref = s.seedStoreRefusal(storeCtx, c, id, failed) |
| - storeCancel() |
| - if ref != nil { |
| - jsonError(w, ref.Status, ref.Msg) |
| - return |
| + if c.m.PrivateStorage { |
| + // The store gates that write nothing run before the reservation, |
| + // so their refusals leave no aborted record holding the host |
| + // (ruling R3). The walk runs them again, after it. |
| + if _, ok := s.instanceStorePreflight(ctx, w, acct, app); !ok { |
| + return nil |
| + } |
| + storeCtx, storeCancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| + ref = s.seedStoreRefusal(storeCtx, c, id, failed) |
| + storeCancel() |
| + if ref != nil { |
| + jsonError(w, ref.Status, ref.Msg) |
| + return nil |
| + } |
| } |
| // Reserve the record (spec §4 step 7.1). It is reserved here, after the |
| @@ -231,61 +261,80 @@ func (s *Server) copyFrom(w http.ResponseWriter, r *http.Request, c seedCreate, |
| rec.Seed = &provision.SeedSource{ID: id, Kind: seeds.KindReplica, From: source, State: seeds.StateReserving, |
| By: c.principal, At: time.Now().UTC()} |
| writeCtx, writeCancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| - err = provision.CreateRecord(writeCtx, s.cfg.Releases, rec) |
| - writeCancel() |
| - if err != nil { |
| + defer writeCancel() |
| + if err := provision.CreateRecord(writeCtx, s.cfg.Releases, rec); err != nil { |
| if errors.Is(err, release.ErrCASConflict) { |
| jsonError(w, http.StatusConflict, "instance "+host+" already exists") |
| - return |
| + return nil |
| } |
| log.Printf("console: from %s: reserve %s: %v", source, host, err) |
| jsonError(w, http.StatusBadGateway, failed) |
| - return |
| + return nil |
| } |
| - // From here every failure moves the record to aborted, for the box and |
| - // the console's finish to clean up (spec §4 step 7), under a context the |
| - // client's departure cannot cancel. A store this seed provisioned is |
| - // left to that finish (Plan 8). |
| - abort := func(why string) { |
| - bg, cancel := context.WithTimeout(context.WithoutCancel(ctx), instanceStoreTimeout) |
| + out := recordView(rec) |
| + out.RegionFrom = c.regionFrom |
| + out.RegionName = s.placeOf(rec.Region, s.candsOrHome(writeCtx)) // never fleet-placed: refused above |
| + writeJSON(w, http.StatusAccepted, out) |
| + handedOver = true |
| + return &fromCopy{ctx: context.WithoutCancel(ctx), c: c, from: from, source: source, failed: failed, |
| + id: id, plan: plan, client: client, release: releaseScratch} |
| +} |
| + |
| +// runFrom is the detached half: the seed's store, the restore, the upload |
| +// and the publish. It never depends on the request, which has been |
| +// answered. Every failure moves the record to aborted with the sentence |
| +// the synchronous door would have answered in Seed.Reason, for the box |
| +// and the console's finish to clean up (spec §4 step 7). On every exit it |
| +// removes its scratch and then releases the reservation. |
| +// |
| +// The console has no shutdown hook to cancel a running copy: a restart |
| +// leaves the record reserving, its scratch is cleared at start, and the |
| +// sweep aborts it (Plan 8). |
| +func (s *Server) runFrom(j *fromCopy) { |
| + defer s.seedCopies.Done() |
| + defer j.release() |
| + ctx := j.ctx |
| + c := j.c |
| + acct, app, host, source := c.acct, c.app, c.rec.Host, j.source |
| + abort := func(reason, why string) { |
| + bg, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| defer cancel() |
| - if _, err := provision.SetRecordSeed(bg, s.cfg.Releases, acct.Sqid, app, host, provision.SeedAbort(id)); err != nil { |
| - log.Printf("console: from %s: abort %s (seed %s) after %s: %v", source, host, id, why, err) |
| + if _, err := provision.SetRecordSeed(bg, s.cfg.Releases, acct.Sqid, app, host, provision.SeedAbortBecause(j.id, reason)); err != nil { |
| + log.Printf("console: from %s: abort %s (seed %s) after %s: %v", source, host, j.id, why, err) |
| } |
| s.bumpSerial(bg) |
| } |
| - privateBucket := "" |
| if c.m.PrivateStorage { |
| - // The request context: the walk takes its own storeGrantTimeout. |
| - storeRec, ok, err := s.instanceStoreProvisionFor(ctx, w, acct, app, rec.Environment, c.principal, id) |
| + // Its refusals are written to a held answer nobody reads but this: |
| + // the sentence becomes the seed's reason. |
| + held := &heldResponse{header: http.Header{}} |
| + storeRec, ok, err := s.instanceStoreProvisionFor(ctx, held, acct, app, c.rec.Environment, c.principal, j.id) |
| if !ok { |
| - abort("store provisioning") |
| + reason := held.message(j.failed) |
| if errors.Is(err, errSeedStoreTaken) { |
| - ref := seedStoreTakenRefusal(rec.Environment, storeBucketName(storeRec)) |
| - jsonError(w, ref.Status, ref.Msg) |
| + reason = seedStoreTakenRefusal(c.rec.Environment, storeBucketName(storeRec)).Msg |
| } |
| + abort(reason, "store provisioning") |
| return |
| } |
| - privateBucket = storeBucketName(storeRec) |
| } |
| dir, err := os.MkdirTemp(s.seedScratch.dir(), seedScratchPrefix) |
| if err != nil { |
| log.Printf("console: from %s: scratch: %v", source, err) |
| - abort("scratch") |
| - ref := seedScratchFull() |
| - jsonError(w, ref.Status, ref.Msg) |
| + abort(seedScratchFull().Msg, "scratch") |
| return |
| } |
| defer os.RemoveAll(dir) |
| // Pinned at the planned TXID, never "latest" again: only that restore |
| // produces the size that was admitted, however far the source has moved |
| // on since it was planned. |
| + plan := j.plan |
| rctx, rcancel := context.WithTimeout(ctx, seedRestoreDeadline) |
| defer rcancel() |
| grew := restoreWatch(rctx, rcancel, dir, plan.RestoredBytes, plan.RestoredBytes+plan.LTXBytes) |
| - restored, err := restoreToTarget(rctx, client, dir, plan.TXID) |
| + restored, err := restoreToTarget(rctx, j.client, dir, plan.TXID) |
| timedOut := errors.Is(rctx.Err(), context.DeadlineExceeded) |
| rcancel() // stops the watcher |
| overgrown := grew() |
| @@ -300,37 +349,31 @@ func (s *Server) copyFrom(w http.ResponseWriter, r *http.Request, c seedCreate, |
| } |
| } |
| if err != nil || overgrown { |
| - abort("restore") |
| + reason := j.failed |
| switch { |
| case overgrown: |
| - jsonError(w, http.StatusRequestEntityTooLarge, fmt.Sprintf("the copy of %s grew past its planned size, so it was stopped and no instance was created.", source)) |
| - case ctx.Err() != nil: |
| - // The client went away: nobody to answer. |
| + reason = fmt.Sprintf("the copy of %s grew past its planned size, so it was stopped and no instance was created.", source) |
| case timedOut: |
| - jsonError(w, http.StatusGatewayTimeout, fmt.Sprintf("copying %s took longer than 30 minutes, so no instance was created.", source)) |
| + reason = fmt.Sprintf("copying %s took longer than 30 minutes, so no instance was created.", source) |
| case errors.Is(err, dbdoor.ErrCheckFailed): |
| - jsonError(w, http.StatusUnprocessableEntity, fmt.Sprintf("the copy of %s failed its integrity check, so no instance was created.", source)) |
| + reason = fmt.Sprintf("the copy of %s failed its integrity check, so no instance was created.", source) |
| case errors.Is(err, dbdoor.ErrHistoryGone) && plan.Target != 0: |
| - jsonError(w, http.StatusConflict, fmt.Sprintf("history before %d is no longer kept", from.TXID)) |
| + reason = fmt.Sprintf("history before %d is no longer kept", j.from.TXID) |
| default: |
| // Including history lost under a latest copy between its plan |
| // and its restore: a retry plans again. |
| log.Printf("console: from %s: restore: %v", source, err) |
| - jsonError(w, http.StatusBadGateway, failed) |
| } |
| + abort(reason, "restore") |
| return |
| } |
| - beforeUpload() |
| uctx, ucancel := context.WithTimeout(ctx, seedUploadDeadline) |
| - sha, size, err := uploadSnapshot(uctx, s.cfg.SeedData, restored.Path, seeds.SnapshotKey(host, id)) |
| + sha, size, err := uploadSnapshot(uctx, s.cfg.SeedData, restored.Path, seeds.SnapshotKey(host, j.id)) |
| ucancel() |
| if err != nil { |
| - abort("upload") |
| - if ctx.Err() == nil { |
| - log.Printf("console: from %s: upload %s: %v", source, host, err) |
| - jsonError(w, http.StatusBadGateway, failed) |
| - } |
| + log.Printf("console: from %s: upload %s: %v", source, host, err) |
| + abort(j.failed, "upload") |
| return |
| } |
| // The create door's re-check after its write (apiinstances.go), made |
| @@ -338,22 +381,16 @@ func (s *Server) copyFrom(w http.ResponseWriter, r *http.Request, c seedCreate, |
| // publish is the point of no return. |
| if c.siblingLabel != "" { |
| if live, trashed, err := s.store.AppNameState(acct.ID, c.siblingLabel); err == nil && (live || trashed) { |
| - abort("a sibling app took the label") |
| - jsonError(w, http.StatusConflict, |
| - host+" is app "+c.siblingLabel+"'s platform apex — it was created while this instance was being written; the instance was not kept") |
| + abort(host+" is app "+c.siblingLabel+"'s platform apex — it was created while this instance was being written; the instance was not kept", "a sibling app took the label") |
| return |
| } |
| } |
| pctx, pcancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| defer pcancel() |
| - published, err := provision.SetRecordSeed(pctx, s.cfg.Releases, acct.Sqid, app, host, |
| - provision.SeedPublishReplica(id, strconv.FormatUint(restored.TXID, 10), sha, size)) |
| - if err != nil { |
| - abort("publish") |
| - if ctx.Err() == nil { |
| - log.Printf("console: from %s: publish %s: %v", source, host, err) |
| - jsonError(w, http.StatusBadGateway, failed) |
| - } |
| + if _, err := provision.SetRecordSeed(pctx, s.cfg.Releases, acct.Sqid, app, host, |
| + provision.SeedPublishReplica(j.id, strconv.FormatUint(restored.TXID, 10), sha, size)); err != nil { |
| + log.Printf("console: from %s: publish %s: %v", source, host, err) |
| + abort(j.failed, "publish") |
| return |
| } |
| s.auditActivity(acct, app, c.principal, Activity{ |
| @@ -361,11 +398,6 @@ func (s *Server) copyFrom(w http.ResponseWriter, r *http.Request, c seedCreate, |
| Detail: fmt.Sprintf("seeded %s from %s at txid %d", host, source, restored.TXID), |
| }) |
| s.bumpSerial(pctx) |
| - out := recordView(*published) |
| - out.PrivateStoreBucket = privateBucket |
| - out.RegionFrom = c.regionFrom |
| - out.RegionName = s.placeOf(published.Region, s.candsOrHome(pctx)) // never fleet-placed: refused above |
| - writeJSON(w, http.StatusCreated, out) |
| } |
| // fromSourceRefusal maps resolveDBSource's answer to --from's sentences. |
| @@ -483,12 +515,25 @@ func (h *heldResponse) Write(b []byte) (int, error) { |
| return h.body.Write(b) |
| } |
| -// flush writes what was held; nothing when the door answered nothing (the |
| -// client went away). |
| -func (h *heldResponse) flush(w http.ResponseWriter) { |
| +// flush writes what was held. A door that held nothing answers failed |
| +// with a 502, never an empty 200. |
| +func (h *heldResponse) flush(w http.ResponseWriter, failed string) { |
| if h.status == 0 { |
| + jsonError(w, http.StatusBadGateway, failed) |
| return |
| } |
| w.WriteHeader(h.status) |
| _, _ = w.Write(h.body.Bytes()) |
| } |
| + |
| +// message is the held refusal's sentence, or fallback when nothing usable |
| +// was held. |
| +func (h *heldResponse) message(fallback string) string { |
| + var out struct { |
| + Error string `json:"error"` |
| + } |
| + if err := json.Unmarshal(h.body.Bytes(), &out); err != nil || out.Error == "" { |
| + return fallback |
| + } |
| + return out.Error |
| +} |
diff --git a/internal/console/apiinstanceseed_test.go b/internal/console/apiinstanceseed_test.go| index 18a3a6a6d..b4ffee8ce 100644 |
| --- a/internal/console/apiinstanceseed_test.go |
| +++ b/internal/console/apiinstanceseed_test.go |
| @@ -13,6 +13,7 @@ import ( |
| "path/filepath" |
| "strconv" |
| "strings" |
| + "sync/atomic" |
| "testing" |
| "time" |
| @@ -33,6 +34,8 @@ const ( |
| fromSource = "main.shop.example.com" |
| fromHost = "canary.shop.example.com" |
| fromBody = `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com"}}` |
| + fromFailed = "could not copy main.shop.example.com. Nothing was created. Try again." |
| + fromGrew = "the copy of main.shop.example.com grew past its planned size, so it was stopped and no instance was created." |
| ) |
| // fromFixture is a console ready for --from: the app shop (no pool, its own |
| @@ -78,6 +81,9 @@ func newFromFixture(t *testing.T, txns int) fromFixture { |
| f.addSource(t, provision.Record{Host: fromSource, Slug: "main"}, txns) |
| publishSeedsFloorFx(t, fx, "v0.27.0") |
| seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false) |
| + // A copy still running when the test ends would write into removed |
| + // directories: every test waits for its copies. |
| + t.Cleanup(fx.srv.seedCopies.Wait) |
| return f |
| } |
| @@ -94,6 +100,7 @@ func (f fromFixture) addSource(t *testing.T, rec provision.Record, txns int) { |
| } |
| } |
| +// create posts a create and returns what the door answered. |
| func (f fromFixture) create(t *testing.T, body string) (int, map[string]any) { |
| t.Helper() |
| return f.createAs(t, f.token, body) |
| @@ -107,17 +114,46 @@ func (f fromFixture) createAs(t *testing.T, token, body string) (int, map[string |
| return code, out |
| } |
| +// copy posts a --from create, requires its 202 with a reserving record, |
| +// waits for the detached copy to finish, and returns the record it left. |
| +func (f fromFixture) copy(t *testing.T, body string) *provision.Record { |
| + t.Helper() |
| + return f.copyAs(t, f.token, body) |
| +} |
| + |
| +func (f fromFixture) copyAs(t *testing.T, token, body string) *provision.Record { |
| + t.Helper() |
| + code, out := f.createAs(t, token, body) |
| + seed, _ := out["seed"].(map[string]any) |
| + if code != http.StatusAccepted || seed == nil || seed["state"] != seeds.StateReserving || seed["kind"] != seeds.KindReplica { |
| + t.Fatalf("%s = %d %v, want 202 with a reserving record", body, code, out) |
| + } |
| + f.fx.srv.seedCopies.Wait() |
| + host, _ := out["host"].(string) |
| + rec, err := f.record(t, host) |
| + if err != nil || rec.Seed == nil || rec.Seed.ID != seed["id"] { |
| + t.Fatalf("record %s after its copy = %+v, %v", host, rec, err) |
| + } |
| + return rec |
| +} |
| + |
| func (f fromFixture) record(t *testing.T, host string) (*provision.Record, error) { |
| t.Helper() |
| return provision.GetRecord(context.Background(), f.fx.rel, "bab", fromApp, host) |
| } |
| -// requireAborted: the record exists with this door's seed aborted. |
| -func (f fromFixture) requireAborted(t *testing.T, host string) { |
| +// requireAborted: the record's replica seed is aborted, saying why. |
| +func requireAborted(t *testing.T, rec *provision.Record, reason string) { |
| t.Helper() |
| - rec, err := f.record(t, host) |
| - if err != nil || rec.Seed == nil || rec.Seed.State != seeds.StateAborted || rec.Seed.Kind != seeds.KindReplica { |
| - t.Fatalf("record %s = %+v, %v; want an aborted replica seed", host, rec, err) |
| + if rec.Seed == nil || rec.Seed.State != seeds.StateAborted || rec.Seed.Kind != seeds.KindReplica || rec.Seed.Reason != reason { |
| + t.Fatalf("record %s seed = %+v; want aborted with reason %q", rec.Host, rec.Seed, reason) |
| + } |
| +} |
| + |
| +func requireReady(t *testing.T, rec *provision.Record) { |
| + t.Helper() |
| + if rec.Seed == nil || rec.Seed.State != seeds.StateReady || rec.Seed.Reason != "" { |
| + t.Fatalf("record %s seed = %+v; want ready", rec.Host, rec.Seed) |
| } |
| } |
| @@ -142,6 +178,13 @@ func (f fromFixture) requireClean(t *testing.T) { |
| rel() |
| } |
| +func (f fromFixture) requireNothingUploaded(t *testing.T) { |
| + t.Helper() |
| + if keys, err := f.data.List(context.Background(), seeds.Prefix); err != nil || len(keys) != 0 { |
| + t.Fatalf("seeds/ = %v, %v; want nothing uploaded", keys, err) |
| + } |
| +} |
| + |
| func snapshotRows(t *testing.T, raw []byte) int { |
| t.Helper() |
| path := filepath.Join(t.TempDir(), "snapshot.db") |
| @@ -162,15 +205,11 @@ func snapshotRows(t *testing.T, raw []byte) int { |
| func TestFromCreatesAReadySeededInstance(t *testing.T) { |
| f := newFromFixture(t, 30) |
| - code, out := f.create(t, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com"},"message":"a copy to try the upgrade on"}`) |
| - if code != http.StatusCreated { |
| - t.Fatalf("create = %d %v", code, out) |
| - } |
| - rec, err := f.record(t, fromHost) |
| - if err != nil || rec.Seed == nil || rec.Seed.State != seeds.StateReady || rec.Seed.Kind != seeds.KindReplica || |
| - rec.Seed.From != fromSource || rec.Seed.TXID == "" || rec.Seed.Bytes == 0 || len(rec.Seed.SHA256) != 64 || |
| - rec.Seed.By != "alice@keymail.dev" || !seeds.ValidID(rec.Seed.ID) { |
| - t.Fatalf("record seed = %+v, %v", rec.Seed, err) |
| + rec := f.copy(t, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com"},"message":"a copy to try the upgrade on"}`) |
| + requireReady(t, rec) |
| + if rec.Seed.Kind != seeds.KindReplica || rec.Seed.From != fromSource || rec.Seed.TXID == "" || rec.Seed.Bytes == 0 || |
| + len(rec.Seed.SHA256) != 64 || rec.Seed.By != "alice@keymail.dev" || !seeds.ValidID(rec.Seed.ID) { |
| + t.Fatalf("record seed = %+v", rec.Seed) |
| } |
| if rec.Environment != "" || rec.Slug != "canary" || rec.Channel != "stable" { |
| t.Fatalf("record = %+v; want the ordinary create's fields", rec) |
| @@ -186,10 +225,6 @@ func TestFromCreatesAReadySeededInstance(t *testing.T) { |
| if n := snapshotRows(t, raw); n != f.txns*dbdoortest.Rows { |
| t.Fatalf("snapshot holds %d rows, want %d", n, f.txns*dbdoortest.Rows) |
| } |
| - seed, _ := out["seed"].(map[string]any) |
| - if seed == nil || seed["state"] != seeds.StateReady || seed["id"] != rec.Seed.ID || out["host"] != fromHost { |
| - t.Fatalf("response = %v; want the published record", out) |
| - } |
| f.requireClean(t) |
| events, err := f.fx.st.Events(OpsAccountID, fromApp, 50) |
| if err != nil { |
| @@ -236,7 +271,7 @@ func TestFromRefusals(t *testing.T) { |
| } |
| // --at: a target inside the history restores exactly there; one at or |
| -// past the end is the latest copy. |
| +// past the end is the latest copy (ruling R5 keeps that). |
| func TestFromAt(t *testing.T) { |
| f := newFromFixture(t, 30) |
| latest, err := dbdoor.PlanRestore(context.Background(), dbdoor.NewClient(replicaDoer(lsfile.NewReplicaClient(filepath.Join(f.root, "instances", fromSource)))), 0) |
| @@ -252,14 +287,13 @@ func TestFromAt(t *testing.T) { |
| {"c4.shop.example.com", 999999, latest.TXID}, |
| } { |
| body, _ := json.Marshal(map[string]any{"host": c.host, "from": map[string]any{"host": fromSource, "txid": c.txid}}) |
| - if code, out := f.create(t, string(body)); code != http.StatusCreated { |
| - t.Fatalf("--at %d = %d %v", c.txid, code, out) |
| - } |
| - rec, err := f.record(t, c.host) |
| - if err != nil || rec.Seed == nil || rec.Seed.State != seeds.StateReady || rec.Seed.TXID != strconv.FormatUint(c.want, 10) { |
| - t.Fatalf("--at %d: record seed = %+v, %v; want ready at txid %d", c.txid, rec.Seed, err, c.want) |
| + rec := f.copy(t, string(body)) |
| + requireReady(t, rec) |
| + if rec.Seed.TXID != strconv.FormatUint(c.want, 10) { |
| + t.Fatalf("--at %d: restored at txid %s, want %d", c.txid, rec.Seed.TXID, c.want) |
| } |
| } |
| + f.requireClean(t) |
| } |
| func TestFromRefusedBeforeTheFloor(t *testing.T) { |
| @@ -291,14 +325,22 @@ func TestFromTooLarge(t *testing.T) { |
| seedMaxBytes = 1024 // the test's database is larger than 1 KiB |
| defer func() { seedMaxBytes = old }() |
| code, out := f.create(t, fromBody) |
| - if code != http.StatusRequestEntityTooLarge || !strings.Contains(str(t, out, "error"), "a copy of main.shop.example.com would be ") || |
| - !strings.Contains(str(t, out, "error"), " GiB, over the 4 GiB limit.") { |
| + if code != http.StatusRequestEntityTooLarge || str(t, out, "error") != "a copy of main.shop.example.com would be 0.1 GiB, over the 4 GiB limit." { |
| t.Fatalf("= %d %v", code, out) |
| } |
| f.requireNoRecord(t, fromHost) |
| f.requireClean(t) |
| } |
| +// SD-N26 rounds up, so a size over the limit never prints as the limit. |
| +func TestSeedGiBRoundsUp(t *testing.T) { |
| + for n, want := range map[int64]string{1: "0.1 GiB", 4 << 30: "4.0 GiB", 4<<30 + 1: "4.1 GiB", 5 << 29: "2.5 GiB"} { |
| + if got := seedGiB(n); got != want { |
| + t.Errorf("seedGiB(%d) = %q, want %q", n, got, want) |
| + } |
| + } |
| +} |
| + |
| func TestFromSourceRefusals(t *testing.T) { |
| f := newFromFixture(t, 3) |
| f.addSource(t, provision.Record{Host: "pooled.shop.example.com", Slug: "pooled", Pool: "shared"}, 0) |
| @@ -351,8 +393,8 @@ func TestFromPlanFailures(t *testing.T) { |
| {dbdoor.ErrHistoryGone, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, http.StatusConflict, "history before 2 is no longer kept"}, |
| {dbdoor.ErrNothingReplicated, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, http.StatusConflict, "history before 2 is no longer kept"}, |
| {dbdoor.ErrNothingReplicated, fromBody, http.StatusConflict, "main.shop.example.com has not replicated anything yet. Wake it once, then try again."}, |
| - {errors.New("dbdoor: the last planned file's header does not describe it"), fromBody, http.StatusBadGateway, "could not copy main.shop.example.com. Nothing was created. Try again."}, |
| - {os.ErrNotExist, fromBody, http.StatusBadGateway, "could not copy main.shop.example.com. Nothing was created. Try again."}, |
| + {errors.New("dbdoor: the last planned file's header does not describe it"), fromBody, http.StatusBadGateway, fromFailed}, |
| + {os.ErrNotExist, fromBody, http.StatusBadGateway, fromFailed}, |
| } { |
| planRestore = func(context.Context, *dbdoor.Client, uint64) (dbdoor.RestorePlan, error) { |
| return dbdoor.RestorePlan{}, c.err |
| @@ -366,35 +408,37 @@ func TestFromPlanFailures(t *testing.T) { |
| f.requireClean(t) |
| } |
| -// Every failure after the record is reserved aborts it, answers its own |
| -// sentence, and leaves nothing in scratch or the pool. |
| +// Every failure after the record is reserved aborts it with the sentence |
| +// the door would have answered, and leaves nothing in scratch or the pool. |
| func TestFromRestoreFailuresAbort(t *testing.T) { |
| oldRestore, oldDeadline := restoreToTarget, seedRestoreDeadline |
| defer func() { restoreToTarget, seedRestoreDeadline = oldRestore, oldDeadline }() |
| for i, c := range []struct { |
| restore func(ctx context.Context) error |
| body string |
| - code int |
| - want string |
| + reason string |
| }{ |
| - {func(context.Context) error { return dbdoor.ErrCheckFailed }, fromBody, http.StatusUnprocessableEntity, "the copy of main.shop.example.com failed its integrity check, so no instance was created."}, |
| - {func(context.Context) error { return dbdoor.ErrHistoryGone }, fromBody, http.StatusBadGateway, "could not copy main.shop.example.com. Nothing was created. Try again."}, |
| - {func(context.Context) error { return dbdoor.ErrHistoryGone }, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, http.StatusConflict, "history before 2 is no longer kept"}, |
| - {func(context.Context) error { return errors.New("bucket read failed") }, fromBody, http.StatusBadGateway, "could not copy main.shop.example.com. Nothing was created. Try again."}, |
| - {func(ctx context.Context) error { <-ctx.Done(); return ctx.Err() }, fromBody, http.StatusGatewayTimeout, "copying main.shop.example.com took longer than 30 minutes, so no instance was created."}, |
| + {func(context.Context) error { return dbdoor.ErrCheckFailed }, fromBody, "the copy of main.shop.example.com failed its integrity check, so no instance was created."}, |
| + {func(context.Context) error { return dbdoor.ErrHistoryGone }, fromBody, fromFailed}, |
| + {func(context.Context) error { return dbdoor.ErrHistoryGone }, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, "history before 2 is no longer kept"}, |
| + {func(context.Context) error { return errors.New("bucket read failed") }, fromBody, fromFailed}, |
| + {func(ctx context.Context) error { <-ctx.Done(); return ctx.Err() }, fromBody, "copying main.shop.example.com took longer than 30 minutes, so no instance was created."}, |
| } { |
| f := newFromFixture(t, 3) |
| seedRestoreDeadline = 200 * time.Millisecond |
| restoreToTarget = func(ctx context.Context, _ *dbdoor.Client, parent string, _ uint64) (*dbdoor.Restored, error) { |
| + if _, err := os.MkdirTemp(parent, "carlos-db-"); err != nil { // scratch the door must clear |
| + return nil, err |
| + } |
| return nil, c.restore(ctx) |
| } |
| - code, out := f.create(t, c.body) |
| - if code != c.code || str(t, out, "error") != c.want { |
| - t.Errorf("%d: = %d %v, want %d %q", i, code, out, c.code, c.want) |
| + rec := f.copy(t, c.body) |
| + if rec.Seed.State != seeds.StateAborted || rec.Seed.Reason != c.reason { |
| + t.Errorf("%d: seed = %+v, want aborted with %q", i, rec.Seed, c.reason) |
| continue |
| } |
| - f.requireAborted(t, fromHost) |
| f.requireClean(t) |
| + f.requireNothingUploaded(t) |
| } |
| } |
| @@ -413,12 +457,8 @@ func TestFromRestoresAtThePlannedTXID(t *testing.T) { |
| if err != nil { |
| t.Fatal(err) |
| } |
| - if code, out := f.create(t, fromBody); code != http.StatusCreated { |
| - t.Fatalf("= %d %v", code, out) |
| - } |
| - if code, out := f.create(t, `{"host":"c2.shop.example.com","from":{"host":"main.shop.example.com","txid":999999}}`); code != http.StatusCreated { |
| - t.Fatalf("= %d %v", code, out) |
| - } |
| + requireReady(t, f.copy(t, fromBody)) |
| + requireReady(t, f.copy(t, `{"host":"c2.shop.example.com","from":{"host":"main.shop.example.com","txid":999999}}`)) |
| if len(got) != 2 || got[0] != plan.TXID || got[1] != plan.TXID { |
| t.Fatalf("restored at %v, want [%d %d]", got, plan.TXID, plan.TXID) |
| } |
| @@ -435,15 +475,9 @@ func TestFromRefusesARestoreLargerThanPlanned(t *testing.T) { |
| return p, err |
| } |
| defer func() { planRestore = oldPlan }() |
| - code, out := f.create(t, fromBody) |
| - if code != http.StatusRequestEntityTooLarge || str(t, out, "error") != "the copy of main.shop.example.com grew past its planned size, so it was stopped and no instance was created." { |
| - t.Fatalf("= %d %v", code, out) |
| - } |
| - f.requireAborted(t, fromHost) |
| + requireAborted(t, f.copy(t, fromBody), fromGrew) |
| f.requireClean(t) |
| - if keys, err := f.data.List(context.Background(), seeds.Prefix); err != nil || len(keys) != 0 { |
| - t.Fatalf("seeds/ = %v, %v; want nothing uploaded", keys, err) |
| - } |
| + f.requireNothingUploaded(t) |
| } |
| // The watcher cancels a restore that is still running: litestream writes |
| @@ -484,96 +518,142 @@ func TestFromWatcherCancelsARestoreInFlight(t *testing.T) { |
| } |
| } |
| } |
| - code, out := f.create(t, fromBody) |
| + rec := f.copy(t, fromBody) |
| if !<-cancelled { |
| t.Fatal("the restore ran to its own timeout: the watcher did not cancel it in flight") |
| } |
| - if code != http.StatusRequestEntityTooLarge || str(t, out, "error") != "the copy of main.shop.example.com grew past its planned size, so it was stopped and no instance was created." { |
| - t.Fatalf("= %d %v", code, out) |
| - } |
| - f.requireAborted(t, fromHost) |
| + requireAborted(t, rec, fromGrew) |
| f.requireClean(t) |
| } |
| -func TestFromDisconnectAbortsTheRecordAndCleansScratch(t *testing.T) { |
| - f := newFromFixture(t, 30) |
| - ctx, cancel := context.WithCancel(context.Background()) |
| - oldRestore := restoreToTarget |
| - restoreToTarget = func(rctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) { |
| - if _, err := os.MkdirTemp(parent, "carlos-db-"); err != nil { // scratch the door must clear |
| - return nil, err |
| - } |
| - cancel() // the client goes away mid-restore |
| - <-rctx.Done() |
| - return nil, rctx.Err() |
| - } |
| - defer func() { restoreToTarget = oldRestore }() |
| - req, _ := http.NewRequestWithContext(ctx, http.MethodPost, f.url, strings.NewReader(fromBody)) |
| - req.Header.Set("Authorization", "Bearer "+f.token) |
| - req.Header.Set("Content-Type", "application/json") |
| - if res, err := http.DefaultClient.Do(req); err == nil { |
| - res.Body.Close() |
| - t.Fatalf("the request finished with %d; want the client to have gone away", res.StatusCode) |
| - } |
| - deadline := time.Now().Add(5 * time.Second) |
| - for time.Now().Before(deadline) { |
| - rec, err := f.record(t, fromHost) |
| - if err == nil && rec.Seed != nil && rec.Seed.State == seeds.StateAborted { |
| - if ents, _ := os.ReadDir(f.fx.srv.seedScratch.dir()); len(ents) == 0 { |
| - if rel, ref := f.fx.srv.seedScratch.reserve("bab/"+fromApp, 1); ref == nil { |
| - rel() |
| - return |
| +// Once the door has answered 202 the copy no longer depends on the |
| +// request: a client that goes away, and takes its connection with it, |
| +// leaves the copy to finish, ready or aborted, and to clean up after |
| +// itself either way. |
| +func TestFromCopyOutlivesItsClient(t *testing.T) { |
| + for _, c := range []struct { |
| + name string |
| + fail error |
| + reason string |
| + }{ |
| + {"ready", nil, ""}, |
| + {"aborted", dbdoor.ErrCheckFailed, "the copy of main.shop.example.com failed its integrity check, so no instance was created."}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + f := newFromFixture(t, 3) |
| + oldRestore := restoreToTarget |
| + defer func() { restoreToTarget = oldRestore }() |
| + started, proceed := make(chan struct{}), make(chan struct{}) |
| + restoreToTarget = func(rctx context.Context, cl *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) { |
| + close(started) |
| + <-proceed |
| + if rctx.Err() != nil { |
| + return nil, rctx.Err() |
| + } |
| + if c.fail != nil { |
| + return nil, c.fail |
| } |
| + return oldRestore(rctx, cl, parent, target) |
| } |
| - } |
| - time.Sleep(20 * time.Millisecond) |
| + tr := &http.Transport{} |
| + client := &http.Client{Transport: tr} |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, f.url, strings.NewReader(fromBody)) |
| + req.Header.Set("Authorization", "Bearer "+f.token) |
| + req.Header.Set("Content-Type", "application/json") |
| + res, err := client.Do(req) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + res.Body.Close() |
| + if res.StatusCode != http.StatusAccepted { |
| + t.Fatalf("= %d, want 202", res.StatusCode) |
| + } |
| + <-started |
| + cancel() // the client goes away, and its connection with it |
| + tr.CloseIdleConnections() |
| + close(proceed) |
| + f.fx.srv.seedCopies.Wait() |
| + rec, err := f.record(t, fromHost) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if c.fail == nil { |
| + requireReady(t, rec) |
| + } else { |
| + requireAborted(t, rec, c.reason) |
| + } |
| + f.requireClean(t) |
| + }) |
| } |
| - t.Fatal("a disconnected --from did not abort its record, clear its scratch and release its reservation") |
| } |
| -// A copy that runs longer than the server's ReadTimeout still completes. |
| -// (On Go 1.26 net/http clears the read deadline once the body is read, so |
| -// this holds with or without the door's own extension; it pins the |
| -// behaviour, not the mechanism.) |
| -func TestFromOutlivesTheServerReadTimeout(t *testing.T) { |
| +func TestFromUploadFailureAborts(t *testing.T) { |
| f := newFromFixture(t, 3) |
| - ts := httptest.NewUnstartedServer(f.fx.srv) |
| - ts.Config.ReadTimeout = 300 * time.Millisecond |
| - ts.Start() |
| - defer ts.Close() |
| + f.fx.srv.cfg.SeedData = failingPuts{f.data} |
| + requireAborted(t, f.copy(t, fromBody), fromFailed) |
| + f.requireClean(t) |
| +} |
| + |
| +type failingPuts struct{ release.Store } |
| + |
| +func (failingPuts) Put(context.Context, string, []byte, string) (string, error) { |
| + return "", errors.New("access denied") |
| +} |
| + |
| +// A publish that fails aborts the seed, with the retry sentence. |
| +func TestFromPublishFailureAborts(t *testing.T) { |
| + f := newFromFixture(t, 3) |
| + rel := &failOnceOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, fromHost)} |
| + f.fx.srv.cfg.Releases = rel |
| oldRestore := restoreToTarget |
| defer func() { restoreToTarget = oldRestore }() |
| restoreToTarget = func(ctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) { |
| - select { |
| - case <-ctx.Done(): |
| - return nil, ctx.Err() |
| - case <-time.After(1200 * time.Millisecond): // four times the server's ReadTimeout |
| - } |
| + rel.armed.Store(true) // the next write of the record is the publish |
| return oldRestore(ctx, c, parent, target) |
| } |
| - code, _, raw := rawReq(t, http.MethodPost, ts.URL+"/api/cli/apps/bab/shop/instances", f.token, fromBody) |
| - if code != http.StatusCreated { |
| - t.Fatalf("a copy longer than the ReadTimeout = %d %s, want 201", code, raw) |
| + requireAborted(t, f.copy(t, fromBody), fromFailed) |
| + f.requireClean(t) |
| +} |
| + |
| +// failOnceOn fails the first Put of key once armed. |
| +type failOnceOn struct { |
| + release.Store |
| + key string |
| + armed atomic.Bool |
| +} |
| + |
| +func (s *failOnceOn) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if key == s.key && s.armed.CompareAndSwap(true, false) { |
| + return "", errors.New("the bucket refused the write") |
| } |
| + return s.Store.Put(ctx, key, body, ifMatch) |
| } |
| -func TestFromUploadFailureAborts(t *testing.T) { |
| +// A record that appears between the create door's checks and this door's |
| +// reservation is answered as the ordinary create answers it, and leaves |
| +// nothing behind. |
| +func TestFromReservationLosesARace(t *testing.T) { |
| f := newFromFixture(t, 3) |
| - f.fx.srv.cfg.SeedData = failingPuts{f.data} |
| + oldPlan := planRestore |
| + defer func() { planRestore = oldPlan }() |
| + planRestore = func(ctx context.Context, c *dbdoor.Client, target uint64) (dbdoor.RestorePlan, error) { |
| + // Another writer's record, written from the handler's goroutine: |
| + // the test checks it below rather than failing here. |
| + _ = provision.CreateRecord(ctx, f.fx.rel, provision.Record{Schema: 1, Host: fromHost, Slug: "canary", |
| + App: fromApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()}) |
| + return oldPlan(ctx, c, target) |
| + } |
| code, out := f.create(t, fromBody) |
| - if code != http.StatusBadGateway || str(t, out, "error") != "could not copy main.shop.example.com. Nothing was created. Try again." { |
| + if code != http.StatusConflict || str(t, out, "error") != "instance canary.shop.example.com already exists" { |
| t.Fatalf("= %d %v", code, out) |
| } |
| - f.requireAborted(t, fromHost) |
| + if rec, err := f.record(t, fromHost); err != nil || rec.Seed != nil { |
| + t.Fatalf("record = %+v, %v; want the other writer's, unseeded", rec, err) |
| + } |
| f.requireClean(t) |
| } |
| -type failingPuts struct{ release.Store } |
| - |
| -func (failingPuts) Put(context.Context, string, []byte, string) (string, error) { |
| - return "", errors.New("access denied") |
| -} |
| - |
| func TestFromNeedsADataBucket(t *testing.T) { |
| f := newFromFixture(t, 3) |
| f.fx.srv.cfg.SeedData = nil |
| @@ -601,12 +681,10 @@ func TestFromRefusesAPooledApp(t *testing.T) { |
| func TestFromByAnInstanceServiceRole(t *testing.T) { |
| f := newFromFixture(t, 3) |
| svc := serviceToken(t, f.fx.st, OpsAccountID, "ci", fromApp, svcrole.Instance) |
| - if code, out := f.createAs(t, svc, fromBody); code != http.StatusCreated { |
| - t.Fatalf("= %d %v", code, out) |
| - } |
| - rec, err := f.record(t, fromHost) |
| - if err != nil || rec.Seed == nil || rec.Seed.State != seeds.StateReady || rec.Seed.By != "service:bab/ci" { |
| - t.Fatalf("record seed = %+v, %v", rec.Seed, err) |
| + rec := f.copyAs(t, svc, fromBody) |
| + requireReady(t, rec) |
| + if rec.Seed.By != "service:bab/ci" { |
| + t.Fatalf("record seed by %q, want the service principal", rec.Seed.By) |
| } |
| } |
| @@ -631,6 +709,39 @@ func TestFromPrivateStorageRules(t *testing.T) { |
| f.requireClean(t) |
| } |
| +// The store gates that write nothing refuse before the record is reserved |
| +// (ruling R3), with the store plane's own answers. |
| +func TestFromStoreGatesRefuseBeforeTheReservation(t *testing.T) { |
| + for name, unwire := range map[string]func(*Server){ |
| + "no backend": func(s *Server) { s.cfg.StoreProvisioner = nil }, |
| + "no sealing key": func(s *Server) { s.cfg.SecretsPub = "" }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + f := newFromFixture(t, 3) |
| + enableInstances(t, f.fx, fromApp, fromManifest(true)) |
| + unwire(f.fx.srv) |
| + code, out := f.create(t, fromBody) |
| + if code < 500 || str(t, out, "error") == "" { |
| + t.Fatalf("= %d %v, want the store plane's refusal", code, out) |
| + } |
| + f.requireNoRecord(t, fromHost) |
| + f.requireClean(t) |
| + }) |
| + } |
| +} |
| + |
| +// A store walk refused after the reservation aborts the seed with the walk's |
| +// own sentence as its reason. |
| +func TestFromStoreWalkRefusalAborts(t *testing.T) { |
| + f := newFromFixture(t, 3) |
| + enableInstances(t, f.fx, fromApp, fromManifest(true)) |
| + takeCanaryBucketName(t, f.fx) |
| + rec := f.copy(t, fromBody) |
| + requireAborted(t, rec, "another store already uses this instance's storage names (app and environment names are joined with \"-\"), so the instance was not created. use a different host name") |
| + f.requireClean(t) |
| + f.requireNothingUploaded(t) |
| +} |
| + |
| func putCanaryStoreOf(t *testing.T, fx *consoleFixture, app string, rec *objstore.Record) { |
| t.Helper() |
| if _, err := objstore.PutIn(context.Background(), fx.rel, release.OpsAccount, app, "canary", rec, ""); err != nil { |
| @@ -655,25 +766,20 @@ func TestFromPrivateStorageGetsANewStoreMarkedWithTheSeed(t *testing.T) { |
| if err != nil { |
| t.Fatal(err) |
| } |
| - code, out := f.create(t, fromBody) |
| - if code != http.StatusCreated { |
| - t.Fatalf("= %d %v", code, out) |
| - } |
| - rec, err := f.record(t, fromHost) |
| - if err != nil || rec.Seed == nil || rec.Seed.State != seeds.StateReady || rec.Environment != "canary" { |
| - t.Fatalf("record = %+v, %v", rec, err) |
| + rec := f.copy(t, fromBody) |
| + requireReady(t, rec) |
| + if rec.Environment != "canary" { |
| + t.Fatalf("record environment = %q, want the host's own", rec.Environment) |
| } |
| store, _, err := objstore.GetIn(context.Background(), f.fx.rel, release.OpsAccount, fromApp, "canary") |
| if err != nil || store == nil || store.Declared == nil || store.Declared.Seed != rec.Seed.ID || storeBucketName(store) == "" { |
| t.Fatalf("canary store = %+v, %v; want one marked with seed %s", store, err, rec.Seed.ID) |
| } |
| - if out["private_store_bucket"] != storeBucketName(store) { |
| - t.Fatalf("response private_store_bucket = %v, want %s", out["private_store_bucket"], storeBucketName(store)) |
| - } |
| after, _, err := f.fx.rel.Get(context.Background(), mainKey) |
| if err != nil || string(after) != string(before) { |
| t.Fatalf("the source's store record changed:\n before %s\n after %s (%v)", before, after, err) |
| } |
| + f.requireClean(t) |
| } |
| // The create door's sibling-apex re-check holds on the long --from path |
| @@ -694,15 +800,21 @@ func TestFromAbortsWhenASiblingTakesTheApex(t *testing.T) { |
| } |
| return oldRestore(ctx, c, parent, target) |
| } |
| - code, out := f.create(t, `{"host":"canary.bab.apps.example.net","from":{"host":"main.bab.apps.example.net"}}`) |
| - want := "canary.bab.apps.example.net is app canary's platform apex — it was created while this instance was being written; the instance was not kept" |
| - if code != http.StatusConflict || str(t, out, "error") != want { |
| - t.Fatalf("= %d %v", code, out) |
| - } |
| - f.requireAborted(t, "canary.bab.apps.example.net") |
| + rec := f.copy(t, `{"host":"canary.bab.apps.example.net","from":{"host":"main.bab.apps.example.net"}}`) |
| + requireAborted(t, rec, "canary.bab.apps.example.net is app canary's platform apex — it was created while this instance was being written; the instance was not kept") |
| f.requireClean(t) |
| } |
| +// A door that held no answer never sends an empty 200 (ruling R1). |
| +func TestHeldResponseNeverAnswersEmpty(t *testing.T) { |
| + w := httptest.NewRecorder() |
| + (&heldResponse{header: w.Header()}).flush(w, fromFailed) |
| + var out map[string]string |
| + if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil || w.Code != http.StatusBadGateway || out["error"] != fromFailed { |
| + t.Fatalf("= %d %s, want 502 %q", w.Code, w.Body, fromFailed) |
| + } |
| +} |
| + |
| func TestRestoreScratchUse(t *testing.T) { |
| dir := t.TempDir() |
| sub := filepath.Join(dir, "carlos-db-x") |
diff --git a/internal/console/instanceprivatestore.go b/internal/console/instanceprivatestore.go| index 4f6475c33..982e1a7c5 100644 |
| --- a/internal/console/instanceprivatestore.go |
| +++ b/internal/console/instanceprivatestore.go |
| @@ -20,6 +20,7 @@ package console |
| import ( |
| "context" |
| + "crypto/ecdh" |
| "errors" |
| "fmt" |
| "log" |
| @@ -203,28 +204,11 @@ var errSeedStoreTaken = errors.New("console: the environment's store is not this |
| // rolled back under the lock: deleted if this call created the record, |
| // else restored to its previous bytes. |
| func (s *Server) instanceStoreProvisionFor(ctx context.Context, w http.ResponseWriter, acct *Account, app, env, actor, seedID string) (*objstore.Record, bool, error) { |
| - if s.refuseTemporary(w, acct, app, "give it object storage") { |
| - return nil, false, nil |
| - } |
| - effectStore, effectProvisioner := claimStoreAdapters(ctx, s.cfg.Releases, s.cfg.StoreProvisioner) |
| - if s.cfg.StoreProvisioner == nil { |
| - // The manifest carries a policy this deployment cannot honour. The |
| - // enable door refuses to write one in this state (§5), so reaching |
| - // here means the backend was unwired afterwards — a configuration |
| - // fact, named as one. |
| - jsonError(w, http.StatusServiceUnavailable, |
| - "object storage is not configured for this deployment (CARLOS_STORE_BACKEND), and "+app+" gives every instance its own storage — no instance can be created until one is wired or the policy is turned off (carlos instances enable --private-storage=false)") |
| - return nil, false, nil |
| - } |
| - preCtx, preCancel := context.WithTimeout(ctx, storeTimeout) |
| - defer preCancel() |
| - if _, ok := s.storeProvisionApproval(preCtx, w, acct); !ok { |
| - return nil, false, nil |
| - } |
| - pub, ok := s.storeSealKey(w) |
| + pub, ok := s.instanceStorePreflight(ctx, w, acct, app) |
| if !ok { |
| return nil, false, nil |
| } |
| + effectStore, effectProvisioner := claimStoreAdapters(ctx, s.cfg.Releases, s.cfg.StoreProvisioner) |
| // The same lock handleAPIStoreProvision takes, keyed the same way, so a |
| // signup creating an instance and a member running `carlos store create |
| // --environment` against the same name serialise against each other. |
| @@ -361,6 +345,32 @@ func (s *Server) instanceStoreProvisionFor(ctx context.Context, w http.ResponseW |
| return out, true, nil |
| } |
| +// instanceStorePreflight is the store plane's gates that write nothing and |
| +// take no lock: a temporary app, no backend, no standing approval, no |
| +// sealing key. It answers w itself on a refusal. instanceStoreProvisionFor |
| +// runs it first; a seed door runs it before reserving its record, so these |
| +// refusals never leave an aborted record behind. |
| +func (s *Server) instanceStorePreflight(ctx context.Context, w http.ResponseWriter, acct *Account, app string) (*ecdh.PublicKey, bool) { |
| + if s.refuseTemporary(w, acct, app, "give it object storage") { |
| + return nil, false |
| + } |
| + if s.cfg.StoreProvisioner == nil { |
| + // The manifest carries a policy this deployment cannot honour. The |
| + // enable door refuses to write one in this state (§5), so reaching |
| + // here means the backend was unwired afterwards — a configuration |
| + // fact, named as one. |
| + jsonError(w, http.StatusServiceUnavailable, |
| + "object storage is not configured for this deployment (CARLOS_STORE_BACKEND), and "+app+" gives every instance its own storage — no instance can be created until one is wired or the policy is turned off (carlos instances enable --private-storage=false)") |
| + return nil, false |
| + } |
| + preCtx, preCancel := context.WithTimeout(ctx, storeTimeout) |
| + defer preCancel() |
| + if _, ok := s.storeProvisionApproval(preCtx, w, acct); !ok { |
| + return nil, false |
| + } |
| + return s.storeSealKey(w) |
| +} |
| + |
| // seedOwnsStore reports whether a seeded create may use rec (spec §4 |
| // Storage): there is no store, the record holds no estate yet, or the |
| // estate was made by this very seed. |
diff --git a/internal/console/server.go b/internal/console/server.go| index f728d69c2..5f3b37eb5 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -837,8 +837,11 @@ type Server struct { |
| claimWorkers map[string]func(context.Context) |
| dbHandleKey []byte |
| seedScratch *seedScratch |
| - cfg Config |
| - store *Store |
| + // seedCopies counts --from copies running detached from their request |
| + // (apiinstanceseed.go), so a test can wait for one to finish. |
| + seedCopies sync.WaitGroup |
| + cfg Config |
| + store *Store |
| // appRegionOf and beforeClaimInsert are TEST SEAMS (nil in |
| // production): a store failure on the app-region read (C20), and a |
| // competing insert landing after the region was resolved. |
diff --git a/internal/provision/seed.go b/internal/provision/seed.go| index 084b04f12..f97abcbc4 100644 |
| --- a/internal/provision/seed.go |
| +++ b/internal/provision/seed.go |
| @@ -21,7 +21,7 @@ type SeedSource struct { |
| SHA256 string `json:"sha256"` // of snapshot.db |
| Bytes int64 `json:"bytes"` |
| State string `json:"state"` // seeds.State* |
| - Reason string `json:"reason,omitempty"` // set with refused |
| + Reason string `json:"reason,omitempty"` // set with refused, and with aborted when the abort names why |
| By string `json:"by"` // the caller |
| At time.Time `json:"at"` |
| // import only: |
| @@ -189,7 +189,13 @@ func SeedRefuse(id, reason string) func(*Record) error { |
| // specs scope that rule to the import; a --from seed's abort is its door's |
| // own failure path, and a run admitted on the reserved record must not |
| // leave it stuck reserving. |
| -func SeedAbort(id string) func(*Record) error { |
| +func SeedAbort(id string) func(*Record) error { return SeedAbortBecause(id, "") } |
| + |
| +// SeedAbortBecause is SeedAbort recording why in Reason: the sentence the |
| +// --from door answers its caller with when its copy fails after the record |
| +// is reserved (spec §4, Plan 5 ruling R2). An empty reason keeps whatever |
| +// Reason the seed already had (an import refused before its abort). |
| +func SeedAbortBecause(id, reason string) func(*Record) error { |
| return func(rec *Record) error { |
| from := []string{seeds.StateReserving, seeds.StateReady} |
| if rec.Seed.Import() { |
| @@ -203,6 +209,9 @@ func SeedAbort(id string) func(*Record) error { |
| return &RunOpenError{Run: *rec.ActiveRun} |
| } |
| rec.Seed.State = seeds.StateAborted |
| + if reason != "" { |
| + rec.Seed.Reason = reason |
| + } |
| return nil |
| } |
| } |
diff --git a/internal/provision/seed_test.go b/internal/provision/seed_test.go| index 7d5c06798..b7ce4f444 100644 |
| --- a/internal/provision/seed_test.go |
| +++ b/internal/provision/seed_test.go |
| @@ -227,6 +227,29 @@ func TestSeedRefuseRecordsItsReason(t *testing.T) { |
| } |
| } |
| +// A --from seed's abort records why; an empty reason keeps the one a |
| +// refused import already carries. |
| +func TestSeedAbortBecauseRecordsItsReason(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + seed := testSeed(t, seeds.KindReplica, seeds.StateReserving) |
| + rec := seedTestRecord(t, s, seed) |
| + got, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbortBecause(seed.ID, "the copy failed its check")) |
| + if err != nil || got.Seed.State != seeds.StateAborted || got.Seed.Reason != "the copy failed its check" { |
| + t.Fatalf("abort because = %+v, %v", got.Seed, err) |
| + } |
| + s = release.NewDirStore(t.TempDir()) |
| + seed = testSeed(t, seeds.KindImport, seeds.StateReady) |
| + rec = seedTestRecord(t, s, seed) |
| + if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedRefuse(seed.ID, "the route belongs to app blog")); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, err = SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(seed.ID)) |
| + if err != nil || got.Seed.State != seeds.StateAborted || got.Seed.Reason != "the route belongs to app blog" { |
| + t.Fatalf("abort of a refused import = %+v, %v; want its refusal reason kept", got.Seed, err) |
| + } |
| +} |
| + |
| func TestSeedColumn(t *testing.T) { |
| if SeedColumn(Record{}) != "" { |
| t.Error("a seedless record has a column value") |