| index 11189bd8e..83b2aa8cb 100644 |
| --- a/internal/console/apiinstanceimport_test.go |
| +++ b/internal/console/apiinstanceimport_test.go |
| @@ -1035,3 +1035,293 @@ func TestImportUploadRefusals(t *testing.T) { |
| f.requireClean(t) |
| f.requireNothingUploaded(t) |
| } |
| + |
| +// postImport posts verb ("stopped" or "commit") to host's import door. |
| +func postImport(t *testing.T, f fromFixture, host, verb string) (int, map[string]any) { |
| + t.Helper() |
| + code, _, raw := rawReq(t, http.MethodPost, f.url+"/"+host+"/import/"+verb, f.token, "") |
| + var out map[string]any |
| + _ = json.Unmarshal([]byte(raw), &out) |
| + return code, out |
| +} |
| + |
| +// importAudit is the detail of every audit row of action on shop. |
| +func importAudit(t *testing.T, f fromFixture, action string) []string { |
| + t.Helper() |
| + events, err := f.fx.st.Events(OpsAccountID, fromApp, 100) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var rows []string |
| + for _, e := range events { |
| + if e.Action == action { |
| + rows = append(rows, e.Detail) |
| + } |
| + } |
| + return rows |
| +} |
| + |
| +// The whole of spec §5.3 and §5.4's commit: stopped before the upload is |
| +// refused, stopped after it is ready and bumps the serial, commit before |
| +// placed is refused, and commit after placed lands once and answers 200 when |
| +// it is retried. The test plays the box for the swap and the place. |
| +func TestImportStoppedAndCommit(t *testing.T) { |
| + f := newImportFixture(t) |
| + var logs bytes.Buffer |
| + log.SetOutput(&logs) |
| + defer log.SetOutput(os.Stderr) |
| + body, sha := sqliteBytes(t) |
| + code, out := beginImport(t, f, importHost, sha, int64(len(body))) |
| + begun, token := f.requireBegun(t, code, out, importHost) |
| + id := begun.Seed.ID |
| + if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict || |
| + str(t, out, "error") != importHost+" is awaiting-upload. Mark it stopped once, after its upload." { |
| + t.Fatalf("stopped before upload = %d %v", code, out) |
| + } |
| + requireImportState(t, f.uploaded(t, importHost, body), id, seeds.StateUploaded, "") |
| + |
| + before := serial(t, f.fx) |
| + if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusOK || len(out) != 1 || out["state"] != "ready" { |
| + t.Fatalf("stopped = %d %v", code, out) |
| + } |
| + if got := serial(t, f.fx); got != before+1 { |
| + t.Fatalf("serial after stopped = %d, want %d", got, before+1) |
| + } |
| + rec, _ := f.record(t, importHost) |
| + requireImportState(t, rec, id, seeds.StateReady, "") |
| + if rows := importAudit(t, f, "import-stopped"); len(rows) != 1 || rows[0] != "marked the old process of "+importHost+" stopped" { |
| + t.Fatalf("import-stopped rows = %q", rows) |
| + } |
| + // A second stopped is refused: the import is no longer uploaded. |
| + if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict || |
| + str(t, out, "error") != importHost+" is ready. Mark it stopped once, after its upload." { |
| + t.Fatalf("second stopped = %d %v", code, out) |
| + } |
| + |
| + for _, state := range []string{seeds.StateReady, seeds.StateSwapped} { |
| + if state == seeds.StateSwapped { |
| + f.setSeed(t, provision.SeedSwapped(id)) |
| + } |
| + before = serial(t, f.fx) |
| + if code, out := postImport(t, f, importHost, "commit"); code != http.StatusConflict || |
| + str(t, out, "error") != "nothing has been verified yet. Send a request with the verify token first, then commit." { |
| + t.Fatalf("commit while %s = %d %v", state, code, out) |
| + } |
| + if serial(t, f.fx) != before { |
| + t.Fatalf("a refused commit while %s bumped the serial", state) |
| + } |
| + } |
| + f.setSeed(t, provision.SeedPlace(id)) |
| + before = serial(t, f.fx) |
| + if code, out := postImport(t, f, importHost, "commit"); code != http.StatusOK || len(out) != 1 || out["state"] != "committed" { |
| + t.Fatalf("commit = %d %v", code, out) |
| + } |
| + if got := serial(t, f.fx); got != before+1 { |
| + t.Fatalf("serial after commit = %d, want %d", got, before+1) |
| + } |
| + rec, _ = f.record(t, importHost) |
| + requireImportState(t, rec, id, seeds.StateCommitted, "") |
| + // The retried commit changes nothing, audits nothing and bumps nothing. |
| + before = serial(t, f.fx) |
| + if code, out := postImport(t, f, importHost, "commit"); code != http.StatusOK || out["state"] != "committed" { |
| + t.Fatalf("a retried commit = %d %v, want 200", code, out) |
| + } |
| + if serial(t, f.fx) != before { |
| + t.Fatal("a retried commit bumped the serial") |
| + } |
| + if rows := importAudit(t, f, "import-committed"); len(rows) != 1 || rows[0] != "committed the import of "+importHost { |
| + t.Fatalf("import-committed rows = %q", rows) |
| + } |
| + // Stopped after commit is refused with the committed state. |
| + if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict || |
| + str(t, out, "error") != importHost+" is committed. Mark it stopped once, after its upload." { |
| + t.Fatalf("stopped after commit = %d %v", code, out) |
| + } |
| + |
| + // The token never leaves begin's body: not the log, not an audit row. |
| + if strings.Contains(logs.String(), token) { |
| + t.Fatal("the token reached the log") |
| + } |
| + events, _ := f.fx.st.Events(OpsAccountID, fromApp, 100) |
| + for _, e := range events { |
| + if strings.Contains(e.Detail+e.Message, token) { |
| + t.Errorf("the token reached an audit row: %+v", e) |
| + } |
| + } |
| +} |
| + |
| +// Commit of an import that was refused, aborted or rolled back is the |
| +// SD-N115 refusal, and writes nothing. |
| +func TestImportCommitRefusesAnAbandonedImport(t *testing.T) { |
| + f := newImportFixture(t) |
| + for _, state := range []string{seeds.StateRefused, seeds.StateAborted, seeds.StateRolledBack} { |
| + host := state + ".shop.example.com" |
| + f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, state) |
| + before := serial(t, f.fx) |
| + if code, out := postImport(t, f, host, "commit"); code != http.StatusConflict || |
| + str(t, out, "error") != host+" is "+state+", so there is nothing to commit." { |
| + t.Errorf("commit of a %s import = %d %v", state, code, out) |
| + } |
| + if rec, _ := f.record(t, host); rec.Seed.State != state { |
| + t.Errorf("commit moved a %s import to %s", state, rec.Seed.State) |
| + } |
| + if serial(t, f.fx) != before { |
| + t.Errorf("a refused commit of a %s import bumped the serial", state) |
| + } |
| + } |
| + if rows := importAudit(t, f, "import-committed"); len(rows) != 0 { |
| + t.Fatalf("import-committed rows = %q", rows) |
| + } |
| +} |
| + |
| +// Stopped refuses every state but uploaded, without writing the record. |
| +func TestImportStoppedRefusesEveryOtherState(t *testing.T) { |
| + f := newImportFixture(t) |
| + for _, state := range []string{seeds.StateAwaitingUpload, seeds.StateReady, seeds.StateSwapped, seeds.StatePlaced, |
| + seeds.StateCommitted, seeds.StateRefused, seeds.StateAborted, seeds.StateRolledBack} { |
| + host := state + ".shop.example.com" |
| + f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, state) |
| + before := serial(t, f.fx) |
| + if code, out := postImport(t, f, host, "stopped"); code != http.StatusConflict || |
| + str(t, out, "error") != host+" is "+state+". Mark it stopped once, after its upload." { |
| + t.Errorf("stopped while %s = %d %v", state, code, out) |
| + } |
| + if rec, _ := f.record(t, host); rec.Seed.State != state { |
| + t.Errorf("stopped moved a %s import to %s", state, rec.Seed.State) |
| + } |
| + if serial(t, f.fx) != before { |
| + t.Errorf("a refused stopped while %s bumped the serial", state) |
| + } |
| + } |
| +} |
| + |
| +// Both doors share the import gate: no record, a malformed host, an |
| +// ordinary instance and a --from seed are refused as the upload refuses |
| +// them, and a claim console refuses before it reads the record. |
| +func TestImportStoppedAndCommitGate(t *testing.T) { |
| + f := newImportFixture(t) |
| + f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReady) |
| + f.seedRecord(t, importHost, "", newSeedID(t), seeds.KindImport, seeds.StatePlaced) |
| + for _, verb := range []string{"stopped", "commit"} { |
| + for _, c := range []struct { |
| + name, host string |
| + code int |
| + msg string |
| + }{ |
| + {"no record", "ghost.shop.example.com", http.StatusNotFound, "ghost.shop.example.com is not an instance of shop"}, |
| + {"not a host", "Not_A_Host", http.StatusNotFound, "Not_A_Host is not an instance of shop"}, |
| + {"an ordinary instance", fromSource, http.StatusConflict, fromSource + " is not an import."}, |
| + {"a --from seed", fromHost, http.StatusConflict, fromHost + " is not an import."}, |
| + } { |
| + if code, out := postImport(t, f, c.host, verb); code != c.code || str(t, out, "error") != c.msg { |
| + t.Errorf("%s %s = %d %v, want %d %q", verb, c.name, code, out, c.code, c.msg) |
| + } |
| + } |
| + f.fx.srv.cfg.HostClaim = true |
| + if code, out := postImport(t, f, importHost, verb); code != http.StatusConflict || |
| + str(t, out, "error") != "This console runs with claim control on. Seeded instances are not supported here yet." { |
| + t.Errorf("%s with claims armed = %d %v", verb, code, out) |
| + } |
| + f.fx.srv.cfg.HostClaim = false |
| + } |
| + if rec, _ := f.record(t, importHost); rec.Seed.State != seeds.StatePlaced { |
| + t.Fatalf("a refused door moved the import to %s", rec.Seed.State) |
| + } |
| +} |
| + |
| +// Stopped asks the region again: a region that can no longer take the |
| +// import leaves it uploaded. |
| +func TestImportStoppedAsksTheRegionAgain(t *testing.T) { |
| + f := newImportFixture(t) |
| + body, _ := sqliteBytes(t) |
| + begun := f.begun(t, importHost, body) |
| + requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "") |
| + seedReadyBox(t, f.fx, "kass-2", "", 1, "v0.27.0", "v0.27.0", "nested", false) |
| + before := serial(t, f.fx) |
| + if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict || |
| + str(t, out, "error") != "the home region has more than one box. An import needs a region with one box for now." { |
| + t.Fatalf("stopped = %d %v", code, out) |
| + } |
| + rec, _ := f.record(t, importHost) |
| + requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "") |
| + if serial(t, f.fx) != before { |
| + t.Fatal("a refused stopped bumped the serial") |
| + } |
| +} |
| + |
| +// movesOnWrite runs hook, once, just before the first CAS write of key: the |
| +// write that follows loses its CAS, as it would to a concurrent request. |
| +type movesOnWrite struct { |
| + release.Store |
| + key string |
| + hook func() |
| + done atomic.Bool |
| +} |
| + |
| +func (s *movesOnWrite) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if key == s.key && ifMatch != "" && s.done.CompareAndSwap(false, true) { |
| + s.hook() |
| + } |
| + return s.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// A stopped or a commit that read the import in one state and finds it in |
| +// another when it writes changes nothing: SD-N60, no audit row and no bump. |
| +// It never acts on an import that has moved on (its transition is guarded |
| +// on the seed and on its from-state). |
| +func TestImportTransitionsLeaveAnImportThatMovedOn(t *testing.T) { |
| + for _, c := range []struct{ verb, action, start string }{ |
| + {"stopped", "import-stopped", seeds.StateUploaded}, |
| + {"commit", "import-committed", seeds.StatePlaced}, |
| + } { |
| + t.Run(c.verb, func(t *testing.T) { |
| + f := newImportFixture(t) |
| + id := newSeedID(t) |
| + f.seedRecord(t, importHost, "", id, seeds.KindImport, c.start) |
| + f.fx.srv.cfg.Releases = &movesOnWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), hook: func() { |
| + // Another request aborts the import after this one read it. |
| + if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, provision.SeedAbort(id)); err != nil { |
| + t.Error(err) |
| + } |
| + }} |
| + before := serial(t, f.fx) |
| + code, out := postImport(t, f, importHost, c.verb) |
| + if code != http.StatusConflict || str(t, out, "error") != |
| + importHost+" changed while this request was made. Read its state with carlos instances list and try again." { |
| + t.Fatalf("%s = %d %v", c.verb, code, out) |
| + } |
| + rec, _ := f.record(t, importHost) |
| + requireImportState(t, rec, id, seeds.StateAborted, "") |
| + if serial(t, f.fx) != before { |
| + t.Fatal("a lost transition bumped the serial") |
| + } |
| + if rows := importAudit(t, f, c.action); len(rows) != 0 { |
| + t.Fatalf("%s rows = %q", c.action, rows) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A record write that fails for any other reason is the retry sentence, and |
| +// the record is left as it was. |
| +func TestImportTransitionsWriteFailure(t *testing.T) { |
| + f := newImportFixture(t) |
| + id := newSeedID(t) |
| + f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StatePlaced) |
| + rel := &failOnceOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost)} |
| + f.fx.srv.cfg.Releases = rel |
| + rel.armed.Store(true) |
| + before := serial(t, f.fx) |
| + if code, out := postImport(t, f, importHost, "commit"); code != http.StatusBadGateway || str(t, out, "error") != importFailed { |
| + t.Fatalf("commit = %d %v", code, out) |
| + } |
| + rec, _ := f.record(t, importHost) |
| + requireImportState(t, rec, id, seeds.StatePlaced, "") |
| + if serial(t, f.fx) != before { |
| + t.Fatal("a failed commit bumped the serial") |
| + } |
| + // The retry lands. |
| + if code, out := postImport(t, f, importHost, "commit"); code != http.StatusOK || out["state"] != "committed" { |
| + t.Fatalf("retried commit = %d %v", code, out) |
| + } |
| +} |
amadan