console: the import upload re-reads under its reservation and never acts on an import that moved on (Plan 6 Task 2 fix round 1)
The upload re-reads the record after reserving and again before the put, and goes on only while it is awaiting-upload with this seed id. The put is create-only and never deletes a snapshot: losing it ends the detached half with no abort. The detached abort is the new SeedAbortUploadBecause, which refuses with ErrSeedConflict once the seed has left awaiting-upload; a moved-on import is logged and left alone.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+342
−37
internal/console/apiinstanceimport.go+110 −28internal/console/apiinstanceimport_test.go+190 −9internal/provision/seed.go+15 −0internal/provision/seed_test.go+27 −0
CI passed — run details
diff --git a/internal/console/apiinstanceimport.go b/internal/console/apiinstanceimport.go| index e8ea428ec..9d02793c3 100644 |
| --- a/internal/console/apiinstanceimport.go |
| +++ b/internal/console/apiinstanceimport.go |
| @@ -255,6 +255,9 @@ var ( |
| // importSpoolHook, when set, sees the spool once the body is read and |
| // before it is checked or removed. |
| importSpoolHook func(path string) |
| + // importReserveHook, when set, runs after the upload's first read of |
| + // the record and before its reservation. |
| + importReserveHook func(host string) |
| ) |
| // importUploadReadDeadline is how long the upload's body may take to |
| @@ -297,7 +300,8 @@ func (s *Server) importGate(w http.ResponseWriter, r *http.Request) (principal s |
| return "", nil, "", "", nil, false |
| } |
| if !rec.Seed.Import() { |
| - jsonError(w, http.StatusConflict, fmt.Sprintf("%s is not an import.", host)) |
| + ref := notAnImportRefusal(host) |
| + jsonError(w, ref.Status, ref.Msg) |
| return "", nil, "", "", nil, false |
| } |
| return principal, acct, app, host, rec, true |
| @@ -332,8 +336,8 @@ func (s *Server) handleAPIImportUpload(w http.ResponseWriter, r *http.Request) { |
| } |
| ctx := r.Context() |
| seed := rec.Seed |
| - if seed.State != seeds.StateAwaitingUpload { |
| - jsonError(w, http.StatusConflict, fmt.Sprintf("%s is %s. It takes an upload only while it waits for one.", host, seed.State)) |
| + if ref := staleUploadRefusal(host, rec, seed.ID); ref != nil { |
| + jsonError(w, ref.Status, ref.Msg) |
| return |
| } |
| // The snapshot key is built from the record (preflight M8): an id that |
| @@ -355,11 +359,34 @@ func (s *Server) handleAPIImportUpload(w http.ResponseWriter, r *http.Request) { |
| jsonError(w, ref.Status, ref.Msg) |
| return |
| } |
| + if importReserveHook != nil { |
| + importReserveHook(host) |
| + } |
| releaseScratch, ref := s.seedScratch.reserve(acct.Sqid+"/"+app, seed.Bytes) |
| if ref != nil { |
| jsonError(w, ref.Status, ref.Msg) |
| return |
| } |
| + // The read above came before the reservation: another upload of this |
| + // import may have landed in between (fix round 1). Under the |
| + // reservation, the record must still wait for this seed's upload. |
| + fresh, err := s.readImportRecord(ctx, acct.Sqid, app, host) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + releaseScratch() |
| + jsonError(w, http.StatusNotFound, fmt.Sprintf("%s is not an instance of %s", host, app)) |
| + return |
| + case err != nil: |
| + releaseScratch() |
| + log.Printf("console: import %s/%s/%s: re-read the record: %v", acct.Sqid, app, host, err) |
| + jsonError(w, http.StatusBadGateway, importFailed) |
| + return |
| + } |
| + if ref := staleUploadRefusal(host, fresh, seed.ID); ref != nil { |
| + releaseScratch() |
| + jsonError(w, ref.Status, ref.Msg) |
| + return |
| + } |
| dir, err := os.MkdirTemp(s.seedScratch.dir(), seedScratchPrefix) |
| if err != nil { |
| releaseScratch() |
| @@ -386,6 +413,37 @@ func (s *Server) handleAPIImportUpload(w http.ResponseWriter, r *http.Request) { |
| id: seed.ID, dir: dir, spool: spool, release: releaseScratch}) |
| } |
| +// readImportRecord reads host's record under instanceStoreTimeout. |
| +func (s *Server) readImportRecord(ctx context.Context, sqid, app, host string) (*provision.Record, error) { |
| + ctx, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| + defer cancel() |
| + return provision.GetRecord(ctx, s.cfg.Releases, sqid, app, host) |
| +} |
| + |
| +// awaitingUpload reports whether rec's seed is the import id, still |
| +// waiting for its upload. |
| +func awaitingUpload(rec *provision.Record, id string) bool { |
| + return rec.Seed.Import() && rec.Seed.ID == id && rec.Seed.State == seeds.StateAwaitingUpload |
| +} |
| + |
| +// notAnImportRefusal is SD-N59. |
| +func notAnImportRefusal(host string) *doorRefusal { |
| + return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s is not an import.", host)} |
| +} |
| + |
| +// staleUploadRefusal is the upload's answer to a record that no longer |
| +// waits for seed id's upload: SD-N53 naming its state, or SD-N59 when it |
| +// is no longer an import at all. |
| +func staleUploadRefusal(host string, rec *provision.Record, id string) *doorRefusal { |
| + switch { |
| + case awaitingUpload(rec, id): |
| + return nil |
| + case !rec.Seed.Import(): |
| + return notAnImportRefusal(host) |
| + } |
| + return &doorRefusal{http.StatusConflict, fmt.Sprintf("%s is %s. It takes an upload only while it waits for one.", host, rec.Seed.State)} |
| +} |
| + |
| // spoolImport reads the body into spool, never more than the declared |
| // length and one byte, hashing as it writes, and reports whether it |
| // matches begin's length and sha256. It has answered w when it reports |
| @@ -431,10 +489,14 @@ func (s *Server) spoolImport(w http.ResponseWriter, r *http.Request, spool strin |
| // runImportUpload is the detached half: the integrity check, the put to |
| // the seed's snapshot key, and the move to uploaded. It never depends on |
| -// the request, which has been answered. Every failure moves the record to |
| +// the request, which has been answered. A failure moves the record to |
| // aborted with the sentence the door would have answered in Seed.Reason |
| -// (SD-N52, SD-N61 or SD-N67). On every exit it removes the spool's |
| -// directory and then releases the reservation, as runFrom does. |
| +// (SD-N52, SD-N61 or SD-N67), but only while the record still waits for |
| +// this upload (SeedAbortUploadBecause): an import that has moved on, by |
| +// another upload, an abort or a delete, is never touched, and neither is |
| +// a snapshot already at its key (fix round 1). On every exit it removes |
| +// the spool's directory and then releases the reservation, as runFrom |
| +// does. |
| // |
| // The console has no shutdown hook to cancel a running check: a restart |
| // leaves the record awaiting-upload, its spool is cleared at start, and |
| @@ -447,7 +509,12 @@ func (s *Server) runImportUpload(j *importUpload) { |
| abort := func(reason, why string) { |
| bg, cancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| defer cancel() |
| - if _, err := provision.SetRecordSeed(bg, s.cfg.Releases, sqid, app, host, provision.SeedAbortBecause(j.id, reason)); err != nil { |
| + _, err := provision.SetRecordSeed(bg, s.cfg.Releases, sqid, app, host, provision.SeedAbortUploadBecause(j.id, reason)) |
| + switch { |
| + case errors.Is(err, provision.ErrSeedConflict), errors.Is(err, os.ErrNotExist): |
| + log.Printf("console: import %s: seed %s moved on before its upload's %s was recorded; not aborted: %v", host, j.id, why, err) |
| + return |
| + case err != nil: |
| log.Printf("console: import %s: abort seed %s after %s: %v", host, j.id, why, err) |
| } |
| s.bumpSerial(bg) |
| @@ -477,9 +544,30 @@ func (s *Server) runImportUpload(j *importUpload) { |
| abort("the upload is not a database that passes an integrity check, so it was not kept.", "the check") |
| return |
| } |
| + // The check can take half an hour: put nothing for an import that has |
| + // moved on meanwhile (finished, aborted, or landed by another upload). |
| + rec, err := s.readImportRecord(ctx, sqid, app, host) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + log.Printf("console: import %s: the record went while seed %s was checked; nothing put", host, j.id) |
| + return |
| + case err != nil: |
| + log.Printf("console: import %s: re-read before the put: %v", host, err) |
| + abort(importFailed, "the re-read") |
| + return |
| + case !awaitingUpload(rec, j.id): |
| + log.Printf("console: import %s: seed %s moved on while it was checked (%+v); nothing put", host, j.id, rec.Seed) |
| + return |
| + } |
| uctx, ucancel := context.WithTimeout(ctx, seedUploadDeadline) |
| err = putSnapshot(uctx, s.cfg.SeedData, j.spool, seeds.SnapshotKey(host, j.id)) |
| ucancel() |
| + if errors.Is(err, release.ErrCASConflict) { |
| + // Another writer put this seed's snapshot first, and owns the |
| + // outcome. Nothing is overwritten and nothing aborted. |
| + log.Printf("console: import %s: the snapshot of seed %s is already there; left as it is", host, j.id) |
| + return |
| + } |
| if err != nil { |
| log.Printf("console: import %s: put the snapshot: %v", host, err) |
| abort(importFailed, "the put") |
| @@ -487,7 +575,12 @@ func (s *Server) runImportUpload(j *importUpload) { |
| } |
| pctx, pcancel := context.WithTimeout(ctx, instanceStoreTimeout) |
| defer pcancel() |
| - if _, err := provision.SetRecordSeed(pctx, s.cfg.Releases, sqid, app, host, provision.SeedUploaded(j.id)); err != nil { |
| + _, err = provision.SetRecordSeed(pctx, s.cfg.Releases, sqid, app, host, provision.SeedUploaded(j.id)) |
| + switch { |
| + case errors.Is(err, provision.ErrSeedConflict), errors.Is(err, os.ErrNotExist): |
| + log.Printf("console: import %s: seed %s moved on before it was marked uploaded; not aborted: %v", host, j.id, err) |
| + return |
| + case err != nil: |
| log.Printf("console: import %s: record uploaded: %v", host, err) |
| abort(importFailed, "the record write") |
| return |
| @@ -497,29 +590,18 @@ func (s *Server) runImportUpload(j *importUpload) { |
| } |
| // putSnapshot writes path to key with one create-only PutFrom (spec §5.2 |
| -// step 4). An object already there is a failed earlier attempt for this |
| -// same seed (the key is scoped to its id): it is deleted and the put |
| -// retried once. |
| +// step 4). It never replaces an object already there: that answers |
| +// release.ErrCASConflict. |
| func putSnapshot(ctx context.Context, data release.Store, path, key string) error { |
| - put := func() error { |
| - f, err := os.Open(path) |
| - if err != nil { |
| - return err |
| - } |
| - defer f.Close() |
| - st, err := f.Stat() |
| - if err != nil { |
| - return err |
| - } |
| - _, err = release.PutObjectFrom(ctx, data, key, f, st.Size(), "") |
| + f, err := os.Open(path) |
| + if err != nil { |
| return err |
| } |
| - err := put() |
| - if errors.Is(err, release.ErrCASConflict) { |
| - if err := data.Delete(ctx, key); err != nil { |
| - return err |
| - } |
| - err = put() |
| + defer f.Close() |
| + st, err := f.Stat() |
| + if err != nil { |
| + return err |
| } |
| + _, err = release.PutObjectFrom(ctx, data, key, f, st.Size(), "") |
| return err |
| } |
diff --git a/internal/console/apiinstanceimport_test.go b/internal/console/apiinstanceimport_test.go| index dffb09711..11189bd8e 100644 |
| --- a/internal/console/apiinstanceimport_test.go |
| +++ b/internal/console/apiinstanceimport_test.go |
| @@ -7,6 +7,7 @@ import ( |
| "database/sql" |
| "encoding/hex" |
| "encoding/json" |
| + "errors" |
| "log" |
| "net/http" |
| "os" |
| @@ -491,29 +492,50 @@ const ( |
| // answered. |
| func uploadImport(t *testing.T, f fromFixture, host string, body []byte) (int, map[string]any) { |
| t.Helper() |
| - req, err := http.NewRequest(http.MethodPut, f.url+"/"+host+"/import", bytes.NewReader(body)) |
| + code, out, err := putImport(f, host, body) |
| if err != nil { |
| t.Fatal(err) |
| } |
| + return code, out |
| +} |
| + |
| +// putImport is uploadImport for a goroutine: it returns its error. |
| +func putImport(f fromFixture, host string, body []byte) (int, map[string]any, error) { |
| + req, err := http.NewRequest(http.MethodPut, f.url+"/"+host+"/import", bytes.NewReader(body)) |
| + if err != nil { |
| + return 0, nil, err |
| + } |
| req.Header.Set("Authorization", "Bearer "+f.token) |
| res, err := http.DefaultClient.Do(req) |
| if err != nil { |
| - t.Fatal(err) |
| + return 0, nil, err |
| } |
| defer res.Body.Close() |
| var out map[string]any |
| _ = json.NewDecoder(res.Body).Decode(&out) |
| - return res.StatusCode, out |
| + return res.StatusCode, out, nil |
| } |
| // sqliteBytes is a small, sound SQLite database and its sha256. |
| func sqliteBytes(t *testing.T) ([]byte, string) { |
| + t.Helper() |
| + return sqliteBytesIn(t, "") |
| +} |
| + |
| +// sqliteBytesIn is sqliteBytes with a journal mode set first ("" keeps |
| +// the default). |
| +func sqliteBytesIn(t *testing.T, journal string) ([]byte, string) { |
| t.Helper() |
| p := filepath.Join(t.TempDir(), "snap.db") |
| db, err := sql.Open("sqlite", p) |
| if err != nil { |
| t.Fatal(err) |
| } |
| + if journal != "" { |
| + if _, err := db.Exec(`PRAGMA journal_mode=` + journal); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| if _, err := db.Exec(`CREATE TABLE t (x INTEGER); INSERT INTO t VALUES (1)`); err != nil { |
| t.Fatal(err) |
| } |
| @@ -739,19 +761,178 @@ func TestImportUploadPutFailureAborts(t *testing.T) { |
| f.requireNothingUploaded(t) |
| } |
| -// An object already at the seed's key is an earlier failed attempt for the |
| -// same seed: it is replaced. |
| -func TestImportUploadReplacesALeftover(t *testing.T) { |
| +// The put is create-only: an object already at the seed's key is never |
| +// replaced, and the upload that lost stops without aborting the import. |
| +func TestImportUploadNeverReplacesASnapshot(t *testing.T) { |
| f := newImportFixture(t) |
| body, _ := sqliteBytes(t) |
| begun := f.begun(t, importHost, body) |
| key := seeds.SnapshotKey(importHost, begun.Seed.ID) |
| - if _, err := f.data.Put(context.Background(), key, []byte("half of an earlier attempt"), ""); err != nil { |
| + if _, err := f.data.Put(context.Background(), key, []byte("another writer's snapshot"), ""); err != nil { |
| t.Fatal(err) |
| } |
| + requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAwaitingUpload, "") |
| + if got := f.rawKey(t, f.data, key); got != "another writer's snapshot" { |
| + t.Fatalf("snapshot = %q; want it left as it was", got) |
| + } |
| + f.requireClean(t) |
| +} |
| + |
| +// A WAL database leaves -shm and -wal beside the spool when it is checked; |
| +// the spool's directory goes with them (preflight M5). |
| +func TestImportUploadOfAWALDatabase(t *testing.T) { |
| + f := newImportFixture(t) |
| + body, _ := sqliteBytesIn(t, "WAL") |
| + begun := f.begun(t, importHost, body) |
| + var sidecars []string |
| + old := checkSeedDBConsole |
| + defer func() { checkSeedDBConsole = old }() |
| + checkSeedDBConsole = func(ctx context.Context, path string) error { |
| + err := old(ctx, path) |
| + sidecars, _ = filepath.Glob(path + "-*") |
| + return err |
| + } |
| requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "") |
| - if got, _, err := f.data.Get(context.Background(), key); err != nil || !bytes.Equal(got, body) { |
| - t.Fatalf("snapshot = %q, %v; want the upload", got, err) |
| + if len(sidecars) == 0 { |
| + t.Fatal("the check left no sidecar files; this test no longer covers their removal") |
| + } |
| + f.requireClean(t) |
| +} |
| + |
| +// Two uploads of one import, the second reading the record before the |
| +// first's last write and reserving after it (Important 1): the second is |
| +// refused under its reservation, and the first's work stands. |
| +func TestImportUploadSecondUploadAfterTheFirstLands(t *testing.T) { |
| + f := newImportFixture(t) |
| + body, _ := sqliteBytes(t) |
| + begun := f.begun(t, importHost, body) |
| + puts := &countingPuts{Store: f.data} |
| + f.fx.srv.cfg.SeedData = puts |
| + |
| + firstChecked := make(chan struct{}) |
| + finishFirst := make(chan struct{}) |
| + old := checkSeedDBConsole |
| + defer func() { checkSeedDBConsole = old }() |
| + checkSeedDBConsole = func(ctx context.Context, path string) error { |
| + close(firstChecked) |
| + <-finishFirst |
| + return old(ctx, path) |
| + } |
| + secondRead := make(chan struct{}) |
| + reserveSecond := make(chan struct{}) |
| + var calls atomic.Int32 |
| + defer func() { importReserveHook = nil }() |
| + importReserveHook = func(string) { |
| + if calls.Add(1) == 2 { |
| + close(secondRead) |
| + <-reserveSecond |
| + } |
| + } |
| + |
| + if code, out := uploadImport(t, f, importHost, body); code != http.StatusAccepted { |
| + t.Fatalf("first upload = %d %v", code, out) |
| + } |
| + <-firstChecked |
| + type answer struct { |
| + code int |
| + out map[string]any |
| + err error |
| + } |
| + second := make(chan answer, 1) |
| + go func() { |
| + code, out, err := putImport(f, importHost, body) |
| + second <- answer{code, out, err} |
| + }() |
| + <-secondRead // the second has read awaiting-upload |
| + close(finishFirst) |
| + f.fx.srv.seedCopies.Wait() // the first is uploaded and has released |
| + close(reserveSecond) |
| + got := <-second |
| + if got.err != nil || got.code != http.StatusConflict || |
| + str(t, got.out, "error") != "old.shop.example.com is uploaded. It takes an upload only while it waits for one." { |
| + t.Fatalf("second upload = %d %v %v; want 409 SD-N53", got.code, got.out, got.err) |
| + } |
| + f.fx.srv.seedCopies.Wait() |
| + rec, _ := f.record(t, importHost) |
| + requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "") |
| + if n := puts.n.Load(); n != 1 { |
| + t.Fatalf("%d snapshot puts; want the first upload's one", n) |
| + } |
| + if got, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(importHost, begun.Seed.ID)); err != nil || !bytes.Equal(got, body) { |
| + t.Fatalf("snapshot = %d bytes, %v", len(got), err) |
| + } |
| + f.requireClean(t) |
| +} |
| + |
| +// countingPuts counts the snapshot writes that reach the bucket. |
| +type countingPuts struct { |
| + release.Store |
| + n atomic.Int32 |
| +} |
| + |
| +func (c *countingPuts) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + c.n.Add(1) |
| + return c.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// The import moves on while its upload is checked: the detached half puts |
| +// nothing and aborts nothing, whether its check then fails or passes. |
| +func TestImportUploadLeavesAnImportThatMovedOn(t *testing.T) { |
| + for name, c := range map[string]struct { |
| + moveOn func(t *testing.T, f fromFixture, id string) |
| + checkOK bool |
| + state string // "" for a record that is gone |
| + }{ |
| + "uploaded by another writer, check fails": {func(t *testing.T, f fromFixture, id string) { f.setSeed(t, provision.SeedUploaded(id)) }, false, seeds.StateUploaded}, |
| + "uploaded by another writer, check passes": {func(t *testing.T, f fromFixture, id string) { f.setSeed(t, provision.SeedUploaded(id)) }, true, seeds.StateUploaded}, |
| + "aborted by the user": {func(t *testing.T, f fromFixture, id string) { f.setSeed(t, provision.SeedAbort(id)) }, true, seeds.StateAborted}, |
| + "deleted before its swap": {func(t *testing.T, f fromFixture, id string) { |
| + if err := f.fx.rel.Delete(context.Background(), provision.RecordKey("bab", fromApp, importHost)); err != nil { |
| + t.Fatal(err) |
| + } |
| + }, true, ""}, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + f := newImportFixture(t) |
| + body, _ := sqliteBytes(t) |
| + begun := f.begun(t, importHost, body) |
| + checking := make(chan struct{}) |
| + proceed := make(chan struct{}) |
| + old := checkSeedDBConsole |
| + defer func() { checkSeedDBConsole = old }() |
| + checkSeedDBConsole = func(ctx context.Context, path string) error { |
| + close(checking) |
| + <-proceed |
| + if !c.checkOK { |
| + return errors.New("not a database") |
| + } |
| + return old(ctx, path) |
| + } |
| + if code, out := uploadImport(t, f, importHost, body); code != http.StatusAccepted { |
| + t.Fatalf("upload = %d %v", code, out) |
| + } |
| + <-checking |
| + c.moveOn(t, f, begun.Seed.ID) |
| + close(proceed) |
| + f.fx.srv.seedCopies.Wait() |
| + rec, err := f.record(t, importHost) |
| + if c.state == "" { |
| + if !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("record = %+v, %v; want it still gone", rec, err) |
| + } |
| + } else { |
| + requireImportState(t, rec, begun.Seed.ID, c.state, "") |
| + } |
| + f.requireNothingUploaded(t) |
| + f.requireClean(t) |
| + }) |
| + } |
| +} |
| + |
| +func (f fromFixture) setSeed(t *testing.T, step func(*provision.Record) error) { |
| + t.Helper() |
| + if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, step); err != nil { |
| + t.Fatal(err) |
| } |
| } |
diff --git a/internal/provision/seed.go b/internal/provision/seed.go| index f97abcbc4..f0f82c7ca 100644 |
| --- a/internal/provision/seed.go |
| +++ b/internal/provision/seed.go |
| @@ -216,6 +216,21 @@ func SeedAbortBecause(id, reason string) func(*Record) error { |
| } |
| } |
| +// SeedAbortUploadBecause is the import upload's own failure (spec §5.2, |
| +// Plan 6 Task 2 fix round 1): SeedAbortBecause, but only while the seed |
| +// still waits for its upload. An upload whose import has moved on (another |
| +// upload landed it, or it was aborted or rolled back) gets ErrSeedConflict |
| +// and aborts nothing. |
| +func SeedAbortUploadBecause(id, reason string) func(*Record) error { |
| + abort := SeedAbortBecause(id, reason) |
| + return func(rec *Record) error { |
| + if err := seedIn(rec, id, seeds.KindImport, seeds.StateAwaitingUpload); err != nil { |
| + return err |
| + } |
| + return abort(rec) |
| + } |
| +} |
| + |
| // SeedRolledBack is the box's last word on an aborted seed (spec §5.6, and |
| // §3.2's no-journal path). The console finishes from here. |
| func SeedRolledBack(id string) func(*Record) error { |
diff --git a/internal/provision/seed_test.go b/internal/provision/seed_test.go| index b7ce4f444..c851712e4 100644 |
| --- a/internal/provision/seed_test.go |
| +++ b/internal/provision/seed_test.go |
| @@ -250,6 +250,31 @@ func TestSeedAbortBecauseRecordsItsReason(t *testing.T) { |
| } |
| } |
| +// The upload's abort records why, and refuses an import that has moved |
| +// on from awaiting-upload: it never aborts another upload's work. |
| +func TestSeedAbortUploadOnlyWhileAwaiting(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + seed := testSeed(t, seeds.KindImport, seeds.StateAwaitingUpload) |
| + rec := seedTestRecord(t, s, seed) |
| + got, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbortUploadBecause(seed.ID, "the upload failed its check")) |
| + if err != nil || got.Seed.State != seeds.StateAborted || got.Seed.Reason != "the upload failed its check" { |
| + t.Fatalf("abort upload = %+v, %v", got.Seed, err) |
| + } |
| + s = release.NewDirStore(t.TempDir()) |
| + seed = testSeed(t, seeds.KindImport, seeds.StateAwaitingUpload) |
| + rec = seedTestRecord(t, s, seed) |
| + if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedUploaded(seed.ID)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbortUploadBecause(seed.ID, "late")); !errors.Is(err, ErrSeedConflict) { |
| + t.Fatalf("abort upload of an uploaded import: err = %v, want ErrSeedConflict", err) |
| + } |
| + if after, err := GetRecord(ctx, s, rec.Account, rec.App, rec.Host); err != nil || after.Seed.State != seeds.StateUploaded || after.Seed.Reason != "" { |
| + t.Fatalf("after a refused upload abort = %+v, %v; want it uploaded", after.Seed, err) |
| + } |
| +} |
| + |
| func TestSeedColumn(t *testing.T) { |
| if SeedColumn(Record{}) != "" { |
| t.Error("a seedless record has a column value") |
| @@ -294,6 +319,8 @@ func TestSeedTransitionTable(t *testing.T) { |
| {seeds.KindImport, seeds.StateSwapped}: seeds.StateAborted, |
| {seeds.KindImport, seeds.StatePlaced}: seeds.StateAborted, |
| {seeds.KindImport, seeds.StateRefused}: seeds.StateAborted}}, |
| + "abort upload": {func(id string) func(*Record) error { return SeedAbortUploadBecause(id, "no") }, |
| + map[cell]string{{seeds.KindImport, seeds.StateAwaitingUpload}: seeds.StateAborted}}, |
| "rolled back": {SeedRolledBack, map[cell]string{ |
| {seeds.KindReplica, seeds.StateAborted}: seeds.StateRolledBack, |
| {seeds.KindImport, seeds.StateAborted}: seeds.StateRolledBack}}, |