amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package console
2
3import (
4 "context"
5 "crypto/sha256"
6 "database/sql"
7 "encoding/hex"
8 "encoding/json"
9 "errors"
10 "fmt"
11 "net/http"
12 "net/http/httptest"
13 "os"
14 "path/filepath"
15 "strconv"
16 "strings"
17 "sync/atomic"
18 "testing"
19 "time"
20
21 lsfile "github.com/benbjohnson/litestream/file"
22
23 "github.com/carlosframework/platform/internal/dbdoor"
24 "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
25 "github.com/carlosframework/platform/internal/objstore"
26 "github.com/carlosframework/platform/internal/provision"
27 "github.com/carlosframework/platform/internal/release"
28 "github.com/carlosframework/platform/internal/seeds"
29 "github.com/carlosframework/platform/internal/svcrole"
30)
31
32const (
33 fromApp = "shop"
34 fromDomain = "shop.example.com"
35 fromSource = "main.shop.example.com"
36 fromHost = "canary.shop.example.com"
37 fromBody = `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com"}}`
38 fromFailed = "could not copy main.shop.example.com. Nothing was created. Try again."
39 // fromAbortedReason is what a copy that fails after the answer leaves on
40 // its record: the host stays taken until the sweep finishes it.
41 fromAbortedReason = "could not copy main.shop.example.com, so no instance was created. canary.shop.example.com is free again within about 5 minutes. Try again then."
42 fromGrew = "the copy of main.shop.example.com grew past its planned size, so it was stopped and no instance was created."
43)
44
45// fromFixture is a console ready for --from: the app shop (no pool, its own
46// domain), a source instance main with a real litestream replica, a data
47// bucket for seeds, a scratch pool, the store plane armed, and a ready
48// home region.
49type fromFixture struct {
50 fx *consoleFixture
51 fake *objstore.MemProvisioner
52 token string
53 data release.Store
54 url string
55 root string // the replica bucket's root
56 txns int
57}
58
59func fromManifest(private bool) provision.Manifest {
60 return provision.Manifest{Schema: 1, App: fromApp, Domain: fromDomain, Channel: "stable", PrivateStorage: private}
61}
62
63func newFromFixture(t *testing.T, txns int) fromFixture {
64 t.Helper()
65 fx, fake, token, _ := privateStorageFixture(t, "code-from")
66 if _, err := fx.st.CreateAppIn(OpsAccountID, fromApp, "paul@keymail.dev", time.Now()); err != nil {
67 t.Fatal(err)
68 }
69 enableInstances(t, fx, fromApp, fromManifest(false))
70 fx.srv.cfg.HibernateBucket = "data-test"
71 fx.srv.cfg.HibernatePrefix = "instances"
72 fx.srv.cfg.Region = ""
73 data := release.NewDirStore(t.TempDir())
74 fx.srv.cfg.SeedData = data
75 scratch := filepath.Join(t.TempDir(), "scratch", "seeds")
76 if err := clearSeedScratch(scratch); err != nil {
77 t.Fatal(err)
78 }
79 fx.srv.seedScratch = newSeedScratch(scratch, func(string) (int64, error) { return 64 << 30, nil })
80 root := t.TempDir()
81 fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
82 return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil
83 }
84 f := fromFixture{fx: fx, fake: fake, token: token, data: data, url: instancesURL(fx, "bab", fromApp), root: root, txns: txns}
85 f.addSource(t, provision.Record{Host: fromSource, Slug: "main"}, txns)
86 publishSeedsFloorFx(t, fx, "v0.27.0")
87 seedReadyBox(t, fx, "kass-1", "", 1, "v0.27.0", "v0.27.0", "nested", false)
88 // A copy still running when the test ends would write into removed
89 // directories: every test waits for its copies.
90 t.Cleanup(fx.srv.seedCopies.Wait)
91 return f
92}
93
94// addSource writes a source record of shop (rec's other fields as given)
95// and, when txns > 0, a replica of that many batches under its prefix.
96func (f fromFixture) addSource(t *testing.T, rec provision.Record, txns int) {
97 t.Helper()
98 if txns > 0 {
99 dbdoortest.Replicate(t, filepath.Join(f.root, "instances", rec.Host), txns)
100 }
101 rec.Schema, rec.App, rec.Account, rec.Channel, rec.CreatedAt = 1, fromApp, "bab", "stable", time.Now().UTC()
102 if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil {
103 t.Fatal(err)
104 }
105}
106
107// create posts a create and returns what the door answered.
108func (f fromFixture) create(t *testing.T, body string) (int, map[string]any) {
109 t.Helper()
110 return f.createAs(t, f.token, body)
111}
112
113func (f fromFixture) createAs(t *testing.T, token, body string) (int, map[string]any) {
114 t.Helper()
115 code, _, raw := rawReq(t, http.MethodPost, f.url, token, body)
116 var out map[string]any
117 _ = json.Unmarshal([]byte(raw), &out)
118 return code, out
119}
120
121// copy posts a --from create, requires its 202 with a reserving record,
122// waits for the detached copy to finish, and returns the record it left.
123func (f fromFixture) copy(t *testing.T, body string) *provision.Record {
124 t.Helper()
125 return f.copyAs(t, f.token, body)
126}
127
128func (f fromFixture) copyAs(t *testing.T, token, body string) *provision.Record {
129 t.Helper()
130 code, out := f.createAs(t, token, body)
131 seed, _ := out["seed"].(map[string]any)
132 if code != http.StatusAccepted || seed == nil || seed["state"] != seeds.StateReserving || seed["kind"] != seeds.KindReplica {
133 t.Fatalf("%s = %d %v, want 202 with a reserving record", body, code, out)
134 }
135 f.fx.srv.seedCopies.Wait()
136 host, _ := out["host"].(string)
137 rec, err := f.record(t, host)
138 if err != nil || rec.Seed == nil || rec.Seed.ID != seed["id"] {
139 t.Fatalf("record %s after its copy = %+v, %v", host, rec, err)
140 }
141 return rec
142}
143
144func (f fromFixture) record(t *testing.T, host string) (*provision.Record, error) {
145 t.Helper()
146 return provision.GetRecord(context.Background(), f.fx.rel, "bab", fromApp, host)
147}
148
149// requireAborted: the record's replica seed is aborted, saying why.
150func requireAborted(t *testing.T, rec *provision.Record, reason string) {
151 t.Helper()
152 if rec.Seed == nil || rec.Seed.State != seeds.StateAborted || rec.Seed.Kind != seeds.KindReplica || rec.Seed.Reason != reason {
153 t.Fatalf("record %s seed = %+v; want aborted with reason %q", rec.Host, rec.Seed, reason)
154 }
155}
156
157func requireReady(t *testing.T, rec *provision.Record) {
158 t.Helper()
159 if rec.Seed == nil || rec.Seed.State != seeds.StateReady || rec.Seed.Reason != "" {
160 t.Fatalf("record %s seed = %+v; want ready", rec.Host, rec.Seed)
161 }
162}
163
164// requireNoRecord: the refusal came before the record was reserved.
165func (f fromFixture) requireNoRecord(t *testing.T, host string) {
166 t.Helper()
167 if _, err := f.record(t, host); !errors.Is(err, os.ErrNotExist) {
168 t.Fatalf("record %s: err = %v, want none written", host, err)
169 }
170}
171
172// requireClean: the scratch is empty and the pool holds nothing for shop.
173func (f fromFixture) requireClean(t *testing.T) {
174 t.Helper()
175 if ents, _ := os.ReadDir(f.fx.srv.seedScratch.dir()); len(ents) != 0 {
176 t.Fatalf("scratch left behind: %v", ents)
177 }
178 rel, ref := f.fx.srv.seedScratch.reserve("bab/"+fromApp, 1)
179 if ref != nil {
180 t.Fatalf("the reservation was not released: %+v", ref)
181 }
182 rel()
183}
184
185func (f fromFixture) requireNothingUploaded(t *testing.T) {
186 t.Helper()
187 if keys, err := f.data.List(context.Background(), seeds.Prefix); err != nil || len(keys) != 0 {
188 t.Fatalf("seeds/ = %v, %v; want nothing uploaded", keys, err)
189 }
190}
191
192func snapshotRows(t *testing.T, raw []byte) int {
193 t.Helper()
194 path := filepath.Join(t.TempDir(), "snapshot.db")
195 if err := os.WriteFile(path, raw, 0o600); err != nil {
196 t.Fatal(err)
197 }
198 db, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
199 if err != nil {
200 t.Fatal(err)
201 }
202 defer db.Close()
203 var n int
204 if err := db.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
205 t.Fatal(err)
206 }
207 return n
208}
209
210func TestFromCreatesAReadySeededInstance(t *testing.T) {
211 f := newFromFixture(t, 30)
212 rec := f.copy(t, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com"},"message":"a copy to try the upgrade on"}`)
213 requireReady(t, rec)
214 if rec.Seed.Kind != seeds.KindReplica || rec.Seed.From != fromSource || rec.Seed.TXID == "" || rec.Seed.Bytes == 0 ||
215 len(rec.Seed.SHA256) != 64 || rec.Seed.By != "alice@keymail.dev" || !seeds.ValidID(rec.Seed.ID) {
216 t.Fatalf("record seed = %+v", rec.Seed)
217 }
218 if rec.Environment != "" || rec.Slug != "canary" || rec.Channel != "stable" {
219 t.Fatalf("record = %+v; want the ordinary create's fields", rec)
220 }
221 raw, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(rec.Host, rec.Seed.ID))
222 if err != nil {
223 t.Fatalf("snapshot: %v", err)
224 }
225 sum := sha256.Sum256(raw)
226 if hex.EncodeToString(sum[:]) != rec.Seed.SHA256 || int64(len(raw)) != rec.Seed.Bytes {
227 t.Fatalf("snapshot is %d bytes, sha %x; record says %d, %s", len(raw), sum, rec.Seed.Bytes, rec.Seed.SHA256)
228 }
229 if n := snapshotRows(t, raw); n != f.txns*dbdoortest.Rows {
230 t.Fatalf("snapshot holds %d rows, want %d", n, f.txns*dbdoortest.Rows)
231 }
232 f.requireClean(t)
233 events, err := f.fx.st.Events(OpsAccountID, fromApp, 50)
234 if err != nil {
235 t.Fatal(err)
236 }
237 var rows []string
238 for _, e := range events {
239 if e.Action == "instance-seed" {
240 rows = append(rows, e.Detail+" | "+e.Address)
241 }
242 if e.Action == "instance-create" {
243 t.Errorf("a seeded create also wrote an instance-create row: %+v", e)
244 }
245 }
246 want := "seeded canary.shop.example.com from main.shop.example.com at txid " + rec.Seed.TXID + " | alice@keymail.dev"
247 if len(rows) != 1 || rows[0] != want {
248 t.Fatalf("instance-seed rows = %q, want [%q]", rows, want)
249 }
250}
251
252func TestFromRefusals(t *testing.T) {
253 f := newFromFixture(t, 30)
254 for _, c := range []struct {
255 body string
256 code int
257 want string
258 }{
259 {`{"host":"c1.shop.example.com","from":{"host":"nope.shop.example.com"}}`, http.StatusConflict, "no instance nope.shop.example.com in shop. A copy can only come from an instance of the same app in the same account."},
260 {`{"host":"c2.shop.example.com","kind":"static","from":{"host":"main.shop.example.com"}}`, http.StatusBadRequest, "a seeded instance is an ordinary instance. Drop --kind, --addr and --role."},
261 {`{"host":"c2.shop.example.com","addrs":["127.0.0.1:9000"],"from":{"host":"main.shop.example.com"}}`, http.StatusBadRequest, "a seeded instance is an ordinary instance"},
262 {`{"host":"c2.shop.example.com","role":"replica","from":{"host":"main.shop.example.com"}}`, http.StatusBadRequest, "a seeded instance is an ordinary instance"},
263 {`{"host":"c2.shop.example.com","replicas":["us"],"from":{"host":"main.shop.example.com"}}`, http.StatusBadRequest, "a seeded instance is an ordinary instance"},
264 {`{"host":"main.shop.example.com","from":{"host":"main.shop.example.com"}}`, http.StatusConflict, "instance main.shop.example.com already exists"},
265 } {
266 code, out := f.create(t, c.body)
267 if code != c.code || !strings.Contains(str(t, out, "error"), c.want) {
268 t.Errorf("%s = %d %v, want %d %q", c.body, code, out, c.code, c.want)
269 }
270 }
271 for _, h := range []string{"c1.shop.example.com", "c2.shop.example.com"} {
272 f.requireNoRecord(t, h)
273 }
274 f.requireClean(t)
275}
276
277// --at: a target inside the history restores exactly there, and one at the
278// end is the latest copy. One past the end is refused before anything is
279// reserved, naming the latest transaction the source has replicated.
280func TestFromAt(t *testing.T) {
281 f := newFromFixture(t, 30)
282 latest, err := dbdoor.PlanRestore(context.Background(), dbdoor.NewClient(replicaDoer(lsfile.NewReplicaClient(filepath.Join(f.root, "instances", fromSource)))), 0)
283 if err != nil {
284 t.Fatal(err)
285 }
286 for _, c := range []struct {
287 host string
288 txid uint64
289 want uint64
290 }{
291 {"c3.shop.example.com", 1, 1},
292 {"c4.shop.example.com", latest.TXID, latest.TXID},
293 } {
294 body, _ := json.Marshal(map[string]any{"host": c.host, "from": map[string]any{"host": fromSource, "txid": c.txid}})
295 rec := f.copy(t, string(body))
296 requireReady(t, rec)
297 if rec.Seed.TXID != strconv.FormatUint(c.want, 10) {
298 t.Fatalf("--at %d: restored at txid %s, want %d", c.txid, rec.Seed.TXID, c.want)
299 }
300 }
301 code, out := f.create(t, `{"host":"c5.shop.example.com","from":{"host":"main.shop.example.com","txid":999999}}`)
302 want := fmt.Sprintf("main.shop.example.com has replicated up to txid %d, so there is no txid 999999 to copy yet. Leave out --at to copy the latest.", latest.TXID)
303 if code != http.StatusConflict || str(t, out, "error") != want {
304 t.Fatalf("--at past the latest = %d %v, want 409 %q", code, out, want)
305 }
306 f.requireNoRecord(t, "c5.shop.example.com")
307 f.requireClean(t)
308}
309
310func TestFromRefusedBeforeTheFloor(t *testing.T) {
311 f := newFromFixture(t, 10)
312 seedReadyBox(t, f.fx, "kass-1", "", 0, "v0.27.0", "v0.27.0", "nested", false)
313 code, out := f.create(t, fromBody)
314 if code != http.StatusConflict || !strings.Contains(str(t, out, "error"), "below the seeds floor") {
315 t.Fatalf("= %d %v", code, out)
316 }
317 f.requireNoRecord(t, fromHost)
318}
319
320// --from is refused while host claims are armed: replicaDoer has no
321// pinned-version branch, and the seed floor's own claim rule refuses
322// first, before any record is written (Task 4's carried item).
323func TestFromRefusedWithClaimsArmed(t *testing.T) {
324 f := newFromFixture(t, 10)
325 f.fx.srv.cfg.HostClaim = true
326 code, out := f.create(t, fromBody)
327 if code != http.StatusConflict || str(t, out, "error") != "This console runs with claim control on. Seeded instances are not supported here yet." {
328 t.Fatalf("= %d %v", code, out)
329 }
330 f.requireNoRecord(t, fromHost)
331}
332
333func TestFromTooLarge(t *testing.T) {
334 f := newFromFixture(t, 10)
335 old := seedMaxBytes
336 seedMaxBytes = 1024 // the test's database is larger than 1 KiB
337 defer func() { seedMaxBytes = old }()
338 code, out := f.create(t, fromBody)
339 if code != http.StatusRequestEntityTooLarge || str(t, out, "error") != "a copy of main.shop.example.com would be 0.1 GiB, over the 4 GiB limit." {
340 t.Fatalf("= %d %v", code, out)
341 }
342 f.requireNoRecord(t, fromHost)
343 f.requireClean(t)
344}
345
346// SD-N26 rounds up, so a size over the limit never prints as the limit.
347func TestSeedGiBRoundsUp(t *testing.T) {
348 for n, want := range map[int64]string{1: "0.1 GiB", 4 << 30: "4.0 GiB", 4<<30 + 1: "4.1 GiB", 5 << 29: "2.5 GiB"} {
349 if got := seedGiB(n); got != want {
350 t.Errorf("seedGiB(%d) = %q, want %q", n, got, want)
351 }
352 }
353}
354
355func TestFromSourceRefusals(t *testing.T) {
356 f := newFromFixture(t, 3)
357 f.addSource(t, provision.Record{Host: "pooled.shop.example.com", Slug: "pooled", Pool: "shared"}, 0)
358 f.addSource(t, provision.Record{Host: "fleet.shop.example.com", Slug: "fleet", Region: "fleet/racks"}, 0)
359 f.addSource(t, provision.Record{Host: "site.shop.example.com", Slug: "site", Kind: "static"}, 0)
360 f.addSource(t, provision.Record{Host: "far.shop.example.com", Slug: "far", Region: "us"}, 0)
361 f.addSource(t, provision.Record{Host: "empty.shop.example.com", Slug: "empty"}, 0)
362 for _, c := range []struct {
363 source string
364 code int
365 want string
366 }{
367 {"pooled.shop.example.com", http.StatusConflict, "pooled.shop.example.com runs in the pool shared, so the platform keeps no replica of it to copy."},
368 {"fleet.shop.example.com", http.StatusConflict, "fleet.shop.example.com runs on your fleet racks, so the platform keeps no replica of it to copy."},
369 {"site.shop.example.com", http.StatusConflict, "site.shop.example.com is not an instance the platform keeps a replica of, so it cannot be copied."},
370 {"far.shop.example.com", http.StatusConflict, "far.shop.example.com keeps its data in "},
371 {"empty.shop.example.com", http.StatusConflict, "empty.shop.example.com has not replicated anything yet. Wake it once, then try again."},
372 } {
373 code, out := f.create(t, `{"host":"canary.shop.example.com","from":{"host":"`+c.source+`"}}`)
374 if code != c.code || !strings.Contains(str(t, out, "error"), c.want) {
375 t.Errorf("from %s = %d %v, want %d %q", c.source, code, out, c.code, c.want)
376 }
377 }
378 f.requireNoRecord(t, fromHost)
379 f.requireClean(t)
380}
381
382// A copy stays in its source's region: a target homed elsewhere is
383// refused, before any record is written.
384func TestFromRefusesAnotherRegion(t *testing.T) {
385 f := newFromFixture(t, 3)
386 seedReadyBox(t, f.fx, "far-1", "us", 1, "v0.27.0", "v0.27.0", "nested", false)
387 code, out := f.create(t, `{"host":"canary.shop.example.com","region":"us","from":{"host":"main.shop.example.com"}}`)
388 if code != http.StatusConflict || str(t, out, "error") != "a copy stays in its source's region. To run it elsewhere, create it here and then run carlos move." {
389 t.Fatalf("= %d %v", code, out)
390 }
391 f.requireNoRecord(t, fromHost)
392}
393
394func TestFromPlanFailures(t *testing.T) {
395 f := newFromFixture(t, 3)
396 oldPlan := planRestore
397 defer func() { planRestore = oldPlan }()
398 for i, c := range []struct {
399 err error
400 body string
401 code int
402 want string
403 }{
404 {dbdoor.ErrHistoryGone, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, http.StatusConflict, "history before 2 is no longer kept"},
405 {dbdoor.ErrNothingReplicated, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, http.StatusConflict, "history before 2 is no longer kept"},
406 {dbdoor.ErrNothingReplicated, fromBody, http.StatusConflict, "main.shop.example.com has not replicated anything yet. Wake it once, then try again."},
407 {nil, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":9}}`, http.StatusConflict, "main.shop.example.com has replicated up to txid 5, so there is no txid 9 to copy yet. Leave out --at to copy the latest."},
408 {errors.New("dbdoor: the last planned file's header does not describe it"), fromBody, http.StatusBadGateway, fromFailed},
409 {os.ErrNotExist, fromBody, http.StatusBadGateway, fromFailed},
410 } {
411 planRestore = func(context.Context, *dbdoor.Client, uint64) (dbdoor.RestorePlan, error) {
412 if c.err == nil {
413 return dbdoor.RestorePlan{TXID: 5}, nil // the plan clamps a later --at to the latest
414 }
415 return dbdoor.RestorePlan{}, c.err
416 }
417 code, out := f.create(t, c.body)
418 if code != c.code || str(t, out, "error") != c.want {
419 t.Errorf("%d: %v = %d %v, want %d %q", i, c.err, code, out, c.code, c.want)
420 }
421 }
422 f.requireNoRecord(t, fromHost)
423 f.requireClean(t)
424}
425
426// Every failure after the record is reserved aborts it with the sentence
427// the door would have answered, and leaves nothing in scratch or the pool.
428func TestFromRestoreFailuresAbort(t *testing.T) {
429 oldRestore, oldDeadline := restoreToTarget, seedRestoreDeadline
430 defer func() { restoreToTarget, seedRestoreDeadline = oldRestore, oldDeadline }()
431 for i, c := range []struct {
432 restore func(ctx context.Context) error
433 body string
434 reason string
435 }{
436 {func(context.Context) error { return dbdoor.ErrCheckFailed }, fromBody, "the copy of main.shop.example.com failed its integrity check, so no instance was created."},
437 {func(context.Context) error { return dbdoor.ErrHistoryGone }, fromBody, fromAbortedReason},
438 {func(context.Context) error { return dbdoor.ErrHistoryGone }, `{"host":"canary.shop.example.com","from":{"host":"main.shop.example.com","txid":2}}`, "history before 2 is no longer kept"},
439 {func(context.Context) error { return errors.New("bucket read failed") }, fromBody, fromAbortedReason},
440 {func(ctx context.Context) error { <-ctx.Done(); return ctx.Err() }, fromBody, "copying main.shop.example.com took longer than 30 minutes, so no instance was created."},
441 } {
442 f := newFromFixture(t, 3)
443 seedRestoreDeadline = 200 * time.Millisecond
444 restoreToTarget = func(ctx context.Context, _ *dbdoor.Client, parent string, _ uint64) (*dbdoor.Restored, error) {
445 if _, err := os.MkdirTemp(parent, "carlos-db-"); err != nil { // scratch the door must clear
446 return nil, err
447 }
448 return nil, c.restore(ctx)
449 }
450 rec := f.copy(t, c.body)
451 if rec.Seed.State != seeds.StateAborted || rec.Seed.Reason != c.reason {
452 t.Errorf("%d: seed = %+v, want aborted with %q", i, rec.Seed, c.reason)
453 continue
454 }
455 f.requireClean(t)
456 f.requireNothingUploaded(t)
457 }
458}
459
460// The restore is pinned at the planned TXID, whatever the request said:
461// only that restore produces the size that was admitted.
462func TestFromRestoresAtThePlannedTXID(t *testing.T) {
463 f := newFromFixture(t, 5)
464 oldRestore := restoreToTarget
465 defer func() { restoreToTarget = oldRestore }()
466 var got []uint64
467 restoreToTarget = func(ctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
468 got = append(got, target)
469 return oldRestore(ctx, c, parent, target)
470 }
471 plan, err := dbdoor.PlanRestore(context.Background(), dbdoor.NewClient(replicaDoer(lsfile.NewReplicaClient(filepath.Join(f.root, "instances", fromSource)))), 0)
472 if err != nil {
473 t.Fatal(err)
474 }
475 requireReady(t, f.copy(t, fromBody))
476 requireReady(t, f.copy(t, fmt.Sprintf(`{"host":"c2.shop.example.com","from":{"host":"main.shop.example.com","txid":%d}}`, plan.TXID)))
477 if len(got) != 2 || got[0] != plan.TXID || got[1] != plan.TXID {
478 t.Fatalf("restored at %v, want [%d %d]", got, plan.TXID, plan.TXID)
479 }
480}
481
482// A restore that finished larger than planned is refused after the fact
483// (the watcher's tick can come too late for a fast restore).
484func TestFromRefusesARestoreLargerThanPlanned(t *testing.T) {
485 f := newFromFixture(t, 30)
486 oldPlan := planRestore
487 planRestore = func(ctx context.Context, c *dbdoor.Client, target uint64) (dbdoor.RestorePlan, error) {
488 p, err := oldPlan(ctx, c, target)
489 p.RestoredBytes = 1 // admit far less than the restore will write
490 return p, err
491 }
492 defer func() { planRestore = oldPlan }()
493 requireAborted(t, f.copy(t, fromBody), fromGrew)
494 f.requireClean(t)
495 f.requireNothingUploaded(t)
496}
497
498// The watcher cancels a restore that is still running: litestream writes
499// replica.db.tmp and renames it only at the end, so the watcher must see
500// the .tmp file grow.
501func TestFromWatcherCancelsARestoreInFlight(t *testing.T) {
502 f := newFromFixture(t, 3)
503 oldRestore, oldEvery := restoreToTarget, restoreWatchEvery
504 defer func() { restoreToTarget, restoreWatchEvery = oldRestore, oldEvery }()
505 restoreWatchEvery = 20 * time.Millisecond
506 cancelled := make(chan bool, 1)
507 restoreToTarget = func(rctx context.Context, _ *dbdoor.Client, parent string, _ uint64) (*dbdoor.Restored, error) {
508 dir, err := os.MkdirTemp(parent, "carlos-db-")
509 if err != nil {
510 return nil, err
511 }
512 defer os.RemoveAll(dir)
513 out, err := os.Create(filepath.Join(dir, "replica.db.tmp"))
514 if err != nil {
515 return nil, err
516 }
517 defer out.Close()
518 chunk := make([]byte, 64<<10)
519 giveUp := time.After(10 * time.Second)
520 for written := 0; ; {
521 if written < 4<<20 { // well past the database's planned size, and bounded
522 n, _ := out.Write(chunk)
523 written += n
524 }
525 select {
526 case <-rctx.Done():
527 cancelled <- true
528 return nil, rctx.Err()
529 case <-giveUp:
530 cancelled <- false
531 return nil, errors.New("the watcher never cancelled the restore")
532 case <-time.After(10 * time.Millisecond):
533 }
534 }
535 }
536 rec := f.copy(t, fromBody)
537 if !<-cancelled {
538 t.Fatal("the restore ran to its own timeout: the watcher did not cancel it in flight")
539 }
540 requireAborted(t, rec, fromGrew)
541 f.requireClean(t)
542}
543
544// Once the door has answered 202 the copy no longer depends on the
545// request: the request's context is cancelled when the handler returns,
546// and the client then cancelling and closing its connection changes
547// nothing. The copy finishes, ready or aborted, and cleans up after itself
548// either way. (A disconnect before the 202 is a refusal like any other.)
549func TestFromCopyOutlivesItsClient(t *testing.T) {
550 for _, c := range []struct {
551 name string
552 fail error
553 reason string
554 }{
555 {"ready", nil, ""},
556 {"aborted", dbdoor.ErrCheckFailed, "the copy of main.shop.example.com failed its integrity check, so no instance was created."},
557 } {
558 t.Run(c.name, func(t *testing.T) {
559 f := newFromFixture(t, 3)
560 oldRestore := restoreToTarget
561 defer func() { restoreToTarget = oldRestore }()
562 started, proceed := make(chan struct{}), make(chan struct{})
563 restoreToTarget = func(rctx context.Context, cl *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
564 close(started)
565 <-proceed
566 if rctx.Err() != nil {
567 return nil, rctx.Err()
568 }
569 if c.fail != nil {
570 return nil, c.fail
571 }
572 return oldRestore(rctx, cl, parent, target)
573 }
574 tr := &http.Transport{}
575 client := &http.Client{Transport: tr}
576 ctx, cancel := context.WithCancel(context.Background())
577 req, _ := http.NewRequestWithContext(ctx, http.MethodPost, f.url, strings.NewReader(fromBody))
578 req.Header.Set("Authorization", "Bearer "+f.token)
579 req.Header.Set("Content-Type", "application/json")
580 res, err := client.Do(req)
581 if err != nil {
582 t.Fatal(err)
583 }
584 res.Body.Close()
585 if res.StatusCode != http.StatusAccepted {
586 t.Fatalf("= %d, want 202", res.StatusCode)
587 }
588 <-started
589 cancel() // the client goes away, and its connection with it
590 tr.CloseIdleConnections()
591 close(proceed)
592 f.fx.srv.seedCopies.Wait()
593 rec, err := f.record(t, fromHost)
594 if err != nil {
595 t.Fatal(err)
596 }
597 if c.fail == nil {
598 requireReady(t, rec)
599 } else {
600 requireAborted(t, rec, c.reason)
601 }
602 f.requireClean(t)
603 })
604 }
605}
606
607// A panic in the detached copy aborts its seed and leaves the console
608// serving: nothing above the copy's goroutine would contain it.
609func TestFromPanicAbortsTheSeedAndSparesTheConsole(t *testing.T) {
610 f := newFromFixture(t, 3)
611 oldRestore := restoreToTarget
612 defer func() { restoreToTarget = oldRestore }()
613 restoreToTarget = func(ctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
614 if _, err := os.MkdirTemp(parent, "carlos-db-"); err != nil { // scratch the door must clear
615 return nil, err
616 }
617 panic("a decoder fault")
618 }
619 requireAborted(t, f.copy(t, fromBody), fromAbortedReason)
620 f.requireClean(t)
621 f.requireNothingUploaded(t)
622 // Still serving: the next copy runs to ready.
623 restoreToTarget = oldRestore
624 requireReady(t, f.copy(t, `{"host":"c2.shop.example.com","from":{"host":"main.shop.example.com"}}`))
625 f.requireClean(t)
626}
627
628func TestFromUploadFailureAborts(t *testing.T) {
629 f := newFromFixture(t, 3)
630 f.fx.srv.cfg.SeedData = failingPuts{f.data}
631 requireAborted(t, f.copy(t, fromBody), fromAbortedReason)
632 f.requireClean(t)
633}
634
635type failingPuts struct{ release.Store }
636
637func (failingPuts) Put(context.Context, string, []byte, string) (string, error) {
638 return "", errors.New("access denied")
639}
640
641// A publish that fails aborts the seed, with the retry sentence.
642func TestFromPublishFailureAborts(t *testing.T) {
643 f := newFromFixture(t, 3)
644 rel := &failOnceOn{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, fromHost)}
645 f.fx.srv.cfg.Releases = rel
646 oldRestore := restoreToTarget
647 defer func() { restoreToTarget = oldRestore }()
648 restoreToTarget = func(ctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
649 rel.armed.Store(true) // the next write of the record is the publish
650 return oldRestore(ctx, c, parent, target)
651 }
652 requireAborted(t, f.copy(t, fromBody), fromAbortedReason)
653 f.requireClean(t)
654}
655
656// failOnceOn fails the first Put of key once armed.
657type failOnceOn struct {
658 release.Store
659 key string
660 armed atomic.Bool
661}
662
663func (s *failOnceOn) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
664 if key == s.key && s.armed.CompareAndSwap(true, false) {
665 return "", errors.New("the bucket refused the write")
666 }
667 return s.Store.Put(ctx, key, body, ifMatch)
668}
669
670// A record that appears between the create door's checks and this door's
671// reservation is answered as the ordinary create answers it, and leaves
672// nothing behind.
673func TestFromReservationLosesARace(t *testing.T) {
674 f := newFromFixture(t, 3)
675 oldPlan := planRestore
676 defer func() { planRestore = oldPlan }()
677 planRestore = func(ctx context.Context, c *dbdoor.Client, target uint64) (dbdoor.RestorePlan, error) {
678 // Another writer's record, written from the handler's goroutine:
679 // the test checks it below rather than failing here.
680 _ = provision.CreateRecord(ctx, f.fx.rel, provision.Record{Schema: 1, Host: fromHost, Slug: "canary",
681 App: fromApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()})
682 return oldPlan(ctx, c, target)
683 }
684 code, out := f.create(t, fromBody)
685 if code != http.StatusConflict || str(t, out, "error") != "instance canary.shop.example.com already exists" {
686 t.Fatalf("= %d %v", code, out)
687 }
688 if rec, err := f.record(t, fromHost); err != nil || rec.Seed != nil {
689 t.Fatalf("record = %+v, %v; want the other writer's, unseeded", rec, err)
690 }
691 f.requireClean(t)
692}
693
694func TestFromNeedsADataBucket(t *testing.T) {
695 f := newFromFixture(t, 3)
696 f.fx.srv.cfg.SeedData = nil
697 code, out := f.create(t, fromBody)
698 if code != http.StatusServiceUnavailable || str(t, out, "error") != "this console has no data bucket set up, so it cannot create a seeded instance." {
699 t.Fatalf("= %d %v", code, out)
700 }
701 f.requireNoRecord(t, fromHost)
702}
703
704func TestFromRefusesAPooledApp(t *testing.T) {
705 f := newFromFixture(t, 3)
706 m := fromManifest(false)
707 m.Pool = "shared"
708 enableInstances(t, f.fx, fromApp, m)
709 code, out := f.create(t, fromBody)
710 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." {
711 t.Fatalf("= %d %v", code, out)
712 }
713 f.requireNoRecord(t, fromHost)
714}
715
716// The instance service role reaches --from through the create door's own
717// route, as spec §6.1 says; the db door stays person-only.
718func TestFromByAnInstanceServiceRole(t *testing.T) {
719 f := newFromFixture(t, 3)
720 svc := serviceToken(t, f.fx.st, OpsAccountID, "ci", fromApp, svcrole.Instance)
721 rec := f.copyAs(t, svc, fromBody)
722 requireReady(t, rec)
723 if rec.Seed.By != "service:bab/ci" {
724 t.Fatalf("record seed by %q, want the service principal", rec.Seed.By)
725 }
726}
727
728func TestFromPrivateStorageRules(t *testing.T) {
729 f := newFromFixture(t, 3)
730 enableInstances(t, f.fx, fromApp, fromManifest(true))
731 code, out := f.create(t, `{"host":"canary.shop.example.com","environment":"main","from":{"host":"main.shop.example.com"}}`)
732 if code != http.StatusConflict || str(t, out, "error") != "a seeded instance gets its own environment and an empty store" {
733 t.Fatalf("explicit environment = %d %v", code, out)
734 }
735 // A store already provisioned for the derived environment, by no seed.
736 at := time.Now().UTC().Add(-time.Hour)
737 putCanaryStoreOf(t, f.fx, fromApp, &objstore.Record{
738 Declared: &objstore.Declared{EnvPrefix: "S3", At: at, By: "bob@keymail.dev"},
739 Grant: &objstore.GrantBlock{Backend: "mem", Bucket: "carlos-bab-shop-canary", Source: objstore.GrantSourceAccountApproval, At: at, By: "operator@keymail.dev"},
740 })
741 code, out = f.create(t, fromBody)
742 if code != http.StatusConflict || str(t, out, "error") != "environment canary already has a store (carlos-bab-shop-canary). A seeded instance needs a new, empty one." {
743 t.Fatalf("an existing store = %d %v", code, out)
744 }
745 f.requireNoRecord(t, fromHost)
746 f.requireClean(t)
747}
748
749// The store gates that write nothing refuse before the record is reserved
750// (ruling R3), with the store plane's own answers.
751func TestFromStoreGatesRefuseBeforeTheReservation(t *testing.T) {
752 for name, c := range map[string]struct {
753 unwire func(*Server)
754 want string
755 }{
756 "no backend": {func(s *Server) { s.cfg.StoreProvisioner = nil },
757 "object storage is not configured for this deployment (CARLOS_STORE_BACKEND), and shop gives every instance its own storage — no instance can be created until one is wired or the policy is turned off (carlos instances enable --private-storage=false)"},
758 "no sealing key": {func(s *Server) { s.cfg.SecretsPub = "" },
759 "this console has no secrets key (CARLOS_SECRETS_PUBKEY) — it cannot deliver a store credential, so it will not mint one"},
760 } {
761 t.Run(name, func(t *testing.T) {
762 f := newFromFixture(t, 3)
763 enableInstances(t, f.fx, fromApp, fromManifest(true))
764 c.unwire(f.fx.srv)
765 code, out := f.create(t, fromBody)
766 if code != http.StatusServiceUnavailable || str(t, out, "error") != c.want {
767 t.Fatalf("= %d %v, want 503 %q", code, out, c.want)
768 }
769 f.requireNoRecord(t, fromHost)
770 f.requireClean(t)
771 })
772 }
773}
774
775// A store walk refused after the reservation aborts the seed with the walk's
776// own sentence as its reason.
777func TestFromStoreWalkRefusalAborts(t *testing.T) {
778 f := newFromFixture(t, 3)
779 enableInstances(t, f.fx, fromApp, fromManifest(true))
780 takeCanaryBucketName(t, f.fx)
781 rec := f.copy(t, fromBody)
782 requireAborted(t, rec, "another store already uses this instance's storage names (app and environment names are joined with \"-\"), so the instance was not created. use a different host name")
783 f.requireClean(t)
784 f.requireNothingUploaded(t)
785}
786
787func putCanaryStoreOf(t *testing.T, fx *consoleFixture, app string, rec *objstore.Record) {
788 t.Helper()
789 if _, err := objstore.PutIn(context.Background(), fx.rel, release.OpsAccount, app, "canary", rec, ""); err != nil {
790 t.Fatal(err)
791 }
792}
793
794// With private storage the copy gets a new store marked with its seed, and
795// the source's own store is not read, bound or touched.
796func TestFromPrivateStorageGetsANewStoreMarkedWithTheSeed(t *testing.T) {
797 f := newFromFixture(t, 3)
798 enableInstances(t, f.fx, fromApp, fromManifest(true))
799 at := time.Now().UTC().Add(-time.Hour)
800 if _, err := objstore.PutIn(context.Background(), f.fx.rel, release.OpsAccount, fromApp, "main", &objstore.Record{
801 Declared: &objstore.Declared{EnvPrefix: "S3", At: at, By: "bob@keymail.dev"},
802 Grant: &objstore.GrantBlock{Backend: "mem", Bucket: "carlos-bab-shop-main", Source: objstore.GrantSourceAccountApproval, At: at, By: "operator@keymail.dev"},
803 }, ""); err != nil {
804 t.Fatal(err)
805 }
806 mainKey := objstore.KeyIn(release.OpsAccount, fromApp, "main")
807 before, _, err := f.fx.rel.Get(context.Background(), mainKey)
808 if err != nil {
809 t.Fatal(err)
810 }
811 rec := f.copy(t, fromBody)
812 requireReady(t, rec)
813 if rec.Environment != "canary" {
814 t.Fatalf("record environment = %q, want the host's own", rec.Environment)
815 }
816 store, _, err := objstore.GetIn(context.Background(), f.fx.rel, release.OpsAccount, fromApp, "canary")
817 if err != nil || store == nil || store.Declared == nil || store.Declared.Seed != rec.Seed.ID || storeBucketName(store) == "" {
818 t.Fatalf("canary store = %+v, %v; want one marked with seed %s", store, err, rec.Seed.ID)
819 }
820 after, _, err := f.fx.rel.Get(context.Background(), mainKey)
821 if err != nil || string(after) != string(before) {
822 t.Fatalf("the source's store record changed:\n before %s\n after %s (%v)", before, after, err)
823 }
824 f.requireClean(t)
825}
826
827// The create door's sibling-apex re-check holds on the long --from path
828// too: an app created under the host's label while the copy ran owns its
829// apex, and the seed is aborted rather than published.
830func TestFromAbortsWhenASiblingTakesTheApex(t *testing.T) {
831 f := newFromFixture(t, 3)
832 f.fx.srv.cfg.AppsDomain = "apps.example.net"
833 m := fromManifest(false)
834 m.Domain = "bab.apps.example.net"
835 enableInstances(t, f.fx, fromApp, m)
836 f.addSource(t, provision.Record{Host: "main.bab.apps.example.net", Slug: "main"}, 3)
837 oldRestore := restoreToTarget
838 defer func() { restoreToTarget = oldRestore }()
839 restoreToTarget = func(ctx context.Context, c *dbdoor.Client, parent string, target uint64) (*dbdoor.Restored, error) {
840 if _, err := f.fx.st.CreateAppIn(OpsAccountID, "canary", "paul@keymail.dev", time.Now()); err != nil {
841 return nil, err
842 }
843 return oldRestore(ctx, c, parent, target)
844 }
845 rec := f.copy(t, `{"host":"canary.bab.apps.example.net","from":{"host":"main.bab.apps.example.net"}}`)
846 requireAborted(t, rec, "canary.bab.apps.example.net is app canary's platform apex. That app was created while this instance was being written, so the instance was not kept.")
847 f.requireClean(t)
848}
849
850// A door that held no answer never sends an empty 200 (ruling R1).
851func TestHeldResponseNeverAnswersEmpty(t *testing.T) {
852 w := httptest.NewRecorder()
853 (&heldResponse{header: w.Header()}).flush(w, fromFailed)
854 var out map[string]string
855 if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil || w.Code != http.StatusBadGateway || out["error"] != fromFailed {
856 t.Fatalf("= %d %s, want 502 %q", w.Code, w.Body, fromFailed)
857 }
858}
859
860func TestRestoreScratchUse(t *testing.T) {
861 dir := t.TempDir()
862 sub := filepath.Join(dir, "carlos-db-x")
863 if err := os.Mkdir(sub, 0o700); err != nil {
864 t.Fatal(err)
865 }
866 for name, n := range map[string]int{"replica.db.tmp": 100, "ltx-1": 40, "ltx-2": 60, "replica.db-shm": 32768} {
867 if err := os.WriteFile(filepath.Join(sub, name), make([]byte, n), 0o600); err != nil {
868 t.Fatal(err)
869 }
870 }
871 db, total := restoreScratchUse(dir)
872 if db != 100 || total != 200 {
873 t.Fatalf("restoreScratchUse = %d, %d; want 100, 200 (SQLite's own -shm is not the restore's)", db, total)
874 }
875}
876