carlos / platform Public

Download this file

1	package console
2
3	import (
4	"bytes"
5	"context"
6	"crypto/sha256"
7	"database/sql"
8	"encoding/hex"
9	"encoding/json"
10	"errors"
11	"log"
12	"net/http"
13	"net/url"
14	"os"
15	"path/filepath"
16	"strconv"
17	"strings"
18	"sync"
19	"sync/atomic"
20	"testing"
21	"time"
22
23	"github.com/carlosframework/platform/internal/objstore"
24	"github.com/carlosframework/platform/internal/provision"
25	"github.com/carlosframework/platform/internal/release"
26	"github.com/carlosframework/platform/internal/seeds"
27	)
28
29	const (
30	importHost = "old.shop.example.com"
31	importAddr = "127.0.0.1:8401"
32	)
33
34	var importSHA = strings.Repeat("a", 64)
35
36	// newImportFixture is Plan 5's --from fixture: the app shop, a ready
37	// one-box home region, a seeds floor, a scratch pool and a data bucket.
38	func newImportFixture(t *testing.T) fromFixture {
39	t.Helper()
40	return newFromFixture(t, 0)
41	}
42
43	func importBody(host, sha string, n int64, replaces string) string {
44	body, _ := json.Marshal(map[string]any{"host": host, "import": map[string]any{"sha256": sha, "bytes": n, "replaces": replaces}})
45	return string(body)
46	}
47
48	func beginImport(t *testing.T, f fromFixture, host, sha string, n int64) (int, map[string]any) {
49	t.Helper()
50	return f.create(t, importBody(host, sha, n, importAddr))
51	}
52
53	// requireBegun: a 201 holding a 43-character token, and an awaiting-upload
54	// import record that keeps only the token's hash.
55	func (f fromFixture) requireBegun(t *testing.T, code int, out map[string]any, host string) (*provision.Record, string) {
56	t.Helper()
57	if code != http.StatusCreated {
58	t.Fatalf("begin = %d %v", code, out)
59	}
60	token := str(t, out, "verify_token")
61	if len(token) != 43 {
62	t.Fatalf("verify token %q", token)
63	}
64	rec, err := f.record(t, host)
65	if err != nil || rec.Seed == nil || rec.Seed.Kind != seeds.KindImport || rec.Seed.State != seeds.StateAwaitingUpload ||
66	!seeds.ValidID(rec.Seed.ID) || rec.Seed.VerifyHash != seeds.HashToken(token) || !seeds.TokenMatches(rec.Seed.VerifyHash, token) {
67	t.Fatalf("record = %+v, %v", rec, err)
68	}
69	return rec, token
70	}
71
72	func TestImportBegin(t *testing.T) {
73	f := newImportFixture(t)
74	var logs bytes.Buffer
75	log.SetOutput(&logs)
76	defer log.SetOutput(os.Stderr)
77	code, out := f.create(t, `{"host":"old.shop.example.com","message":"moving the shop","import":{"sha256":"`+importSHA+`","bytes":4096,"replaces":"127.0.0.1:8401"}}`)
78	rec, token := f.requireBegun(t, code, out, importHost)
79	if rec.Seed.SHA256 != importSHA || rec.Seed.Bytes != 4096 || rec.Seed.Replaces != importAddr || rec.Seed.By != "alice@keymail.dev" ||
80	rec.Seed.From != "" || rec.Seed.TXID != "" || rec.Seed.Reason != "" || rec.Seed.At.IsZero() {
81	t.Fatalf("record seed = %+v", rec.Seed)
82	}
83	if rec.Environment != "" || rec.Slug != "old" || rec.Channel != "stable" || rec.Region != "" {
84	t.Fatalf("record = %+v; want the ordinary create's fields", rec)
85	}
86	seed, _ := out["seed"].(map[string]any)
87	if out["host"] != importHost || seed == nil || seed["id"] != rec.Seed.ID || seed["state"] != seeds.StateAwaitingUpload {
88	t.Fatalf("201 body = %v", out)
89	}
90	if _, ok := out["private_store_bucket"]; ok {
91	t.Fatalf("201 body names a store on an app without private storage: %v", out)
92	}
93
94	// The token is answered once, here, and kept nowhere: not in the
95	// bucket, not in the log, not in the audit row.
96	raw, _, err := f.fx.rel.Get(context.Background(), provision.RecordKey("bab", fromApp, importHost))
97	if err != nil || strings.Contains(string(raw), token) {
98	t.Fatalf("the token itself reached the bucket (%v)", err)
99	}
100	if strings.Contains(logs.String(), token) {
101	t.Fatal("the token reached the log")
102	}
103	events, err := f.fx.st.Events(OpsAccountID, fromApp, 50)
104	if err != nil {
105	t.Fatal(err)
106	}
107	var rows []string
108	for _, e := range events {
109	if strings.Contains(e.Detail+e.Message, token) {
110	t.Errorf("the token reached an audit row: %+v", e)
111	}
112	switch e.Action {
113	case "instance-seed":
114	rows = append(rows, e.Detail+" | "+e.Message+" | "+e.Address)
115	case "instance-create":
116	t.Errorf("an import also wrote an instance-create row: %+v", e)
117	}
118	}
119	want := "began an import of old.shop.example.com over 127.0.0.1:8401 (aaaaaaaaaaaa, 4096 bytes) | moving the shop | alice@keymail.dev"
120	if len(rows) != 1 || rows[0] != want {
121	t.Fatalf("instance-seed rows = %q, want [%q]", rows, want)
122	}
123
124	// A second begin for the same host: the record holds it.
125	code, out = beginImport(t, f, importHost, importSHA, 4096)
126	if code != http.StatusConflict || str(t, out, "error") != "instance old.shop.example.com already exists" {
127	t.Fatalf("second begin = %d %v, want 409", code, out)
128	}
129	if again, err := f.record(t, importHost); err != nil || again.Seed.ID != rec.Seed.ID || again.Seed.VerifyHash != rec.Seed.VerifyHash {
130	t.Fatalf("the second begin changed the record: %+v, %v", again, err)
131	}
132	}
133
134	// Each begin mints its own seed id and token.
135	func TestImportBeginMintsFreshTokens(t *testing.T) {
136	f := newImportFixture(t)
137	code, out := beginImport(t, f, importHost, importSHA, 10)
138	a, ta := f.requireBegun(t, code, out, importHost)
139	code, out = beginImport(t, f, "new.shop.example.com", importSHA, 10)
140	b, tb := f.requireBegun(t, code, out, "new.shop.example.com")
141	if a.Seed.ID == b.Seed.ID || ta == tb || a.Seed.VerifyHash == b.Seed.VerifyHash {
142	t.Fatalf("two begins share a seed or a token: %+v %+v", a.Seed, b.Seed)
143	}
144	}
145
146	func TestImportBeginRefusals(t *testing.T) {
147	f := newImportFixture(t)
148	const ordinary = "a seeded instance is an ordinary instance. Drop --kind, --addr and --role."
149	const oneDoor = "a create takes from or import, not both."
150	const size = "bytes must be more than 0 and at most 4 GiB"
151	const hash = "sha256 must be 64 lowercase hex characters"
152	const replaces = "replaces must name the address the import takes over, as host:port"
153	for _, c := range []struct {
154	body string
155	code int
156	want string
157	}{
158	{importBody("a.shop.example.com", importSHA, 0, "127.0.0.1:1"), http.StatusBadRequest, size},
159	{importBody("a.shop.example.com", importSHA, -1, "127.0.0.1:1"), http.StatusBadRequest, size},
160	{importBody("a.shop.example.com", importSHA, 4294967297, "127.0.0.1:1"), http.StatusRequestEntityTooLarge, size},
161	{importBody("a.shop.example.com", "ABC", 10, "127.0.0.1:1"), http.StatusBadRequest, hash},
162	{importBody("a.shop.example.com", strings.Repeat("A", 64), 10, "127.0.0.1:1"), http.StatusBadRequest, hash},
163	{importBody("a.shop.example.com", strings.Repeat("a", 63), 10, "127.0.0.1:1"), http.StatusBadRequest, hash},
164	{importBody("a.shop.example.com", strings.Repeat("g", 64), 10, "127.0.0.1:1"), http.StatusBadRequest, hash},
165	{importBody("a.shop.example.com", importSHA, 10, "nope"), http.StatusBadRequest, replaces},
166	{importBody("a.shop.example.com", importSHA, 10, ":8401"), http.StatusBadRequest, replaces},
167	{importBody("a.shop.example.com", importSHA, 10, "127.0.0.1:http"), http.StatusBadRequest, replaces},
168	{importBody("a.shop.example.com", importSHA, 10, "127.0.0.1:0"), http.StatusBadRequest, replaces},
169	{importBody("a.shop.example.com", importSHA, 10, "127.0.0.1:65536"), http.StatusBadRequest, replaces},
170	{importBody("a.shop.example.com", importSHA, 10, ""), http.StatusBadRequest, replaces},
171	{`{"host":"a.shop.example.com","kind":"service","import":{"sha256":"` + importSHA + `","bytes":10,"replaces":"127.0.0.1:1"}}`, http.StatusBadRequest, ordinary},
172	{`{"host":"a.shop.example.com","addrs":["127.0.0.1:9"],"import":{"sha256":"` + importSHA + `","bytes":10,"replaces":"127.0.0.1:1"}}`, http.StatusBadRequest, ordinary},
173	// One seed door per create.
174	{`{"host":"a.shop.example.com","from":{"host":"main.shop.example.com"},"import":{"sha256":"` + importSHA + `","bytes":10,"replaces":"127.0.0.1:1"}}`, http.StatusBadRequest, oneDoor},
175	} {
176	if code, out := f.create(t, c.body); code != c.code || str(t, out, "error") != c.want {
177	t.Errorf("%s = %d %v, want %d %q", c.body, code, out, c.code, c.want)
178	}
179	f.requireNoRecord(t, "a.shop.example.com")
180	}
181	// An addr with a host name and the largest size are taken.
182	code, out := f.create(t, importBody(importHost, importSHA, 4294967296, "legacy.internal:8401"))
183	f.requireBegun(t, code, out, importHost)
184	}
185
186	func TestImportBeginNeedsOneBox(t *testing.T) {
187	f := newImportFixture(t)
188	seedReadyBox(t, f.fx, "kass-2", "", 1, "v0.27.0", "v0.27.0", "nested", false)
189	code, out := beginImport(t, f, importHost, importSHA, 10)
190	if 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." {
191	t.Fatalf("= %d %v", code, out)
192	}
193	f.requireNoRecord(t, importHost)
194	}
195
196	func TestImportBeginRefusedBeforeTheFloor(t *testing.T) {
197	f := newImportFixture(t)
198	seedReadyBox(t, f.fx, "kass-1", "", 0, "v0.27.0", "v0.27.0", "nested", false)
199	code, out := beginImport(t, f, importHost, importSHA, 10)
200	if code != http.StatusConflict || !strings.Contains(str(t, out, "error"), "below the seeds floor") {
201	t.Fatalf("= %d %v", code, out)
202	}
203	f.requireNoRecord(t, importHost)
204	}
205
206	func TestImportBeginRefusedWithClaimsArmed(t *testing.T) {
207	f := newImportFixture(t)
208	f.fx.srv.cfg.HostClaim = true
209	code, out := beginImport(t, f, importHost, importSHA, 10)
210	if code != http.StatusConflict || str(t, out, "error") != "This console runs with claim control on. Seeded instances are not supported here yet." {
211	t.Fatalf("= %d %v", code, out)
212	}
213	f.requireNoRecord(t, importHost)
214	}
215
216	// The import is homed in this console's own region: its snapshot goes to
217	// this region's data bucket.
218	func TestImportBeginRefusesAnotherRegion(t *testing.T) {
219	f := newImportFixture(t)
220	seedReadyBox(t, f.fx, "far-1", "us", 1, "v0.27.0", "v0.27.0", "nested", false)
221	code, out := f.create(t, `{"host":"old.shop.example.com","region":"us","import":{"sha256":"`+importSHA+`","bytes":10,"replaces":"127.0.0.1:8401"}}`)
222	if code != http.StatusConflict || str(t, out, "error") != "an import runs in this console's region, the home region. To run old.shop.example.com elsewhere, import it here, commit it, and then run carlos move." {
223	t.Fatalf("= %d %v", code, out)
224	}
225	f.requireNoRecord(t, importHost)
226	}
227
228	func TestImportBeginNeedsADataBucket(t *testing.T) {
229	f := newImportFixture(t)
230	f.fx.srv.cfg.SeedData = nil
231	code, out := beginImport(t, f, importHost, importSHA, 10)
232	if code != http.StatusServiceUnavailable || str(t, out, "error") != "this console has no data bucket set up, so it cannot create a seeded instance." {
233	t.Fatalf("= %d %v", code, out)
234	}
235	f.requireNoRecord(t, importHost)
236	}
237
238	func TestImportBeginRefusesAPooledApp(t *testing.T) {
239	f := newImportFixture(t)
240	m := fromManifest(false)
241	m.Pool = "shared"
242	enableInstances(t, f.fx, fromApp, m)
243	code, out := beginImport(t, f, importHost, importSHA, 10)
244	if code != http.StatusConflict || str(t, out, "error") != "shop places its instances in a pool or on a fleet, so it cannot take a seeded instance." {
245	t.Fatalf("= %d %v", code, out)
246	}
247	f.requireNoRecord(t, importHost)
248	}
249
250	// A failed record write answers SD-N67 and leaves the host free.
251	func TestImportBeginRecordWriteFails(t *testing.T) {
252	f := newImportFixture(t)
253	fail := &failOnceOn{Store: f.fx.srv.cfg.Releases, key: provision.RecordKey("bab", fromApp, importHost)}
254	fail.armed.Store(true)
255	f.fx.srv.cfg.Releases = fail
256	code, out := beginImport(t, f, importHost, importSHA, 10)
257	if code != http.StatusBadGateway || str(t, out, "error") != "could not store the upload. Try again." {
258	t.Fatalf("= %d %v", code, out)
259	}
260	f.requireNoRecord(t, importHost)
261	code, out = beginImport(t, f, importHost, importSHA, 10)
262	f.requireBegun(t, code, out, importHost)
263	}
264
265	// With private storage the import gets the derived environment and a new
266	// store marked with its seed, named on the 201 as the ordinary create does.
267	func TestImportBeginPrivateStorage(t *testing.T) {
268	f := newImportFixture(t)
269	enableInstances(t, f.fx, fromApp, fromManifest(true))
270	code, out := f.create(t, `{"host":"old.shop.example.com","environment":"main","import":{"sha256":"`+importSHA+`","bytes":10,"replaces":"127.0.0.1:8401"}}`)
271	if code != http.StatusConflict || str(t, out, "error") != "a seeded instance gets its own environment and an empty store" {
272	t.Fatalf("explicit environment = %d %v", code, out)
273	}
274	f.requireNoRecord(t, importHost)
275
276	code, out = beginImport(t, f, importHost, importSHA, 10)
277	rec, _ := f.requireBegun(t, code, out, importHost)
278	if rec.Environment != "old" {
279	t.Fatalf("record environment = %q, want the host's own", rec.Environment)
280	}
281	store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "old")
282	if err != nil || store == nil || store.Declared == nil || store.Declared.Seed != rec.Seed.ID || storeBucketName(store) == "" {
283	t.Fatalf("old store = %+v, %v; want one marked with seed %s", store, err, rec.Seed.ID)
284	}
285	if str(t, out, "private_store_bucket") != storeBucketName(store) {
286	t.Fatalf("201 private_store_bucket = %v, want %s", out["private_store_bucket"], storeBucketName(store))
287	}
288	}
289
290	// An existing store for the derived environment is refused before the
291	// record is written.
292	func TestImportBeginRefusesATakenStore(t *testing.T) {
293	f := newImportFixture(t)
294	enableInstances(t, f.fx, fromApp, fromManifest(true))
295	foreign := grantedStore("carlos-bab-shop-old")
296	if _, err := objstore.PutIn(context.Background(), f.fx.rel, "bab", fromApp, "old", foreign, ""); err != nil {
297	t.Fatal(err)
298	}
299	var seen *provision.Record
300	f.fx.srv.cfg.Releases = &recordSpy{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), seen: &seen}
301	code, out := beginImport(t, f, importHost, importSHA, 10)
302	if seen != nil {
303	t.Fatalf("a record was written before the refusal: %+v", seen)
304	}
305	if code != http.StatusConflict || str(t, out, "error") != "environment old already has a store (carlos-bab-shop-old). A seeded instance needs a new, empty one." {
306	t.Fatalf("= %d %v", code, out)
307	}
308	f.requireNoRecord(t, importHost)
309	}
310
311	// The store gates that write nothing refuse before the record is written
312	// (ruling R3, preflight M2).
313	func TestImportBeginStoreGatesRefuseBeforeTheRecord(t *testing.T) {
314	f := newImportFixture(t)
315	enableInstances(t, f.fx, fromApp, fromManifest(true))
316	f.fx.srv.cfg.SecretsPub = ""
317	var seen *provision.Record
318	f.fx.srv.cfg.Releases = &recordSpy{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), seen: &seen}
319	code, out := beginImport(t, f, importHost, importSHA, 10)
320	if code != http.StatusServiceUnavailable || str(t, out, "error") != "this console has no secrets key (CARLOS_SECRETS_PUBKEY) — it cannot deliver a store credential, so it will not mint one" {
321	t.Fatalf("= %d %v", code, out)
322	}
323	if seen != nil {
324	t.Fatalf("a record was written before the refusal: %+v", seen)
325	}
326	f.requireNoRecord(t, importHost)
327	}
328
329	func grantedStore(bucket string) *objstore.Record {
330	at := time.Now().UTC().Add(-time.Hour)
331	return &objstore.Record{
332	Declared: &objstore.Declared{EnvPrefix: "S3", At: at, By: "bob@keymail.dev"},
333	Grant: &objstore.GrantBlock{Backend: "mem", Bucket: bucket, Source: objstore.GrantSourceAccountApproval, At: at, By: "operator@keymail.dev"},
334	}
335	}
336
337	// A store walk refused after the record was written rolls the import back
338	// and finishes it before answering, so the same begin can run again at once.
339	func TestImportBeginStoreWalkRefusalFreesTheHost(t *testing.T) {
340	f := newImportFixture(t)
341	enableInstances(t, f.fx, fromApp, fromManifest(true))
342	// Another environment's store already uses the name the walk derives.
343	if _, err := objstore.PutIn(context.Background(), f.fx.rel, "bab", fromApp, "staging", grantedStore("carlos-bab-shop-old"), ""); err != nil {
344	t.Fatal(err)
345	}
346	code, out := beginImport(t, f, importHost, importSHA, 10)
347	if code != http.StatusConflict || str(t, out, "error") != "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" {
348	t.Fatalf("= %d %v", code, out)
349	}
350	f.requireNoRecord(t, importHost)
351	if err := f.fx.rel.Delete(context.Background(), objstore.KeyIn("bab", fromApp, "staging")); err != nil {
352	t.Fatal(err)
353	}
354	code, out = beginImport(t, f, importHost, importSHA, 10)
355	f.requireBegun(t, code, out, importHost)
356	}
357
358	// storeOnRecordWrite writes a foreign, granted store for env right after
359	// the record's create lands: the store another writer made between the
360	// door's check and its walk.
361	type storeOnRecordWrite struct {
362	release.Store
363	key, env string
364	once sync.Once
365	}
366
367	func (s *storeOnRecordWrite) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
368	tag, err := s.Store.Put(ctx, key, body, ifMatch)
369	if err == nil && key == s.key && ifMatch == "" {
370	s.once.Do(func() {
371	if _, perr := objstore.PutIn(ctx, s.Store, "bab", fromApp, s.env, grantedStore("carlos-bab-shop-"+s.env), ""); perr != nil {
372	panic(perr)
373	}
374	})
375	}
376	return tag, err
377	}
378
379	func (s *storeOnRecordWrite) DeleteIfMatch(ctx context.Context, key, etag string) error {
380	return release.DeleteIfMatch(ctx, s.Store, key, etag)
381	}
382
383	// The walk's own re-check under the store lock (errSeedStoreTaken) answers
384	// SD-N34, rolls the import back and frees the host; the other writer's
385	// store is left as it is.
386	func TestImportBeginStoreTakenUnderTheLock(t *testing.T) {
387	f := newImportFixture(t)
388	enableInstances(t, f.fx, fromApp, fromManifest(true))
389	f.fx.srv.cfg.Releases = &storeOnRecordWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), env: "old"}
390	code, out := beginImport(t, f, importHost, importSHA, 10)
391	if code != http.StatusConflict || str(t, out, "error") != "environment old already has a store (carlos-bab-shop-old). A seeded instance needs a new, empty one." {
392	t.Fatalf("= %d %v", code, out)
393	}
394	f.requireNoRecord(t, importHost)
395	store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "old")
396	if err != nil || store == nil || store.Grant == nil || store.Grant.Bucket != "carlos-bab-shop-old" {
397	t.Fatalf("the other writer's store = %+v, %v; want it kept", store, err)
398	}
399	}
400
401	// apexOnBucket creates the app label as the store walk makes its bucket:
402	// a sibling app taking the host's platform apex mid-begin. It records the
403	// seed the store was declared for, as the walk sees it.
404	type apexOnBucket struct {
405	*objstore.MemProvisioner
406	f fromFixture
407	label string
408	declared string
409	}
410
411	func (p *apexOnBucket) EnsureBucket(ctx context.Context, name string) (string, error) {
412	if store, _, err := objstore.GetIn(ctx, p.f.fx.rel, "bab", fromApp, p.label); err == nil && store != nil && store.Declared != nil {
413	p.declared = store.Declared.Seed
414	}
415	if _, err := p.f.fx.st.CreateAppIn(OpsAccountID, p.label, "paul@keymail.dev", time.Now()); err != nil {
416	return "", err
417	}
418	return p.MemProvisioner.EnsureBucket(ctx, name)
419	}
420
421	// Preflight m11 and M3: a begin on a private-storage app whose host is
422	// claimed as a sibling app's apex after the record was written. The store
423	// was made for this seed; the begin rolls the import back, the finish
424	// tears that store down and deletes the record, and the create door's own
425	// sentence is answered.
426	func TestImportBeginSiblingApexTearsDownTheStore(t *testing.T) {
427	f := newImportFixture(t)
428	f.fx.srv.cfg.AppsDomain = "apps.example.net"
429	m := fromManifest(true)
430	m.Domain = "bab.apps.example.net"
431	enableInstances(t, f.fx, fromApp, m)
432	p := &apexOnBucket{MemProvisioner: f.fake, f: f, label: "old"}
433	f.fx.srv.cfg.StoreProvisioner = p
434	var seen *provision.Record
435	f.fx.srv.cfg.Releases = &recordSpy{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, "old.bab.apps.example.net"), seen: &seen}
436
437	code, out := beginImport(t, f, "old.bab.apps.example.net", importSHA, 10)
438	if code != http.StatusConflict || str(t, out, "error") != "old.bab.apps.example.net is app old's platform apex. That app was created while this instance was being written, so the instance was not kept." {
439	t.Fatalf("= %d %v", code, out)
440	}
441	if seen == nil || seen.Seed == nil || !seeds.ValidID(seen.Seed.ID) || p.declared != seen.Seed.ID {
442	t.Fatalf("the store was declared for seed %q; the record written was %+v", p.declared, seen)
443	}
444	f.requireNoRecord(t, "old.bab.apps.example.net")
445	if store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "old"); err != nil || store != nil {
446	t.Fatalf("the seed's store record = %+v, %v; want it torn down", store, err)
447	}
448	if _, ok := f.fake.Buckets["carlos-bab-shop-old"]; ok {
449	t.Fatal("the seed's bucket survived the finish")
450	}
451	if len(f.fake.Buckets) != 0 {
452	t.Fatalf("buckets left = %v", f.fake.Buckets)
453	}
454	}
455
456	// recordSpy keeps the first create-only write of key's body.
457	type recordSpy struct {
458	release.Store
459	key string
460	seen **provision.Record
461	}
462
463	func (s *recordSpy) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
464	if key == s.key && ifMatch == "" && *s.seen == nil {
465	var rec provision.Record
466	if err := json.Unmarshal(body, &rec); err == nil {
467	*s.seen = &rec
468	}
469	}
470	return s.Store.Put(ctx, key, body, ifMatch)
471	}
472
473	func (s *recordSpy) DeleteIfMatch(ctx context.Context, key, etag string) error {
474	return release.DeleteIfMatch(ctx, s.Store, key, etag)
475	}
476
477	// Begin uploads nothing: the snapshot is the upload's (Task 2).
478	func TestImportBeginUploadsNothing(t *testing.T) {
479	f := newImportFixture(t)
480	code, out := beginImport(t, f, importHost, importSHA, 10)
481	f.requireBegun(t, code, out, importHost)
482	f.requireNothingUploaded(t)
483	f.requireClean(t)
484	}
485
486	// ---- Upload (spec §5.2, preflight M5 to M8) ----
487
488	const (
489	importMismatch = "the upload does not match the length and sha256 this import began with, so it was not kept."
490	importNotADB = "the upload is not a database that passes an integrity check, so it was not kept."
491	importSlow = "checking the upload took longer than 30 minutes, so it was not kept."
492	)
493
494	// uploadImport puts body to host's import door and returns what it
495	// answered.
496	func uploadImport(t *testing.T, f fromFixture, host string, body []byte) (int, map[string]any) {
497	t.Helper()
498	code, out, err := putImport(f, host, body)
499	if err != nil {
500	t.Fatal(err)
501	}
502	return code, out
503	}
504
505	// putImport is uploadImport for a goroutine: it returns its error. It
506	// names the seed host's record holds as the request is made, as the
507	// standalone verbs do (final review F1).
508	func putImport(f fromFixture, host string, body []byte) (int, map[string]any, error) {
509	return putImportFor(f, host, heldSeed(f, host), body)
510	}
511
512	// heldSeed is the seed id host's record holds now, or "" with no record or
513	// no seed. It never fails a test, so goroutines may call it.
514	func heldSeed(f fromFixture, host string) string {
515	rec, err := provision.GetRecord(context.Background(), f.fx.rel, "bab", fromApp, host)
516	if err != nil || rec.Seed == nil {
517	return ""
518	}
519	return rec.Seed.ID
520	}
521
522	// importDoorURL is host's import door (verb "" for the upload), naming id.
523	func importDoorURL(f fromFixture, host, verb, id string) string {
524	u := f.url + "/" + host + "/import"
525	if verb != "" {
526	u += "/" + verb
527	}
528	return u + "?seed=" + url.QueryEscape(id)
529	}
530
531	// putImportFor is putImport naming seed id.
532	func putImportFor(f fromFixture, host, id string, body []byte) (int, map[string]any, error) {
533	req, err := http.NewRequest(http.MethodPut, importDoorURL(f, host, "", id), bytes.NewReader(body))
534	if err != nil {
535	return 0, nil, err
536	}
537	req.Header.Set("Authorization", "Bearer "+f.token)
538	res, err := http.DefaultClient.Do(req)
539	if err != nil {
540	return 0, nil, err
541	}
542	defer res.Body.Close()
543	var out map[string]any
544	_ = json.NewDecoder(res.Body).Decode(&out)
545	return res.StatusCode, out, nil
546	}
547
548	// sqliteBytes is a small, sound SQLite database and its sha256.
549	func sqliteBytes(t *testing.T) ([]byte, string) {
550	t.Helper()
551	return sqliteBytesIn(t, "")
552	}
553
554	// sqliteBytesIn is sqliteBytes with a journal mode set first ("" keeps
555	// the default).
556	func sqliteBytesIn(t *testing.T, journal string) ([]byte, string) {
557	t.Helper()
558	p := filepath.Join(t.TempDir(), "snap.db")
559	db, err := sql.Open("sqlite", p)
560	if err != nil {
561	t.Fatal(err)
562	}
563	if journal != "" {
564	if _, err := db.Exec(`PRAGMA journal_mode=` + journal); err != nil {
565	t.Fatal(err)
566	}
567	}
568	if _, err := db.Exec(`CREATE TABLE t (x INTEGER); INSERT INTO t VALUES (1)`); err != nil {
569	t.Fatal(err)
570	}
571	db.Close()
572	b, err := os.ReadFile(p)
573	if err != nil {
574	t.Fatal(err)
575	}
576	return b, sha256Of(b)
577	}
578
579	func sha256Of(b []byte) string {
580	sum := sha256.Sum256(b)
581	return hex.EncodeToString(sum[:])
582	}
583
584	// begun begins an import of host for body and returns its record.
585	func (f fromFixture) begun(t *testing.T, host string, body []byte) *provision.Record {
586	t.Helper()
587	code, out := beginImport(t, f, host, sha256Of(body), int64(len(body)))
588	rec, _ := f.requireBegun(t, code, out, host)
589	return rec
590	}
591
592	// uploaded puts body, requires the 202 that hands it to the check, and
593	// waits for the detached half to finish.
594	func (f fromFixture) uploaded(t *testing.T, host string, body []byte) *provision.Record {
595	t.Helper()
596	if code, out := uploadImport(t, f, host, body); code != http.StatusAccepted || len(out) != 1 || out["state"] != "checking" {
597	t.Fatalf("upload = %d %v, want 202 {\"state\":\"checking\"}", code, out)
598	}
599	f.fx.srv.seedCopies.Wait()
600	rec, err := f.record(t, host)
601	if err != nil || rec.Seed == nil {
602	t.Fatalf("record %s after its upload = %+v, %v", host, rec, err)
603	}
604	return rec
605	}
606
607	func requireImportState(t *testing.T, rec *provision.Record, id, state, reason string) {
608	t.Helper()
609	if rec.Seed == nil || rec.Seed.ID != id || rec.Seed.Kind != seeds.KindImport || rec.Seed.State != state || rec.Seed.Reason != reason {
610	t.Fatalf("record %s seed = %+v; want %s %s with reason %q", rec.Host, rec.Seed, id, state, reason)
611	}
612	}
613
614	func TestImportUploadStoresTheSnapshot(t *testing.T) {
615	f := newImportFixture(t)
616	body, _ := sqliteBytes(t)
617	begun := f.begun(t, importHost, body)
618	rec := f.uploaded(t, importHost, body)
619	requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
620	got, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(importHost, begun.Seed.ID))
621	if err != nil || !bytes.Equal(got, body) {
622	t.Fatalf("snapshot = %d bytes, %v; want the upload", len(got), err)
623	}
624	if keys, _ := f.data.List(context.Background(), seeds.Prefix); len(keys) != 1 {
625	t.Fatalf("seeds/ = %v; want the one snapshot", keys)
626	}
627	f.requireClean(t)
628	events, err := f.fx.st.Events(OpsAccountID, fromApp, 50)
629	if err != nil {
630	t.Fatal(err)
631	}
632	var rows []string
633	for _, e := range events {
634	if e.Action == "import-uploaded" {
635	rows = append(rows, e.Detail+" | "+e.Target+" | "+e.Address)
636	}
637	}
638	if want := "uploaded the snapshot for old.shop.example.com | old.shop.example.com | alice@keymail.dev"; len(rows) != 1 || rows[0] != want {
639	t.Fatalf("import-uploaded rows = %q, want [%q]", rows, want)
640	}
641	}
642
643	// The check runs after the answer, detached from the request (preflight
644	// M6): the record waits in awaiting-upload, the spool sits in its own
645	// from- directory (M5), and the reservation holds off a second upload.
646	func TestImportUploadChecksAfterTheAnswer(t *testing.T) {
647	f := newImportFixture(t)
648	body, _ := sqliteBytes(t)
649	begun := f.begun(t, importHost, body)
650	type checking struct {
651	path string
652	deadline time.Duration
653	}
654	entered := make(chan checking, 1)
655	proceed := make(chan struct{})
656	old := checkSeedDBConsole
657	defer func() { checkSeedDBConsole = old }()
658	checkSeedDBConsole = func(ctx context.Context, path string) error {
659	d, _ := ctx.Deadline()
660	entered <- checking{path, time.Until(d)}
661	<-proceed
662	if ctx.Err() != nil {
663	return ctx.Err()
664	}
665	return old(ctx, path)
666	}
667	if code, out := uploadImport(t, f, importHost, body); code != http.StatusAccepted || out["state"] != "checking" {
668	t.Fatalf("upload = %d %v", code, out)
669	}
670	c := <-entered
671	if dir := filepath.Dir(c.path); filepath.Base(c.path) != importHost+".db" || filepath.Dir(dir) != f.fx.srv.seedScratch.dir() ||
672	!strings.HasPrefix(filepath.Base(dir), seedScratchPrefix) {
673	t.Errorf("spool %s; want <scratch>/%s*/%s.db", c.path, seedScratchPrefix, importHost)
674	}
675	if c.deadline < 29*time.Minute || c.deadline > 30*time.Minute {
676	t.Errorf("the check's deadline is %v away, want 30 minutes", c.deadline)
677	}
678	rec, _ := f.record(t, importHost)
679	requireImportState(t, rec, begun.Seed.ID, seeds.StateAwaitingUpload, "")
680	if code, out := uploadImport(t, f, importHost, body); code != http.StatusTooManyRequests ||
681	str(t, out, "error") != "another copy or upload is running for shop. Try again when it finishes." {
682	t.Errorf("a second upload while the first is checked = %d %v, want 429", code, out)
683	}
684	close(proceed)
685	f.fx.srv.seedCopies.Wait()
686	rec, _ = f.record(t, importHost)
687	requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
688	f.requireClean(t)
689	}
690
691	// A length or hash that differs is refused before the answer, and the
692	// import stays awaiting-upload, so the right file can still be sent.
693	func TestImportUploadRefusesLengthAndHash(t *testing.T) {
694	f := newImportFixture(t)
695	body, _ := sqliteBytes(t)
696	begun := f.begun(t, importHost, body)
697	bad := append([]byte{}, body...)
698	bad[len(bad)-1] ^= 0xff
699	for name, b := range map[string][]byte{
700	"one byte long": append(append([]byte{}, body...), 'x'),
701	"one byte short": body[:len(body)-1],
702	"empty": {},
703	"wrong hash": bad,
704	} {
705	if code, out := uploadImport(t, f, importHost, b); code != http.StatusUnprocessableEntity || str(t, out, "error") != importMismatch {
706	t.Errorf("%s = %d %v, want 422", name, code, out)
707	}
708	f.fx.srv.seedCopies.Wait()
709	rec, _ := f.record(t, importHost)
710	requireImportState(t, rec, begun.Seed.ID, seeds.StateAwaitingUpload, "")
711	f.requireClean(t)
712	f.requireNothingUploaded(t)
713	}
714	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
715	}
716
717	// A body far over its declared length is cut at bytes+1: the spool never
718	// holds more than that (preflight M7).
719	func TestImportUploadCutsALongBody(t *testing.T) {
720	f := newImportFixture(t)
721	body, _ := sqliteBytes(t)
722	begun := f.begun(t, importHost, body)
723	var spooled atomic.Int64
724	spooled.Store(-1)
725	defer func() { importSpoolHook = nil }()
726	importSpoolHook = func(path string) {
727	if st, err := os.Stat(path); err == nil {
728	spooled.Store(st.Size())
729	}
730	}
731	long := append(append([]byte{}, body...), bytes.Repeat([]byte{'x'}, 1<<20)...)
732	if code, out := uploadImport(t, f, importHost, long); code != http.StatusUnprocessableEntity || str(t, out, "error") != importMismatch {
733	t.Fatalf("a body 1 MiB long = %d %v, want 422", code, out)
734	}
735	if n := spooled.Load(); n < 0 || n > int64(len(body))+1 {
736	t.Fatalf("the spool held %d bytes; want at most %d", n, len(body)+1)
737	}
738	rec, _ := f.record(t, importHost)
739	requireImportState(t, rec, begun.Seed.ID, seeds.StateAwaitingUpload, "")
740	f.requireClean(t)
741	f.requireNothingUploaded(t)
742	}
743
744	func TestImportUploadOnlyOnce(t *testing.T) {
745	f := newImportFixture(t)
746	body, _ := sqliteBytes(t)
747	f.begun(t, importHost, body)
748	f.uploaded(t, importHost, body)
749	if code, out := uploadImport(t, f, importHost, body); code != http.StatusConflict ||
750	str(t, out, "error") != "old.shop.example.com is uploaded. It takes an upload only while it waits for one." {
751	t.Fatalf("second upload = %d %v", code, out)
752	}
753	f.requireClean(t)
754	}
755
756	// A file of the right length and hash that is not a database is aborted,
757	// saying why (SD-N52 as the seed's reason).
758	func TestImportUploadRefusesANonDatabase(t *testing.T) {
759	f := newImportFixture(t)
760	junk := bytes.Repeat([]byte("j"), 100)
761	begun := f.begun(t, "junk.shop.example.com", junk)
762	requireImportState(t, f.uploaded(t, "junk.shop.example.com", junk), begun.Seed.ID, seeds.StateAborted, importNotADB)
763	f.requireClean(t)
764	f.requireNothingUploaded(t)
765	}
766
767	func TestImportUploadCheckDeadline(t *testing.T) {
768	f := newImportFixture(t)
769	body, _ := sqliteBytes(t)
770	begun := f.begun(t, importHost, body)
771	oldCheck, oldDeadline := checkSeedDBConsole, importCheckDeadline
772	defer func() { checkSeedDBConsole, importCheckDeadline = oldCheck, oldDeadline }()
773	importCheckDeadline = time.Nanosecond
774	checkSeedDBConsole = func(ctx context.Context, path string) error {
775	<-ctx.Done()
776	return ctx.Err()
777	}
778	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importSlow)
779	f.requireClean(t)
780	f.requireNothingUploaded(t)
781	}
782
783	func TestImportUploadPutFailureAborts(t *testing.T) {
784	f := newImportFixture(t)
785	body, _ := sqliteBytes(t)
786	begun := f.begun(t, importHost, body)
787	f.fx.srv.cfg.SeedData = failingPuts{f.data}
788	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importFailed)
789	f.requireClean(t)
790	f.requireNothingUploaded(t)
791	}
792
793	// The put is create-only: an object already at the seed's key is never
794	// replaced, and the upload that lost stops without aborting the import.
795	func TestImportUploadNeverReplacesASnapshot(t *testing.T) {
796	f := newImportFixture(t)
797	body, _ := sqliteBytes(t)
798	begun := f.begun(t, importHost, body)
799	key := seeds.SnapshotKey(importHost, begun.Seed.ID)
800	if _, err := f.data.Put(context.Background(), key, []byte("another writer's snapshot"), ""); err != nil {
801	t.Fatal(err)
802	}
803	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAwaitingUpload, "")
804	if got := f.rawKey(t, f.data, key); got != "another writer's snapshot" {
805	t.Fatalf("snapshot = %q; want it left as it was", got)
806	}
807	f.requireClean(t)
808	}
809
810	// A WAL database leaves -shm and -wal beside the spool when it is checked;
811	// the spool's directory goes with them (preflight M5).
812	func TestImportUploadOfAWALDatabase(t *testing.T) {
813	f := newImportFixture(t)
814	body, _ := sqliteBytesIn(t, "WAL")
815	begun := f.begun(t, importHost, body)
816	var sidecars []string
817	old := checkSeedDBConsole
818	defer func() { checkSeedDBConsole = old }()
819	checkSeedDBConsole = func(ctx context.Context, path string) error {
820	err := old(ctx, path)
821	sidecars, _ = filepath.Glob(path + "-*")
822	return err
823	}
824	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
825	if len(sidecars) == 0 {
826	t.Fatal("the check left no sidecar files; this test no longer covers their removal")
827	}
828	f.requireClean(t)
829	}
830
831	// Two uploads of one import, the second reading the record before the
832	// first's last write and reserving after it (Important 1): the second is
833	// refused under its reservation, and the first's work stands.
834	func TestImportUploadSecondUploadAfterTheFirstLands(t *testing.T) {
835	f := newImportFixture(t)
836	body, _ := sqliteBytes(t)
837	begun := f.begun(t, importHost, body)
838	puts := &countingPuts{Store: f.data}
839	f.fx.srv.cfg.SeedData = puts
840
841	firstChecked := make(chan struct{})
842	finishFirst := make(chan struct{})
843	old := checkSeedDBConsole
844	defer func() { checkSeedDBConsole = old }()
845	checkSeedDBConsole = func(ctx context.Context, path string) error {
846	close(firstChecked)
847	<-finishFirst
848	return old(ctx, path)
849	}
850	secondRead := make(chan struct{})
851	reserveSecond := make(chan struct{})
852	var calls atomic.Int32
853	defer func() { importReserveHook = nil }()
854	importReserveHook = func(string) {
855	if calls.Add(1) == 2 {
856	close(secondRead)
857	<-reserveSecond
858	}
859	}
860
861	if code, out := uploadImport(t, f, importHost, body); code != http.StatusAccepted {
862	t.Fatalf("first upload = %d %v", code, out)
863	}
864	<-firstChecked
865	type answer struct {
866	code int
867	out map[string]any
868	err error
869	}
870	second := make(chan answer, 1)
871	go func() {
872	code, out, err := putImport(f, importHost, body)
873	second <- answer{code, out, err}
874	}()
875	<-secondRead // the second has read awaiting-upload
876	close(finishFirst)
877	f.fx.srv.seedCopies.Wait() // the first is uploaded and has released
878	close(reserveSecond)
879	got := <-second
880	if got.err != nil || got.code != http.StatusConflict ||
881	str(t, got.out, "error") != "old.shop.example.com is uploaded. It takes an upload only while it waits for one." {
882	t.Fatalf("second upload = %d %v %v; want 409 SD-N53", got.code, got.out, got.err)
883	}
884	f.fx.srv.seedCopies.Wait()
885	rec, _ := f.record(t, importHost)
886	requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
887	if n := puts.n.Load(); n != 1 {
888	t.Fatalf("%d snapshot puts; want the first upload's one", n)
889	}
890	if got, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(importHost, begun.Seed.ID)); err != nil || !bytes.Equal(got, body) {
891	t.Fatalf("snapshot = %d bytes, %v", len(got), err)
892	}
893	f.requireClean(t)
894	}
895
896	// countingPuts counts the snapshot writes that reach the bucket.
897	type countingPuts struct {
898	release.Store
899	n atomic.Int32
900	}
901
902	func (c *countingPuts) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
903	c.n.Add(1)
904	return c.Store.Put(ctx, key, body, ifMatch)
905	}
906
907	// The import moves on while its upload is checked: the detached half puts
908	// nothing and aborts nothing, whether its check then fails or passes.
909	func TestImportUploadLeavesAnImportThatMovedOn(t *testing.T) {
910	for name, c := range map[string]struct {
911	moveOn func(t *testing.T, f fromFixture, id string)
912	checkOK bool
913	state string // "" for a record that is gone
914	}{
915	"uploaded by another writer, check fails": {func(t *testing.T, f fromFixture, id string) { f.setSeed(t, provision.SeedUploaded(id)) }, false, seeds.StateUploaded},
916	"uploaded by another writer, check passes": {func(t *testing.T, f fromFixture, id string) { f.setSeed(t, provision.SeedUploaded(id)) }, true, seeds.StateUploaded},
917	"aborted by the user": {func(t *testing.T, f fromFixture, id string) { f.setSeed(t, provision.SeedAbort(id)) }, true, seeds.StateAborted},
918	"deleted before its swap": {func(t *testing.T, f fromFixture, id string) {
919	if err := f.fx.rel.Delete(context.Background(), provision.RecordKey("bab", fromApp, importHost)); err != nil {
920	t.Fatal(err)
921	}
922	}, true, ""},
923	} {
924	t.Run(name, func(t *testing.T) {
925	f := newImportFixture(t)
926	body, _ := sqliteBytes(t)
927	begun := f.begun(t, importHost, body)
928	checking := make(chan struct{})
929	proceed := make(chan struct{})
930	old := checkSeedDBConsole
931	defer func() { checkSeedDBConsole = old }()
932	checkSeedDBConsole = func(ctx context.Context, path string) error {
933	close(checking)
934	<-proceed
935	if !c.checkOK {
936	return errors.New("not a database")
937	}
938	return old(ctx, path)
939	}
940	if code, out := uploadImport(t, f, importHost, body); code != http.StatusAccepted {
941	t.Fatalf("upload = %d %v", code, out)
942	}
943	<-checking
944	c.moveOn(t, f, begun.Seed.ID)
945	close(proceed)
946	f.fx.srv.seedCopies.Wait()
947	rec, err := f.record(t, importHost)
948	if c.state == "" {
949	if !errors.Is(err, os.ErrNotExist) {
950	t.Fatalf("record = %+v, %v; want it still gone", rec, err)
951	}
952	} else {
953	requireImportState(t, rec, begun.Seed.ID, c.state, "")
954	}
955	f.requireNothingUploaded(t)
956	f.requireClean(t)
957	})
958	}
959	}
960
961	func (f fromFixture) setSeed(t *testing.T, step func(*provision.Record) error) {
962	t.Helper()
963	if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, step); err != nil {
964	t.Fatal(err)
965	}
966	}
967
968	// The check panics: the seed is aborted with the retry sentence and the
969	// console keeps serving.
970	func TestImportUploadPanicAborts(t *testing.T) {
971	f := newImportFixture(t)
972	body, _ := sqliteBytes(t)
973	begun := f.begun(t, importHost, body)
974	old := checkSeedDBConsole
975	defer func() { checkSeedDBConsole = old }()
976	checkSeedDBConsole = func(context.Context, string) error { panic("a fault in the check") }
977	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importFailed)
978	f.requireClean(t)
979	f.requireNothingUploaded(t)
980	checkSeedDBConsole = old
981	other := f.begun(t, "new.shop.example.com", body)
982	requireImportState(t, f.uploaded(t, "new.shop.example.com", body), other.Seed.ID, seeds.StateUploaded, "")
983	}
984
985	// The uploaded write fails: the seed is aborted with the retry sentence.
986	func TestImportUploadRecordWriteFailureAborts(t *testing.T) {
987	f := newImportFixture(t)
988	body, _ := sqliteBytes(t)
989	begun := f.begun(t, importHost, body)
990	rel := &failOnceOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost)}
991	f.fx.srv.cfg.Releases = rel
992	old := checkSeedDBConsole
993	defer func() { checkSeedDBConsole = old }()
994	checkSeedDBConsole = func(ctx context.Context, path string) error {
995	rel.armed.Store(true) // the next write of the record is uploaded
996	return old(ctx, path)
997	}
998	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateAborted, importFailed)
999	f.requireClean(t)
1000	}
1001
1002	func TestImportUploadRefusals(t *testing.T) {
1003	f := newImportFixture(t)
1004	body, _ := sqliteBytes(t)
1005	f.begun(t, importHost, body)
1006	for _, c := range []struct {
1007	name, host string
1008	code int
1009	msg string
1010	}{
1011	{"no record", "ghost.shop.example.com", http.StatusNotFound, "ghost.shop.example.com is not an instance of shop"},
1012	{"not a host", "Not_A_Host", http.StatusNotFound, "Not_A_Host is not an instance of shop"},
1013	{"an ordinary instance", fromSource, http.StatusConflict, "main.shop.example.com is not an import."},
1014	} {
1015	code, out := uploadImport(t, f, c.host, body)
1016	if code != c.code || str(t, out, "error") != c.msg {
1017	t.Errorf("%s = %d %v, want %d %q", c.name, code, out, c.code, c.msg)
1018	}
1019	}
1020
1021	// A --from seed is not an import either.
1022	f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReserving)
1023	if code, out := uploadImport(t, f, fromHost, body); code != http.StatusConflict || str(t, out, "error") != fromHost+" is not an import." {
1024	t.Errorf("a --from seed = %d %v", code, out)
1025	}
1026
1027	// A seed id that cannot name a key (preflight M8): nothing is put.
1028	f.seedRecord(t, "bad.shop.example.com", "", "not-a-seed-id", seeds.KindImport, seeds.StateAwaitingUpload)
1029	if code, out := uploadImport(t, f, "bad.shop.example.com", body); code != http.StatusBadGateway || str(t, out, "error") != importFailed {
1030	t.Errorf("a bad seed id = %d %v", code, out)
1031	}
1032	f.requireNothingUploaded(t)
1033
1034	// A claim console refuses before it reads the record.
1035	f.fx.srv.cfg.HostClaim = true
1036	if code, out := uploadImport(t, f, importHost, body); code != http.StatusConflict ||
1037	str(t, out, "error") != "This console runs with claim control on. Seeded instances are not supported here yet." {
1038	t.Errorf("claims armed = %d %v", code, out)
1039	}
1040	f.fx.srv.cfg.HostClaim = false
1041
1042	// No data bucket (SD-N38).
1043	data := f.fx.srv.cfg.SeedData
1044	f.fx.srv.cfg.SeedData = nil
1045	if code, out := uploadImport(t, f, importHost, body); code != http.StatusServiceUnavailable ||
1046	str(t, out, "error") != "this console has no data bucket set up, so it cannot create a seeded instance." {
1047	t.Errorf("no data bucket = %d %v", code, out)
1048	}
1049	f.fx.srv.cfg.SeedData = data
1050
1051	// No room in the scratch (SD-N27).
1052	pool := f.fx.srv.seedScratch
1053	f.fx.srv.seedScratch = newSeedScratch(pool.dir(), func(string) (int64, error) { return 1 << 30, nil })
1054	if code, out := uploadImport(t, f, importHost, body); code != http.StatusInsufficientStorage ||
1055	str(t, out, "error") != "the console has no room for this right now. Try again later." {
1056	t.Errorf("no scratch = %d %v", code, out)
1057	}
1058	f.fx.srv.seedScratch = pool
1059	f.fx.srv.seedCopies.Wait()
1060	rec, _ := f.record(t, importHost)
1061	if rec.Seed.State != seeds.StateAwaitingUpload {
1062	t.Fatalf("a refused upload moved the import to %s", rec.Seed.State)
1063	}
1064	f.requireClean(t)
1065	f.requireNothingUploaded(t)
1066	}
1067
1068	// postImport posts verb ("stopped" or "commit") to host's import door,
1069	// naming the seed host's record holds now.
1070	func postImport(t *testing.T, f fromFixture, host, verb string) (int, map[string]any) {
1071	t.Helper()
1072	return postImportFor(t, f, host, heldSeed(f, host), verb)
1073	}
1074
1075	// postImportFor is postImport naming seed id.
1076	func postImportFor(t *testing.T, f fromFixture, host, id, verb string) (int, map[string]any) {
1077	t.Helper()
1078	code, _, raw := rawReq(t, http.MethodPost, importDoorURL(f, host, verb, id), f.token, "")
1079	var out map[string]any
1080	_ = json.Unmarshal([]byte(raw), &out)
1081	return code, out
1082	}
1083
1084	// importAudit is the detail of every audit row of action on shop.
1085	func importAudit(t *testing.T, f fromFixture, action string) []string {
1086	t.Helper()
1087	events, err := f.fx.st.Events(OpsAccountID, fromApp, 100)
1088	if err != nil {
1089	t.Fatal(err)
1090	}
1091	var rows []string
1092	for _, e := range events {
1093	if e.Action == action {
1094	rows = append(rows, e.Detail)
1095	}
1096	}
1097	return rows
1098	}
1099
1100	// The whole of spec §5.3 and §5.4's commit: stopped before the upload is
1101	// refused, stopped after it is ready and bumps the serial, commit before
1102	// placed is refused, and commit after placed lands once and answers 200 when
1103	// it is retried. The test plays the box for the swap and the place.
1104	func TestImportStoppedAndCommit(t *testing.T) {
1105	f := newImportFixture(t)
1106	var logs bytes.Buffer
1107	log.SetOutput(&logs)
1108	defer log.SetOutput(os.Stderr)
1109	body, sha := sqliteBytes(t)
1110	code, out := beginImport(t, f, importHost, sha, int64(len(body)))
1111	begun, token := f.requireBegun(t, code, out, importHost)
1112	id := begun.Seed.ID
1113	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict ||
1114	str(t, out, "error") != importHost+" is awaiting-upload. Mark it stopped once, after its upload." {
1115	t.Fatalf("stopped before upload = %d %v", code, out)
1116	}
1117	requireImportState(t, f.uploaded(t, importHost, body), id, seeds.StateUploaded, "")
1118
1119	before := serial(t, f.fx)
1120	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusOK || len(out) != 1 || out["state"] != "ready" {
1121	t.Fatalf("stopped = %d %v", code, out)
1122	}
1123	if got := serial(t, f.fx); got != before+1 {
1124	t.Fatalf("serial after stopped = %d, want %d", got, before+1)
1125	}
1126	rec, _ := f.record(t, importHost)
1127	requireImportState(t, rec, id, seeds.StateReady, "")
1128	if rows := importAudit(t, f, "import-stopped"); len(rows) != 1 || rows[0] != "marked the old process of "+importHost+" stopped" {
1129	t.Fatalf("import-stopped rows = %q", rows)
1130	}
1131	// A second stopped is refused: the import is no longer uploaded.
1132	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict ||
1133	str(t, out, "error") != importHost+" is ready. Mark it stopped once, after its upload." {
1134	t.Fatalf("second stopped = %d %v", code, out)
1135	}
1136
1137	for _, state := range []string{seeds.StateReady, seeds.StateSwapped} {
1138	if state == seeds.StateSwapped {
1139	f.setSeed(t, provision.SeedSwapped(id))
1140	}
1141	before = serial(t, f.fx)
1142	if code, out := postImport(t, f, importHost, "commit"); code != http.StatusConflict ||
1143	str(t, out, "error") != "nothing has been verified yet. Send a request with the verify token first, then commit." {
1144	t.Fatalf("commit while %s = %d %v", state, code, out)
1145	}
1146	if serial(t, f.fx) != before {
1147	t.Fatalf("a refused commit while %s bumped the serial", state)
1148	}
1149	}
1150	f.setSeed(t, provision.SeedPlace(id))
1151	before = serial(t, f.fx)
1152	if code, out := postImport(t, f, importHost, "commit"); code != http.StatusOK || len(out) != 1 || out["state"] != "committed" {
1153	t.Fatalf("commit = %d %v", code, out)
1154	}
1155	if got := serial(t, f.fx); got != before+1 {
1156	t.Fatalf("serial after commit = %d, want %d", got, before+1)
1157	}
1158	rec, _ = f.record(t, importHost)
1159	requireImportState(t, rec, id, seeds.StateCommitted, "")
1160	// The retried commit changes nothing, audits nothing and bumps nothing.
1161	before = serial(t, f.fx)
1162	if code, out := postImport(t, f, importHost, "commit"); code != http.StatusOK || out["state"] != "committed" {
1163	t.Fatalf("a retried commit = %d %v, want 200", code, out)
1164	}
1165	if serial(t, f.fx) != before {
1166	t.Fatal("a retried commit bumped the serial")
1167	}
1168	if rows := importAudit(t, f, "import-committed"); len(rows) != 1 || rows[0] != "committed the import of "+importHost {
1169	t.Fatalf("import-committed rows = %q", rows)
1170	}
1171	// Stopped after commit is refused with the committed state.
1172	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict ||
1173	str(t, out, "error") != importHost+" is committed. Mark it stopped once, after its upload." {
1174	t.Fatalf("stopped after commit = %d %v", code, out)
1175	}
1176
1177	// The token never leaves begin's body: not the log, not an audit row.
1178	if strings.Contains(logs.String(), token) {
1179	t.Fatal("the token reached the log")
1180	}
1181	events, _ := f.fx.st.Events(OpsAccountID, fromApp, 100)
1182	for _, e := range events {
1183	if strings.Contains(e.Detail+e.Message, token) {
1184	t.Errorf("the token reached an audit row: %+v", e)
1185	}
1186	}
1187	}
1188
1189	// Commit of an import that was refused, aborted or rolled back is the
1190	// SD-N115 refusal, and writes nothing.
1191	func TestImportCommitRefusesAnAbandonedImport(t *testing.T) {
1192	f := newImportFixture(t)
1193	for _, state := range []string{seeds.StateRefused, seeds.StateAborted, seeds.StateRolledBack} {
1194	host := state + ".shop.example.com"
1195	f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, state)
1196	before := serial(t, f.fx)
1197	if code, out := postImport(t, f, host, "commit"); code != http.StatusConflict ||
1198	str(t, out, "error") != host+" is "+state+", so there is nothing to commit." {
1199	t.Errorf("commit of a %s import = %d %v", state, code, out)
1200	}
1201	if rec, _ := f.record(t, host); rec.Seed.State != state {
1202	t.Errorf("commit moved a %s import to %s", state, rec.Seed.State)
1203	}
1204	if serial(t, f.fx) != before {
1205	t.Errorf("a refused commit of a %s import bumped the serial", state)
1206	}
1207	}
1208	if rows := importAudit(t, f, "import-committed"); len(rows) != 0 {
1209	t.Fatalf("import-committed rows = %q", rows)
1210	}
1211	}
1212
1213	// Stopped refuses every state but uploaded, without writing the record.
1214	func TestImportStoppedRefusesEveryOtherState(t *testing.T) {
1215	f := newImportFixture(t)
1216	for _, state := range []string{seeds.StateAwaitingUpload, seeds.StateReady, seeds.StateSwapped, seeds.StatePlaced,
1217	seeds.StateCommitted, seeds.StateRefused, seeds.StateAborted, seeds.StateRolledBack} {
1218	host := state + ".shop.example.com"
1219	f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, state)
1220	before := serial(t, f.fx)
1221	if code, out := postImport(t, f, host, "stopped"); code != http.StatusConflict ||
1222	str(t, out, "error") != host+" is "+state+". Mark it stopped once, after its upload." {
1223	t.Errorf("stopped while %s = %d %v", state, code, out)
1224	}
1225	if rec, _ := f.record(t, host); rec.Seed.State != state {
1226	t.Errorf("stopped moved a %s import to %s", state, rec.Seed.State)
1227	}
1228	if serial(t, f.fx) != before {
1229	t.Errorf("a refused stopped while %s bumped the serial", state)
1230	}
1231	}
1232	}
1233
1234	// Both doors share the import gate: no record, a malformed host, an
1235	// ordinary instance and a --from seed are refused as the upload refuses
1236	// them, and a claim console refuses before it reads the record.
1237	func TestImportStoppedAndCommitGate(t *testing.T) {
1238	f := newImportFixture(t)
1239	f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReady)
1240	f.seedRecord(t, importHost, "", newSeedID(t), seeds.KindImport, seeds.StatePlaced)
1241	for _, verb := range []string{"stopped", "commit"} {
1242	for _, c := range []struct {
1243	name, host string
1244	code int
1245	msg string
1246	}{
1247	{"no record", "ghost.shop.example.com", http.StatusNotFound, "ghost.shop.example.com is not an instance of shop"},
1248	{"not a host", "Not_A_Host", http.StatusNotFound, "Not_A_Host is not an instance of shop"},
1249	{"an ordinary instance", fromSource, http.StatusConflict, fromSource + " is not an import."},
1250	{"a --from seed", fromHost, http.StatusConflict, fromHost + " is not an import."},
1251	} {
1252	if code, out := postImport(t, f, c.host, verb); code != c.code || str(t, out, "error") != c.msg {
1253	t.Errorf("%s %s = %d %v, want %d %q", verb, c.name, code, out, c.code, c.msg)
1254	}
1255	}
1256	f.fx.srv.cfg.HostClaim = true
1257	if code, out := postImport(t, f, importHost, verb); code != http.StatusConflict ||
1258	str(t, out, "error") != "This console runs with claim control on. Seeded instances are not supported here yet." {
1259	t.Errorf("%s with claims armed = %d %v", verb, code, out)
1260	}
1261	f.fx.srv.cfg.HostClaim = false
1262	}
1263	if rec, _ := f.record(t, importHost); rec.Seed.State != seeds.StatePlaced {
1264	t.Fatalf("a refused door moved the import to %s", rec.Seed.State)
1265	}
1266	}
1267
1268	// Stopped asks the region again: a region that can no longer take the
1269	// import leaves it uploaded.
1270	func TestImportStoppedAsksTheRegionAgain(t *testing.T) {
1271	f := newImportFixture(t)
1272	body, _ := sqliteBytes(t)
1273	begun := f.begun(t, importHost, body)
1274	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
1275	seedReadyBox(t, f.fx, "kass-2", "", 1, "v0.27.0", "v0.27.0", "nested", false)
1276	before := serial(t, f.fx)
1277	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusConflict ||
1278	str(t, out, "error") != "the home region has more than one box. An import needs a region with one box for now." {
1279	t.Fatalf("stopped = %d %v", code, out)
1280	}
1281	rec, _ := f.record(t, importHost)
1282	requireImportState(t, rec, begun.Seed.ID, seeds.StateUploaded, "")
1283	if serial(t, f.fx) != before {
1284	t.Fatal("a refused stopped bumped the serial")
1285	}
1286	}
1287
1288	// movesOnWrite runs hook, once, just before the first CAS write of key: the
1289	// write that follows loses its CAS, as it would to a concurrent request.
1290	type movesOnWrite struct {
1291	release.Store
1292	key string
1293	hook func()
1294	done atomic.Bool
1295	}
1296
1297	func (s *movesOnWrite) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
1298	if key == s.key && ifMatch != "" && s.done.CompareAndSwap(false, true) {
1299	s.hook()
1300	}
1301	return s.Store.Put(ctx, key, body, ifMatch)
1302	}
1303
1304	// DeleteIfMatch forwards, so a finish behind the wrapper can delete the
1305	// record.
1306	func (s *movesOnWrite) DeleteIfMatch(ctx context.Context, key, etag string) error {
1307	return release.DeleteIfMatch(ctx, s.Store, key, etag)
1308	}
1309
1310	// A stopped or a commit that read the import in one state and finds it in
1311	// another when it writes changes nothing: SD-N60, no audit row and no bump.
1312	// It never acts on an import that has moved on (its transition is guarded
1313	// on the seed and on its from-state).
1314	func TestImportTransitionsLeaveAnImportThatMovedOn(t *testing.T) {
1315	for _, c := range []struct{ verb, action, start string }{
1316	{"stopped", "import-stopped", seeds.StateUploaded},
1317	{"commit", "import-committed", seeds.StatePlaced},
1318	} {
1319	t.Run(c.verb, func(t *testing.T) {
1320	f := newImportFixture(t)
1321	id := newSeedID(t)
1322	f.seedRecord(t, importHost, "", id, seeds.KindImport, c.start)
1323	f.fx.srv.cfg.Releases = &movesOnWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), hook: func() {
1324	// Another request aborts the import after this one read it.
1325	if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, provision.SeedAbort(id)); err != nil {
1326	t.Error(err)
1327	}
1328	}}
1329	before := serial(t, f.fx)
1330	code, out := postImport(t, f, importHost, c.verb)
1331	if code != http.StatusConflict || str(t, out, "error") !=
1332	importHost+" changed while this request was made. Read its state with carlos instances list and try again." {
1333	t.Fatalf("%s = %d %v", c.verb, code, out)
1334	}
1335	rec, _ := f.record(t, importHost)
1336	requireImportState(t, rec, id, seeds.StateAborted, "")
1337	if serial(t, f.fx) != before {
1338	t.Fatal("a lost transition bumped the serial")
1339	}
1340	if rows := importAudit(t, f, c.action); len(rows) != 0 {
1341	t.Fatalf("%s rows = %q", c.action, rows)
1342	}
1343	})
1344	}
1345	}
1346
1347	// A record write that fails for any other reason is the retry sentence, and
1348	// the record is left as it was.
1349	func TestImportTransitionsWriteFailure(t *testing.T) {
1350	f := newImportFixture(t)
1351	id := newSeedID(t)
1352	f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StatePlaced)
1353	rel := &failOnceOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost)}
1354	f.fx.srv.cfg.Releases = rel
1355	rel.armed.Store(true)
1356	before := serial(t, f.fx)
1357	if code, out := postImport(t, f, importHost, "commit"); code != http.StatusBadGateway || str(t, out, "error") != "could not update the import of "+importHost+". Try again." {
1358	t.Fatalf("commit = %d %v", code, out)
1359	}
1360	rec, _ := f.record(t, importHost)
1361	requireImportState(t, rec, id, seeds.StatePlaced, "")
1362	if serial(t, f.fx) != before {
1363	t.Fatal("a failed commit bumped the serial")
1364	}
1365	// The retry lands.
1366	if code, out := postImport(t, f, importHost, "commit"); code != http.StatusOK || out["state"] != "committed" {
1367	t.Fatalf("retried commit = %d %v", code, out)
1368	}
1369	}
1370
1371	// An overlapping duplicate: another request lands the same transition
1372	// between this one's read and its write. A commit is then told what a
1373	// sequential retry is told (200, committed), and writes no audit row and no
1374	// serial bump. A stopped is told what a sequential second stopped is told
1375	// (409 naming the state it is in), not that it changed under it. A hook
1376	// that moves the import somewhere else still gets SD-N60.
1377	func TestImportOverlappingTransitions(t *testing.T) {
1378	changed := func(host string) string {
1379	return host + " changed while this request was made. Read its state with carlos instances list and try again."
1380	}
1381	for _, c := range []struct {
1382	name, verb, action, start string
1383	hook func(id string) []func(string) func(*provision.Record) error
1384	code int
1385	msg string // the error, "" for a 200
1386	state string // the record's state after
1387	}{
1388	{"commit after commit", "commit", "import-committed", seeds.StatePlaced,
1389	func(string) []func(string) func(*provision.Record) error {
1390	return []func(string) func(*provision.Record) error{provision.SeedCommit}
1391	}, http.StatusOK, "", seeds.StateCommitted},
1392	{"commit after abort", "commit", "import-committed", seeds.StatePlaced,
1393	func(string) []func(string) func(*provision.Record) error {
1394	return []func(string) func(*provision.Record) error{provision.SeedAbort}
1395	}, http.StatusConflict, changed(importHost), seeds.StateAborted},
1396	{"stopped after stopped", "stopped", "import-stopped", seeds.StateUploaded,
1397	func(string) []func(string) func(*provision.Record) error {
1398	return []func(string) func(*provision.Record) error{provision.SeedStopped}
1399	}, http.StatusConflict, importHost + " is ready. Mark it stopped once, after its upload.", seeds.StateReady},
1400	{"stopped after stopped and swapped", "stopped", "import-stopped", seeds.StateUploaded,
1401	func(string) []func(string) func(*provision.Record) error {
1402	return []func(string) func(*provision.Record) error{provision.SeedStopped, provision.SeedSwapped}
1403	}, http.StatusConflict, importHost + " is swapped. Mark it stopped once, after its upload.", seeds.StateSwapped},
1404	{"stopped after abort", "stopped", "import-stopped", seeds.StateUploaded,
1405	func(string) []func(string) func(*provision.Record) error {
1406	return []func(string) func(*provision.Record) error{provision.SeedAbort}
1407	}, http.StatusConflict, changed(importHost), seeds.StateAborted},
1408	} {
1409	t.Run(c.name, func(t *testing.T) {
1410	f := newImportFixture(t)
1411	id := newSeedID(t)
1412	f.seedRecord(t, importHost, "", id, seeds.KindImport, c.start)
1413	steps := c.hook(id)
1414	f.fx.srv.cfg.Releases = &movesOnWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), hook: func() {
1415	for _, step := range steps {
1416	if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, step(id)); err != nil {
1417	t.Error(err)
1418	}
1419	}
1420	}}
1421	before := serial(t, f.fx)
1422	code, out := postImport(t, f, importHost, c.verb)
1423	if code != c.code || str(t, out, "error") != c.msg {
1424	t.Fatalf("%s = %d %v, want %d %q", c.verb, code, out, c.code, c.msg)
1425	}
1426	if c.code == http.StatusOK && (len(out) != 1 || out["state"] != c.state) {
1427	t.Fatalf("%s answered %v, want {state: %s}", c.verb, out, c.state)
1428	}
1429	rec, _ := f.record(t, importHost)
1430	requireImportState(t, rec, id, c.state, "")
1431	if serial(t, f.fx) != before {
1432	t.Fatal("the overlapping request bumped the serial")
1433	}
1434	if rows := importAudit(t, f, c.action); len(rows) != 0 {
1435	t.Fatalf("the overlapping request audited %s: %q", c.action, rows)
1436	}
1437	})
1438	}
1439	}
1440
1441	// A delayed request of an import that has gone lands on nothing (final
1442	// review F1): import A is aborted and finished, import B begins on the same
1443	// host, and A's upload, stopped, commit and abort, arriving late, each
1444	// answer SD-N60 and leave B as it was. A request naming no seed is refused
1445	// the same way. B's own requests, and their idempotent repeats, answer as
1446	// before.
1447	func TestImportDelayedMutationsOfAnEarlierImport(t *testing.T) {
1448	f := newImportFixture(t)
1449	body, _ := sqliteBytes(t)
1450	a := f.begun(t, importHost, body).Seed.ID
1451	requireImportState(t, f.uploaded(t, importHost, body), a, seeds.StateUploaded, "")
1452	f.setSeed(t, provision.SeedAbort(a))
1453	f.setSeed(t, provision.SeedRolledBack(a))
1454	if code, out := postImportFor(t, f, importHost, a, "abort"); code != http.StatusOK || out["state"] != "rolled-back" {
1455	t.Fatalf("A's finishing abort = %d %v", code, out)
1456	}
1457	f.requireRecordGone(t, importHost)
1458
1459	b := f.begun(t, importHost, body).Seed.ID
1460	if b == a {
1461	t.Fatal("B reused A's seed id")
1462	}
1463	stale := func(state string, send func(id string) (int, map[string]any)) {
1464	t.Helper()
1465	before := serial(t, f.fx)
1466	for _, id := range []string{a, ""} {
1467	if code, out := send(id); code != http.StatusConflict || str(t, out, "error") != importChanged(importHost) {
1468	t.Fatalf("a request naming seed %q while B is %s = %d %v, want SD-N60", id, state, code, out)
1469	}
1470	f.fx.srv.seedCopies.Wait()
1471	rec, _ := f.record(t, importHost)
1472	requireImportState(t, rec, b, state, "")
1473	}
1474	if serial(t, f.fx) != before {
1475	t.Fatalf("a stale request while B is %s bumped the serial", state)
1476	}
1477	}
1478	stale(seeds.StateAwaitingUpload, func(id string) (int, map[string]any) {
1479	code, out, err := putImportFor(f, importHost, id, body)
1480	if err != nil {
1481	t.Fatal(err)
1482	}
1483	return code, out
1484	})
1485	f.requireNothingUploaded(t)
1486	requireImportState(t, f.uploaded(t, importHost, body), b, seeds.StateUploaded, "")
1487	stale(seeds.StateUploaded, func(id string) (int, map[string]any) { return postImportFor(t, f, importHost, id, "stopped") })
1488	if code, out := postImportFor(t, f, importHost, b, "stopped"); code != http.StatusOK || out["state"] != "ready" {
1489	t.Fatalf("B's stopped = %d %v", code, out)
1490	}
1491	f.setSeed(t, provision.SeedSwapped(b))
1492	f.setSeed(t, provision.SeedPlace(b))
1493	for _, verb := range []string{"commit", "abort"} {
1494	stale(seeds.StatePlaced, func(id string) (int, map[string]any) { return postImportFor(t, f, importHost, id, verb) })
1495	}
1496	if rows := importAudit(t, f, "import-committed"); len(rows) != 0 {
1497	t.Fatalf("a stale commit audited: %q", rows)
1498	}
1499	for range 2 { // B's commit, then its retry
1500	if code, out := postImportFor(t, f, importHost, b, "commit"); code != http.StatusOK || out["state"] != "committed" {
1501	t.Fatalf("B's commit = %d %v", code, out)
1502	}
1503	}
1504	if rows := importAudit(t, f, "import-committed"); len(rows) != 1 {
1505	t.Fatalf("import-committed rows = %q, want B's one", rows)
1506	}
1507	// A's abort after B committed is still SD-N60, not B's SD-N56.
1508	if code, out := postImportFor(t, f, importHost, a, "abort"); code != http.StatusConflict || str(t, out, "error") != importChanged(importHost) {
1509	t.Fatalf("A's abort after B committed = %d %v", code, out)
1510	}
1511	}
1512
1513	// ---- Abort and the delete door (spec §5.6, preflight M8, M14, M15) ----
1514
1515	const (
1516	importDeleteRetry = "could not delete the instance. Try again."
1517	importRunBusy = "carlos run is busy on " + importHost + ". Wait for it to finish, then abort."
1518	)
1519
1520	func importChanged(host string) string {
1521	return host + " changed while this request was made. Read its state with carlos instances list and try again."
1522	}
1523
1524	func importCommitted(host string) string {
1525	return host + " is committed. Delete it with carlos instances delete instead."
1526	}
1527
1528	func importAbortFirst(host string) string {
1529	return host + " is an import that is not committed. Abort the import instead."
1530	}
1531
1532	func importAlreadyAborted(host string) string {
1533	return "the import of " + host + " is already aborted. Once the box puts the old route back on its next pass, delete " + host + " again."
1534	}
1535
1536	// postImportErr is postImport for a goroutine: it returns its error and
1537	// never calls t.Fatal (preflight m9).
1538	func postImportErr(f fromFixture, host, verb string) (int, map[string]any, error) {
1539	req, err := http.NewRequest(http.MethodPost, importDoorURL(f, host, verb, heldSeed(f, host)), nil)
1540	if err != nil {
1541	return 0, nil, err
1542	}
1543	req.Header.Set("Authorization", "Bearer "+f.token)
1544	res, err := http.DefaultClient.Do(req)
1545	if err != nil {
1546	return 0, nil, err
1547	}
1548	defer res.Body.Close()
1549	var out map[string]any
1550	_ = json.NewDecoder(res.Body).Decode(&out)
1551	return res.StatusCode, out, nil
1552	}
1553
1554	// deleteInstance sends the ordinary delete for host and returns what it
1555	// answered.
1556	func deleteInstance(t *testing.T, f fromFixture, host string) (int, map[string]any) {
1557	t.Helper()
1558	code, _, raw := rawReq(t, http.MethodDelete, f.url+"/"+host, f.token, "")
1559	var out map[string]any
1560	_ = json.Unmarshal([]byte(raw), &out)
1561	return code, out
1562	}
1563
1564	// setRun writes the run series' field directly, through the public
1565	// etag-guarded writer: run admission refuses an uncommitted import, so a
1566	// run open on one is only reachable by a write like this (a run admitted
1567	// before the import began, in the spec's words).
1568	func setRun(t *testing.T, f fromFixture, host, state string) {
1569	t.Helper()
1570	rec, etag, err := provision.GetRecordETag(context.Background(), f.fx.rel, "bab", fromApp, host)
1571	if err != nil {
1572	t.Fatal(err)
1573	}
1574	rec.ActiveRun = &provision.RunState{ID: "run_1", State: state, Deadline: time.Now().Add(time.Hour)}
1575	if err := provision.UpdateRecord(context.Background(), f.fx.rel, *rec, etag); err != nil {
1576	t.Fatal(err)
1577	}
1578	}
1579
1580	// A commit and an abort race on the record's etag, and exactly one wins
1581	// (spec §5.6). The loser is told why: an abort that lost to the commit is
1582	// SD-N56, a commit that lost to the abort is SD-N60 (or SD-N115 when the
1583	// abort landed before its read). Neither is a 502.
1584	func TestImportCommitAbortRace(t *testing.T) {
1585	f := newImportFixture(t)
1586	body, _ := sqliteBytes(t)
1587	begun := f.begun(t, importHost, body)
1588	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
1589	if code, out := postImport(t, f, importHost, "stopped"); code != http.StatusOK {
1590	t.Fatalf("stopped = %d %v", code, out)
1591	}
1592	f.setSeed(t, provision.SeedSwapped(begun.Seed.ID))
1593	f.setSeed(t, provision.SeedPlace(begun.Seed.ID))
1594	hosts := []string{importHost}
1595	for i := range 8 {
1596	host := "race" + strconv.Itoa(i) + ".shop.example.com"
1597	f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, seeds.StatePlaced)
1598	hosts = append(hosts, host)
1599	}
1600	type answer struct {
1601	verb string
1602	code int
1603	out map[string]any
1604	err error
1605	}
1606	for _, host := range hosts {
1607	answers := make(chan answer, 2)
1608	for _, verb := range []string{"commit", "abort"} {
1609	go func(verb string) {
1610	code, out, err := postImportErr(f, host, verb)
1611	answers <- answer{verb, code, out, err}
1612	}(verb)
1613	}
1614	got := map[string]answer{}
1615	for range 2 {
1616	a := <-answers
1617	if a.err != nil {
1618	t.Fatalf("%s %s: %v", host, a.verb, a.err)
1619	}
1620	got[a.verb] = a
1621	}
1622	rec, err := f.record(t, host)
1623	if err != nil {
1624	t.Fatal(err)
1625	}
1626	commit, abort := got["commit"], got["abort"]
1627	switch rec.Seed.State {
1628	case seeds.StateCommitted:
1629	if commit.code != http.StatusOK || abort.code != http.StatusConflict || abort.out["error"] != importCommitted(host) {
1630	t.Fatalf("%s committed: commit %d %v, abort %d %v", host, commit.code, commit.out, abort.code, abort.out)
1631	}
1632	case seeds.StateAborted:
1633	lost := commit.out["error"] == importChanged(host) || commit.out["error"] == host+" is aborted, so there is nothing to commit."
1634	if abort.code != http.StatusOK || abort.out["state"] != seeds.StateAborted || commit.code != http.StatusConflict || !lost {
1635	t.Fatalf("%s aborted: commit %d %v, abort %d %v", host, commit.code, commit.out, abort.code, abort.out)
1636	}
1637	default:
1638	t.Fatalf("%s ended %s: commit %d %v, abort %d %v; want exactly one winner", host, rec.Seed.State, commit.code, commit.out, abort.code, abort.out)
1639	}
1640	}
1641	committed, aborted := importAudit(t, f, "import-committed"), importAudit(t, f, "import-aborted")
1642	if len(committed)+len(aborted) != len(hosts) {
1643	t.Fatalf("audit rows: %d committed, %d aborted, want one per host (%d)", len(committed), len(aborted), len(hosts))
1644	}
1645	}
1646
1647	// Abort is refused while a run is open and after commit, lands from every
1648	// state before commit, and is safe to repeat.
1649	func TestImportAbortRefusals(t *testing.T) {
1650	f := newImportFixture(t)
1651	body, sha := sqliteBytes(t)
1652	code, out := beginImport(t, f, importHost, sha, int64(len(body)))
1653	begun, _ := f.requireBegun(t, code, out, importHost)
1654	id := begun.Seed.ID
1655
1656	setRun(t, f, importHost, provision.RunPicked)
1657	before := serial(t, f.fx)
1658	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusConflict || str(t, out, "error") != importRunBusy {
1659	t.Fatalf("abort with a run open = %d %v", code, out)
1660	}
1661	rec, _ := f.record(t, importHost)
1662	requireImportState(t, rec, id, seeds.StateAwaitingUpload, "")
1663	if serial(t, f.fx) != before || len(importAudit(t, f, "import-aborted")) != 0 {
1664	t.Fatal("a refused abort bumped the serial or wrote an audit row")
1665	}
1666
1667	setRun(t, f, importHost, provision.RunDone)
1668	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusOK || len(out) != 1 || out["state"] != "aborted" {
1669	t.Fatalf("abort = %d %v", code, out)
1670	}
1671	rec, _ = f.record(t, importHost)
1672	requireImportState(t, rec, id, seeds.StateAborted, "")
1673	if got := serial(t, f.fx); got != before+1 {
1674	t.Fatalf("serial after abort = %d, want %d", got, before+1)
1675	}
1676	if rows := importAudit(t, f, "import-aborted"); len(rows) != 1 || rows[0] != "aborted the import of "+importHost {
1677	t.Fatalf("import-aborted rows = %q", rows)
1678	}
1679
1680	// Again: 200, and nothing written, audited or bumped.
1681	before = serial(t, f.fx)
1682	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusOK || len(out) != 1 || out["state"] != "aborted" {
1683	t.Fatalf("abort again = %d %v, want 200", code, out)
1684	}
1685	if serial(t, f.fx) != before || len(importAudit(t, f, "import-aborted")) != 1 {
1686	t.Fatal("a repeated abort bumped the serial or wrote an audit row")
1687	}
1688
1689	// After commit, abort is refused: delete is the verb then.
1690	f.seedRecord(t, "done.shop.example.com", "", newSeedID(t), seeds.KindImport, seeds.StateCommitted)
1691	if code, out := postImport(t, f, "done.shop.example.com", "abort"); code != http.StatusConflict ||
1692	str(t, out, "error") != importCommitted("done.shop.example.com") {
1693	t.Fatalf("abort after commit = %d %v", code, out)
1694	}
1695	if rec, _ := f.record(t, "done.shop.example.com"); rec.Seed.State != seeds.StateCommitted {
1696	t.Fatalf("a refused abort moved a committed import to %s", rec.Seed.State)
1697	}
1698	}
1699
1700	// Every state before committed aborts. A refused import keeps the reason
1701	// the box gave (preflight M14: the no-reason SeedAbort).
1702	func TestImportAbortFromEveryStateBeforeCommit(t *testing.T) {
1703	f := newImportFixture(t)
1704	for _, state := range []string{seeds.StateAwaitingUpload, seeds.StateUploaded, seeds.StateReady,
1705	seeds.StateSwapped, seeds.StatePlaced, seeds.StateRefused} {
1706	host := state + ".shop.example.com"
1707	id := newSeedID(t)
1708	f.seedRecord(t, host, "", id, seeds.KindImport, state)
1709	reason := ""
1710	if state == seeds.StateRefused {
1711	reason = "the box found no addr row"
1712	rec, etag, err := provision.GetRecordETag(context.Background(), f.fx.rel, "bab", fromApp, host)
1713	if err != nil {
1714	t.Fatal(err)
1715	}
1716	rec.Seed.Reason = reason
1717	raw, _ := json.Marshal(rec)
1718	if _, err := f.fx.rel.Put(context.Background(), provision.RecordKey("bab", fromApp, host), raw, etag); err != nil {
1719	t.Fatal(err)
1720	}
1721	}
1722	if code, out := postImport(t, f, host, "abort"); code != http.StatusOK || out["state"] != "aborted" {
1723	t.Errorf("abort while %s = %d %v", state, code, out)
1724	continue
1725	}
1726	rec, _ := f.record(t, host)
1727	requireImportState(t, rec, id, seeds.StateAborted, reason)
1728	}
1729	}
1730
1731	// A rolled-back import is finished by its abort: the record and the
1732	// snapshot go, and the answer says rolled-back. A finish that fails still
1733	// answers 200 (the abort has landed), and leaves the record rolled-back
1734	// for the next abort or the sweep.
1735	func TestImportAbortFinishesARolledBackImport(t *testing.T) {
1736	f := newImportFixture(t)
1737	body, _ := sqliteBytes(t)
1738	begun := f.begun(t, importHost, body)
1739	id := begun.Seed.ID
1740	requireImportState(t, f.uploaded(t, importHost, body), id, seeds.StateUploaded, "")
1741	f.setSeed(t, provision.SeedAbort(id))
1742	f.setSeed(t, provision.SeedRolledBack(id))
1743
1744	data := f.fx.srv.cfg.SeedData
1745	f.fx.srv.cfg.SeedData = nil // the finish cannot delete the snapshot
1746	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusOK || len(out) != 1 || out["state"] != "rolled-back" {
1747	t.Fatalf("abort with a failing finish = %d %v", code, out)
1748	}
1749	rec, _ := f.record(t, importHost)
1750	requireImportState(t, rec, id, seeds.StateRolledBack, "")
1751
1752	f.fx.srv.cfg.SeedData = data
1753	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusOK || len(out) != 1 || out["state"] != "rolled-back" {
1754	t.Fatalf("abort of a rolled-back import = %d %v", code, out)
1755	}
1756	f.requireRecordGone(t, importHost)
1757	f.requireNothingUploaded(t)
1758	if rows := importAudit(t, f, "import-aborted"); len(rows) != 0 {
1759	t.Fatalf("finishing a rolled-back import audited an abort: %q", rows)
1760	}
1761	}
1762
1763	// An abort that loses its race re-reads once and answers by what it finds
1764	// (preflight M14): another abort is a 200, a commit is SD-N56, a rollback
1765	// is finished and a 200, a run that opened is SD-N58, and anything else is
1766	// SD-N60. The loser writes no audit row and bumps nothing.
1767	func TestImportAbortLosesARace(t *testing.T) {
1768	ctx := context.Background()
1769	for _, c := range []struct {
1770	name string
1771	hook func(f fromFixture, id string) error
1772	code int
1773	state string // the 200's state, or "" for an error
1774	msg string
1775	after string // the record's state after, "" for gone
1776	}{
1777	{"to another abort", func(f fromFixture, id string) error {
1778	_, err := provision.SetRecordSeed(ctx, f.fx.rel, "bab", fromApp, importHost, provision.SeedAbort(id))
1779	return err
1780	}, http.StatusOK, seeds.StateAborted, "", seeds.StateAborted},
1781	{"to a commit", func(f fromFixture, id string) error {
1782	_, err := provision.SetRecordSeed(ctx, f.fx.rel, "bab", fromApp, importHost, provision.SeedCommit(id))
1783	return err
1784	}, http.StatusConflict, "", importCommitted(importHost), seeds.StateCommitted},
1785	{"to a rollback", func(f fromFixture, id string) error {
1786	for _, step := range []func(string) func(*provision.Record) error{provision.SeedAbort, provision.SeedRolledBack} {
1787	if _, err := provision.SetRecordSeed(ctx, f.fx.rel, "bab", fromApp, importHost, step(id)); err != nil {
1788	return err
1789	}
1790	}
1791	return nil
1792	}, http.StatusOK, seeds.StateRolledBack, "", ""},
1793	{"to a run", func(f fromFixture, id string) error {
1794	rec, etag, err := provision.GetRecordETag(ctx, f.fx.rel, "bab", fromApp, importHost)
1795	if err != nil {
1796	return err
1797	}
1798	rec.ActiveRun = &provision.RunState{ID: "run_1", State: provision.RunQueued, Deadline: time.Now().Add(time.Hour)}
1799	return provision.UpdateRecord(ctx, f.fx.rel, *rec, etag)
1800	}, http.StatusConflict, "", importRunBusy, seeds.StatePlaced},
1801	{"to another import of the host", func(f fromFixture, id string) error {
1802	if err := f.fx.rel.Delete(ctx, provision.RecordKey("bab", fromApp, importHost)); err != nil {
1803	return err
1804	}
1805	other, err := seeds.NewID()
1806	if err != nil {
1807	return err
1808	}
1809	rec := provision.Record{Schema: 1, Host: importHost, Slug: "old", App: fromApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC(),
1810	Seed: &provision.SeedSource{ID: other, Kind: seeds.KindImport, State: seeds.StateAborted, At: time.Now().UTC()}}
1811	return provision.CreateRecord(ctx, f.fx.rel, rec)
1812	}, http.StatusConflict, "", importChanged(importHost), seeds.StateAborted},
1813	} {
1814	t.Run(c.name, func(t *testing.T) {
1815	f := newImportFixture(t)
1816	id := newSeedID(t)
1817	f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StatePlaced)
1818	f.fx.srv.cfg.Releases = &movesOnWrite{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), hook: func() {
1819	if err := c.hook(f, id); err != nil {
1820	t.Error(err)
1821	}
1822	}}
1823	before := serial(t, f.fx)
1824	code, out := postImport(t, f, importHost, "abort")
1825	if code != c.code || (c.state != "" && (len(out) != 1 || out["state"] != c.state)) || (c.msg != "" && str(t, out, "error") != c.msg) {
1826	t.Fatalf("abort = %d %v, want %d %q %q", code, out, c.code, c.state, c.msg)
1827	}
1828	rec, err := f.record(t, importHost)
1829	switch {
1830	case c.after == "":
1831	f.requireRecordGone(t, importHost)
1832	case err != nil || rec.Seed == nil || rec.Seed.State != c.after:
1833	t.Fatalf("record after = %+v, %v; want %s", rec, err, c.after)
1834	}
1835	if c.after != "" && serial(t, f.fx) != before {
1836	t.Fatal("the losing abort bumped the serial")
1837	}
1838	if rows := importAudit(t, f, "import-aborted"); len(rows) != 0 {
1839	t.Fatalf("the losing abort audited: %q", rows)
1840	}
1841	})
1842	}
1843	}
1844
1845	// Abort shares the import gate.
1846	func TestImportAbortGate(t *testing.T) {
1847	f := newImportFixture(t)
1848	f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReady)
1849	f.seedRecord(t, importHost, "", newSeedID(t), seeds.KindImport, seeds.StatePlaced)
1850	for _, c := range []struct {
1851	name, host string
1852	code int
1853	msg string
1854	}{
1855	{"no record", "ghost.shop.example.com", http.StatusNotFound, "ghost.shop.example.com is not an instance of shop"},
1856	{"not a host", "Not_A_Host", http.StatusNotFound, "Not_A_Host is not an instance of shop"},
1857	{"an ordinary instance", fromSource, http.StatusConflict, fromSource + " is not an import."},
1858	{"a --from seed", fromHost, http.StatusConflict, fromHost + " is not an import."},
1859	} {
1860	if code, out := postImport(t, f, c.host, "abort"); code != c.code || str(t, out, "error") != c.msg {
1861	t.Errorf("abort %s = %d %v, want %d %q", c.name, code, out, c.code, c.msg)
1862	}
1863	}
1864	f.fx.srv.cfg.HostClaim = true
1865	if code, out := postImport(t, f, importHost, "abort"); code != http.StatusConflict ||
1866	str(t, out, "error") != seedClaimConsoleRefusal().Msg {
1867	t.Errorf("abort with claims armed = %d %v", code, out)
1868	}
1869	f.fx.srv.cfg.HostClaim = false
1870	if rec, _ := f.record(t, importHost); rec.Seed.State != seeds.StatePlaced {
1871	t.Fatalf("a refused abort moved the import to %s", rec.Seed.State)
1872	}
1873	}
1874
1875	// The delete door's import rule (spec §5.6, codex R2-7): an import no box
1876	// has swapped is rolled back and finished by the delete, answered as the
1877	// ordinary delete is; a rolled-back one is finished; every other
1878	// uncommitted import is refused with SD-N57; a committed one is an
1879	// ordinary instance.
1880	func TestImportDeleteRefusedUntilCommittedOrUnswapped(t *testing.T) {
1881	f := newImportFixture(t)
1882	body, sha := sqliteBytes(t)
1883	code, out := beginImport(t, f, "early.shop.example.com", sha, int64(len(body)))
1884	f.requireBegun(t, code, out, "early.shop.example.com")
1885	before := serial(t, f.fx)
1886	if code, out := deleteInstance(t, f, "early.shop.example.com"); code != http.StatusOK || len(out) != 1 || out["result"] != "deleted" {
1887	t.Fatalf("delete before upload = %d %v, want 200 deleted", code, out)
1888	}
1889	f.requireRecordGone(t, "early.shop.example.com")
1890	if serial(t, f.fx) <= before {
1891	t.Fatal("the delete did not bump the serial")
1892	}
1893	if rows := importAudit(t, f, "instance-delete"); len(rows) != 1 || rows[0] != "deleted instance early.shop.example.com (cli)" {
1894	t.Fatalf("instance-delete rows = %q", rows)
1895	}
1896	// Again: the ordinary delete's idempotent 200.
1897	if code, out := deleteInstance(t, f, "early.shop.example.com"); code != http.StatusOK || out["result"] != "deleted" {
1898	t.Fatalf("second delete = %d %v", code, out)
1899	}
1900
1901	// Uploaded: the snapshot goes too.
1902	begun := f.begun(t, importHost, body)
1903	requireImportState(t, f.uploaded(t, importHost, body), begun.Seed.ID, seeds.StateUploaded, "")
1904	if keys, _ := f.data.List(context.Background(), seeds.Prefix); len(keys) != 1 {
1905	t.Fatalf("seeds/ before the delete = %v, want the snapshot", keys)
1906	}
1907	if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
1908	t.Fatalf("delete after upload = %d %v", code, out)
1909	}
1910	f.requireRecordGone(t, importHost)
1911	f.requireNothingUploaded(t)
1912
1913	// Refused (no box swapped) and rolled-back are finished; aborted,
1914	// ready, swapped and placed are the box's to roll back first.
1915	for _, c := range []struct {
1916	state string
1917	gone bool
1918	}{
1919	{seeds.StateRefused, true},
1920	{seeds.StateRolledBack, true},
1921	{seeds.StateReady, false},
1922	{seeds.StateSwapped, false},
1923	{seeds.StatePlaced, false},
1924	{seeds.StateAborted, false},
1925	} {
1926	host := c.state + ".shop.example.com"
1927	f.seedRecord(t, host, "", newSeedID(t), seeds.KindImport, c.state)
1928	code, out := deleteInstance(t, f, host)
1929	if c.gone {
1930	if code != http.StatusOK || out["result"] != "deleted" {
1931	t.Errorf("delete while %s = %d %v, want 200", c.state, code, out)
1932	}
1933	f.requireRecordGone(t, host)
1934	continue
1935	}
1936	want := importAbortFirst(host)
1937	if c.state == seeds.StateAborted {
1938	want = importAlreadyAborted(host)
1939	}
1940	if code != http.StatusConflict || str(t, out, "error") != want {
1941	t.Errorf("delete while %s = %d %v, want 409 %q", c.state, code, out, want)
1942	}
1943	if rec, err := f.record(t, host); err != nil || rec.Seed.State != c.state {
1944	t.Errorf("a refused delete while %s left %+v, %v", c.state, rec, err)
1945	}
1946	}
1947
1948	// Committed, a --from seed in any state, and an unseeded instance: the
1949	// ordinary delete.
1950	f.seedRecord(t, "done.shop.example.com", "", newSeedID(t), seeds.KindImport, seeds.StateCommitted)
1951	f.seedRecord(t, fromHost, "", newSeedID(t), seeds.KindReplica, seeds.StateReady)
1952	for _, host := range []string{"done.shop.example.com", fromHost, fromSource} {
1953	if code, out := deleteInstance(t, f, host); code != http.StatusOK || out["result"] != "deleted" {
1954	t.Fatalf("delete of %s = %d %v", host, code, out)
1955	}
1956	f.requireRecordGone(t, host)
1957	}
1958	}
1959
1960	// A record the delete door can read but not decode, or one that disagrees
1961	// with its own path, is no import: the ordinary delete removes it, as it
1962	// did before the import rule (review I1).
1963	func TestImportDeleteRemovesARecordItCannotDecode(t *testing.T) {
1964	f := newImportFixture(t)
1965	var logs bytes.Buffer
1966	log.SetOutput(&logs)
1967	defer log.SetOutput(os.Stderr)
1968	other, _ := json.Marshal(provision.Record{Schema: 1, Host: "other.shop.example.com", Slug: "other", App: fromApp, Account: "bab",
1969	Channel: "stable", CreatedAt: time.Now().UTC(),
1970	Seed: &provision.SeedSource{ID: newSeedID(t), Kind: seeds.KindImport, State: seeds.StateAwaitingUpload, At: time.Now().UTC()}})
1971	for host, body := range map[string][]byte{
1972	"junk.shop.example.com": []byte("not json"),
1973	"mismatch.shop.example.com": other,
1974	} {
1975	key := provision.RecordKey("bab", fromApp, host)
1976	if _, err := f.fx.rel.Put(context.Background(), key, body, ""); err != nil {
1977	t.Fatal(err)
1978	}
1979	if code, out := deleteInstance(t, f, host); code != http.StatusOK || out["result"] != "deleted" {
1980	t.Fatalf("delete of %s = %d %v, want the ordinary 200", host, code, out)
1981	}
1982	if _, _, err := f.fx.rel.Get(context.Background(), key); !errors.Is(err, os.ErrNotExist) {
1983	t.Fatalf("%s survived its delete: %v", host, err)
1984	}
1985	if !strings.Contains(logs.String(), host+": the record could not be read as an import") {
1986	t.Fatalf("no log line names %s: %s", host, logs.String())
1987	}
1988	}
1989	}
1990
1991	// deleteRace plays a second delete of the same import against this one:
1992	// hook runs just before this delete's rollback CAS write (so it loses),
1993	// and gone, when set, deletes the record just before the re-read that
1994	// follows the loss, as the other delete's finish would.
1995	type deleteRace struct {
1996	release.Store
1997	key string
1998	hook func()
1999	gone bool
2000	puts atomic.Int32
2001	reads atomic.Int32
2002	}
2003
2004	func (s *deleteRace) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
2005	if key == s.key && ifMatch != "" && s.puts.Add(1) == 1 {
2006	s.hook()
2007	}
2008	return s.Store.Put(ctx, key, body, ifMatch)
2009	}
2010
2011	func (s *deleteRace) Get(ctx context.Context, key string) ([]byte, string, error) {
2012	// Reads after the hook: the CAS's retry, then the delete door's re-read.
2013	if key == s.key && s.gone && s.puts.Load() > 0 && s.reads.Add(1) == 2 {
2014	if err := s.Store.Delete(ctx, key); err != nil {
2015	return nil, "", err
2016	}
2017	}
2018	return s.Store.Get(ctx, key)
2019	}
2020
2021	func (s *deleteRace) DeleteIfMatch(ctx context.Context, key, etag string) error {
2022	return release.DeleteIfMatch(ctx, s.Store, key, etag)
2023	}
2024
2025	// Two deletes of one pre-swap import overlap. The one whose rollback CAS
2026	// loses finishes the import when it finds it rolled back, and answers the
2027	// ordinary idempotent 200 when the other has already removed the record
2028	// (review M3).
2029	func TestImportDeletesOverlap(t *testing.T) {
2030	for _, gone := range []bool{false, true} {
2031	t.Run(map[bool]string{false: "rolled back", true: "finished"}[gone], func(t *testing.T) {
2032	f := newImportFixture(t)
2033	id := newSeedID(t)
2034	f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StateUploaded)
2035	f.fx.srv.cfg.Releases = &deleteRace{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost), gone: gone, hook: func() {
2036	if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, "bab", fromApp, importHost, provision.SeedRolledBackBeforeSwap(id)); err != nil {
2037	t.Error(err)
2038	}
2039	}}
2040	if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
2041	t.Fatalf("overlapping delete = %d %v, want 200", code, out)
2042	}
2043	f.requireRecordGone(t, importHost)
2044	})
2045	}
2046	}
2047
2048	// The pre-swap delete of a private-storage import tears down the empty
2049	// store its begin made (preflight m11).
2050	func TestImportDeleteTearsDownTheStore(t *testing.T) {
2051	f := newImportFixture(t)
2052	enableInstances(t, f.fx, fromApp, fromManifest(true))
2053	code, out := beginImport(t, f, importHost, importSHA, 10)
2054	rec, _ := f.requireBegun(t, code, out, importHost)
2055	store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, rec.Environment)
2056	if err != nil || store == nil || store.Declared == nil || store.Declared.Seed != rec.Seed.ID {
2057	t.Fatalf("store = %+v, %v; want one marked with seed %s", store, err, rec.Seed.ID)
2058	}
2059	bucket := storeBucketName(store)
2060	if _, ok := f.fake.Buckets[bucket]; !ok {
2061	t.Fatalf("bucket %s was never made", bucket)
2062	}
2063	if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
2064	t.Fatalf("delete = %d %v", code, out)
2065	}
2066	f.requireRecordGone(t, importHost)
2067	if store, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, rec.Environment); err != nil || store != nil {
2068	t.Fatalf("store after the delete = %+v, %v; want it gone", store, err)
2069	}
2070	if _, ok := f.fake.Buckets[bucket]; ok {
2071	t.Fatalf("bucket %s survived the delete", bucket)
2072	}
2073	}
2074
2075	// failGetOn fails reads of key while armed.
2076	type failGetOn struct {
2077	release.Store
2078	key string
2079	armed atomic.Bool
2080	}
2081
2082	func (s *failGetOn) Get(ctx context.Context, key string) ([]byte, string, error) {
2083	if key == s.key && s.armed.Load() {
2084	return nil, "", errors.New("the bucket could not be read")
2085	}
2086	return s.Store.Get(ctx, key)
2087	}
2088
2089	// The delete door's import rule fails closed (preflight M15, M8): a record
2090	// it cannot read, a finish that fails, and a seed id that cannot name a key
2091	// each answer main's delete sentence and delete nothing. A finish that
2092	// fails leaves the record rolled-back for a retry, which completes it.
2093	func TestImportDeleteFailsClosed(t *testing.T) {
2094	f := newImportFixture(t)
2095	id := newSeedID(t)
2096	f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StateUploaded)
2097
2098	rel := &failGetOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, importHost)}
2099	rel.armed.Store(true)
2100	f.fx.srv.cfg.Releases = rel
2101	if code, out := deleteInstance(t, f, importHost); code != http.StatusBadGateway || str(t, out, "error") != importDeleteRetry {
2102	t.Fatalf("delete with an unreadable record = %d %v", code, out)
2103	}
2104	f.fx.srv.cfg.Releases = f.fx.rel
2105	rec, _ := f.record(t, importHost)
2106	requireImportState(t, rec, id, seeds.StateUploaded, "")
2107
2108	data := f.fx.srv.cfg.SeedData
2109	f.fx.srv.cfg.SeedData = nil
2110	if code, out := deleteInstance(t, f, importHost); code != http.StatusBadGateway || str(t, out, "error") != importDeleteRetry {
2111	t.Fatalf("delete with a failing finish = %d %v", code, out)
2112	}
2113	rec, _ = f.record(t, importHost)
2114	requireImportState(t, rec, id, seeds.StateRolledBack, "")
2115	if rows := importAudit(t, f, "instance-delete"); len(rows) != 0 {
2116	t.Fatalf("a failed delete audited: %q", rows)
2117	}
2118	f.fx.srv.cfg.SeedData = data
2119	if code, out := deleteInstance(t, f, importHost); code != http.StatusOK || out["result"] != "deleted" {
2120	t.Fatalf("retried delete = %d %v", code, out)
2121	}
2122	f.requireRecordGone(t, importHost)
2123
2124	// A seed id that cannot name a key: nothing is written or deleted, and
2125	// the shared sentinel key survives.
2126	sentinel := seeds.SnapshotKey("bad.shop.example.com", "not-a-seed-id")
2127	if _, err := f.data.Put(context.Background(), sentinel, []byte("another record's object"), ""); err != nil {
2128	t.Fatal(err)
2129	}
2130	f.seedRecord(t, "bad.shop.example.com", "", "not-a-seed-id", seeds.KindImport, seeds.StateUploaded)
2131	if code, out := deleteInstance(t, f, "bad.shop.example.com"); code != http.StatusBadGateway || str(t, out, "error") != importDeleteRetry {
2132	t.Fatalf("delete of a bad seed id = %d %v", code, out)
2133	}
2134	if rec, err := f.record(t, "bad.shop.example.com"); err != nil || rec.Seed.State != seeds.StateUploaded {
2135	t.Fatalf("record of a bad seed id = %+v, %v; want it untouched", rec, err)
2136	}
2137	if got := f.rawKey(t, f.data, sentinel); got != "another record's object" {
2138	t.Fatalf("sentinel = %q", got)
2139	}
2140	}
2141
2142	// On a claim console the delete of an uncommitted import is refused as
2143	// every import door is: the finish runs nowhere there, and the claim's
2144	// ordinary retire would delete the record and leak what the seed made.
2145	func TestImportDeleteOnAClaimConsole(t *testing.T) {
2146	f := newImportFixture(t)
2147	id := newSeedID(t)
2148	f.seedRecord(t, importHost, "", id, seeds.KindImport, seeds.StateAwaitingUpload)
2149	f.fx.srv.cfg.HostClaim = true
2150	defer func() { f.fx.srv.cfg.HostClaim = false }()
2151	if code, out := deleteInstance(t, f, importHost); code != http.StatusConflict || str(t, out, "error") != seedClaimConsoleRefusal().Msg {
2152	t.Fatalf("delete on a claim console = %d %v", code, out)
2153	}
2154	rec, _ := f.record(t, importHost)
2155	requireImportState(t, rec, id, seeds.StateAwaitingUpload, "")
2156	}
2157