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
3// The leftover-replica rail on the two create doors (discussion #199). A
4// deleted instance or sidecar leaves its replica at instances/<host> and any
5// move staging at moves/<host>/, and both are keyed by HOST alone, so the next
6// record created on that host would inherit the old database on its first
7// wake or through `carlos db pull`. Until the full fix lands, both doors
8// refuse a create on a host whose replica or staging prefix in this
9// console's data bucket still holds an LTX file. These tests use real
10// litestream replicas on disk behind the db door's replica seam, so "holds an
11// LTX file" means exactly what the activator would restore from.
12
13import (
14 "context"
15 "encoding/xml"
16 "errors"
17 "fmt"
18 "io"
19 "net/http"
20 "net/http/httptest"
21 "os"
22 "path/filepath"
23 "strings"
24 "sync"
25 "sync/atomic"
26 "testing"
27 "time"
28
29 lsfile "github.com/benbjohnson/litestream/file"
30 "github.com/superfly/ltx"
31
32 "github.com/carlosframework/platform/internal/activator"
33 "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
34 "github.com/carlosframework/platform/internal/provision"
35 "github.com/carlosframework/platform/internal/release"
36)
37
38type leftoverFx struct {
39 *consoleFixture
40 token string
41 root string // the fake data bucket: one file replica per prefix
42 opens *atomic.Int32 // how many times a door reached for a replica client
43}
44
45// leftoverFixture is a console in region ie with a data bucket configured,
46// and titogo enabled for both instances and sidecars in ie. Neither manifest
47// names a pool: a pooled record's data is on the pool box, which the rail
48// skips (TestCreateSkipsTheCheckForPooledRecords).
49func leftoverFixture(t *testing.T, code string) *leftoverFx {
50 t.Helper()
51 fx, token := instanceFixture(t, code)
52 im := titoManifest()
53 im.Pool = ""
54 enableInstances(t, fx, "titogo", im)
55 sm := titoSidecarManifest()
56 sm.Pool = ""
57 enableSidecars(t, fx, "titogo", sm)
58 root := t.TempDir()
59 fx.srv.cfg.HibernateBucket = "data-test"
60 fx.srv.cfg.HibernatePrefix = "instances"
61 fx.srv.cfg.Region = "ie"
62 opens := new(atomic.Int32)
63 fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
64 opens.Add(1)
65 return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil
66 }
67 return &leftoverFx{consoleFixture: fx, token: token, root: root, opens: opens}
68}
69
70// leaveReplica writes a real litestream replica where a deleted record's
71// activator left it.
72func (fx *leftoverFx) leaveReplica(t *testing.T, host string) {
73 t.Helper()
74 dbdoortest.Replicate(t, filepath.Join(fx.root, activator.HibernatePrefix("instances", host)), 1)
75}
76
77// leaveStaging writes one where an abandoned move staged it.
78func (fx *leftoverFx) leaveStaging(t *testing.T, host string) {
79 t.Helper()
80 dbdoortest.Replicate(t, filepath.Join(fx.root, activator.MovesPrefix(host)), 1)
81}
82
83// createDoor is one of the two doors, reduced to what the matrix needs.
84type createDoor struct {
85 name string
86 url func(fx *consoleFixture) string
87 body string // a create the door accepts on a clean host, in region ie
88 host string // the host that create claims
89 exists func(t *testing.T, fx *consoleFixture) bool
90}
91
92var leftoverDoors = []createDoor{
93 {
94 name: "instance",
95 url: func(fx *consoleFixture) string { return instancesURL(fx, "bab", "titogo") },
96 body: `{"host":"acme.go.tito.io","region":"ie"}`,
97 host: "acme.go.tito.io",
98 exists: func(t *testing.T, fx *consoleFixture) bool {
99 t.Helper()
100 _, err := provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "acme.go.tito.io")
101 if err != nil && !errors.Is(err, os.ErrNotExist) {
102 t.Fatal(err)
103 }
104 return err == nil
105 },
106 },
107 {
108 name: "sidecar",
109 url: func(fx *consoleFixture) string { return sidecarsURL(fx, "bab", "titogo") },
110 body: `{"name":"bot","region":"ie"}`,
111 host: botHost,
112 exists: func(t *testing.T, fx *consoleFixture) bool {
113 t.Helper()
114 _, err := provision.GetSidecarRecord(context.Background(), fx.rel, "bab", "titogo", "bot")
115 if err != nil && !errors.Is(err, os.ErrNotExist) {
116 t.Fatal(err)
117 }
118 return err == nil
119 },
120 },
121}
122
123// A host whose replica prefix still holds LTX files is refused with 409,
124// and the refusal writes no record.
125func TestCreateRefusesLeftoverReplica(t *testing.T) {
126 for _, d := range leftoverDoors {
127 t.Run(d.name, func(t *testing.T) {
128 fx := leftoverFixture(t, "code-left-rep-"+d.name)
129 fx.leaveReplica(t, d.host)
130 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
131 if status != http.StatusConflict || !strings.Contains(out, "still holds data from a deleted") {
132 t.Fatalf("create = %d %s, want the leftover-data 409", status, out)
133 }
134 if d.exists(t, fx.consoleFixture) {
135 t.Fatal("a refused create wrote a record anyway")
136 }
137 })
138 }
139}
140
141// Move staging counts too: a host whose only leftover is an abandoned
142// moves/<host>/ is refused the same way, because a landing promotes that
143// staging into the instance's prefix.
144func TestCreateRefusesLeftoverMoveStaging(t *testing.T) {
145 for _, d := range leftoverDoors {
146 t.Run(d.name, func(t *testing.T) {
147 fx := leftoverFixture(t, "code-left-mov-"+d.name)
148 fx.leaveStaging(t, d.host)
149 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
150 if status != http.StatusConflict || !strings.Contains(out, "still holds data from a deleted") {
151 t.Fatalf("create = %d %s, want the leftover-data 409", status, out)
152 }
153 if d.exists(t, fx.consoleFixture) {
154 t.Fatal("a refused create wrote a record anyway")
155 }
156 })
157 }
158}
159
160// A clean host is created as before, and the check really looked: both
161// prefixes were opened.
162func TestCreateAllowsACleanHost(t *testing.T) {
163 for _, d := range leftoverDoors {
164 t.Run(d.name, func(t *testing.T) {
165 fx := leftoverFixture(t, "code-left-clean-"+d.name)
166 // Another host's leftovers are not this host's business.
167 fx.leaveReplica(t, "other.go.tito.io")
168 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
169 if status != http.StatusCreated {
170 t.Fatalf("create = %d %s, want 201", status, out)
171 }
172 if !d.exists(t, fx.consoleFixture) {
173 t.Fatal("the create answered 201 and wrote no record")
174 }
175 if n := fx.opens.Load(); n != 2 {
176 t.Fatalf("replica clients opened = %d, want 2 (the replica prefix and the staging prefix)", n)
177 }
178 })
179 }
180}
181
182// A record homed in another region is not checked: its data lives in that
183// region's bucket, which this console cannot read. The gap is documented at
184// the rail; this pins that the rail does not refuse what it cannot see.
185func TestCreateSkipsTheCheckInAnotherRegion(t *testing.T) {
186 for _, d := range leftoverDoors {
187 t.Run(d.name, func(t *testing.T) {
188 fx := leftoverFixture(t, "code-left-region-"+d.name)
189 fx.srv.cfg.Region = "uk"
190 fx.leaveReplica(t, d.host)
191 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
192 if status != http.StatusCreated {
193 t.Fatalf("create = %d %s, want 201 (unchecked)", status, out)
194 }
195 if n := fx.opens.Load(); n != 0 {
196 t.Fatalf("replica clients opened = %d for an out-of-region record, want 0", n)
197 }
198 })
199 }
200}
201
202// A console with no data bucket creates as it always did, and never reaches
203// for a replica client.
204func TestCreateSkipsTheCheckWithNoDataBucket(t *testing.T) {
205 for _, d := range leftoverDoors {
206 t.Run(d.name, func(t *testing.T) {
207 fx := leftoverFixture(t, "code-left-nobucket-"+d.name)
208 fx.srv.cfg.HibernateBucket = ""
209 fx.leaveReplica(t, d.host)
210 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
211 if status != http.StatusCreated {
212 t.Fatalf("create = %d %s, want 201 (unchecked)", status, out)
213 }
214 if n := fx.opens.Load(); n != 0 {
215 t.Fatalf("replica clients opened = %d with no bucket, want 0", n)
216 }
217 })
218 }
219}
220
221// failingReplica fails every listing, with a message carrying a bucket name
222// the way a real S3 error does.
223type failingReplica struct{}
224
225func (failingReplica) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
226 return nil, errors.New("list secret-bucket: connection reset")
227}
228
229func (failingReplica) OpenLTXFile(context.Context, int, ltx.TXID, ltx.TXID, int64, int64) (io.ReadCloser, error) {
230 return nil, errors.New("not used")
231}
232
233// erroringIterator fails on the first Next, the way the S3 client surfaces
234// a paging error.
235type erroringIterator struct{}
236
237func (erroringIterator) Close() error { return nil }
238func (erroringIterator) Next() bool { return false }
239func (erroringIterator) Err() error { return errors.New("page secret-bucket: timeout") }
240func (erroringIterator) Item() *ltx.FileInfo { return nil }
241
242type pagingFailReplica struct{ failingReplica }
243
244func (pagingFailReplica) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
245 return erroringIterator{}, nil
246}
247
248// A check that cannot complete refuses with 503: never create on
249// uncertainty. The store's own error, which names the bucket, stays in the
250// log.
251func TestCreateRefusesWhenTheCheckFails(t *testing.T) {
252 seams := map[string]func(context.Context, string) (replicaReader, error){
253 "client": func(context.Context, string) (replicaReader, error) {
254 return nil, errors.New("resolve region for secret-bucket: no credentials")
255 },
256 "listing": func(context.Context, string) (replicaReader, error) { return failingReplica{}, nil },
257 "paging": func(context.Context, string) (replicaReader, error) { return pagingFailReplica{}, nil },
258 }
259 for _, d := range leftoverDoors {
260 for seam, open := range seams {
261 t.Run(d.name+"/"+seam, func(t *testing.T) {
262 fx := leftoverFixture(t, "code-left-fail-"+d.name+"-"+seam)
263 fx.srv.dbReplica = open
264 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
265 if status != http.StatusServiceUnavailable || !strings.Contains(out, "could not check whether") {
266 t.Fatalf("create = %d %s, want the 503 that refuses on uncertainty", status, out)
267 }
268 if strings.Contains(out, "secret-bucket") || strings.Contains(out, "data-test") {
269 t.Fatalf("the refusal leaked a bucket name: %s", out)
270 }
271 if d.exists(t, fx.consoleFixture) {
272 t.Fatal("a create that could not be checked wrote a record anyway")
273 }
274 })
275 }
276 }
277}
278
279// Every existing validation answers first: a create the door would refuse
280// anyway never reaches for the bucket, and keeps its own answer.
281func TestCreateValidatesBeforeTheCheck(t *testing.T) {
282 // consoleRegion is the region the rail would check, so a check run
283 // too early would reach for the bucket rather than skip.
284 cases := []struct{ door, url, body, consoleRegion string }{
285 {"instance", "instances", `{"host":"acme.elsewhere.io","region":"ie"}`, "ie"},
286 {"instance", "instances", `{"host":"acme.go.tito.io","region":"mars"}`, "mars"},
287 {"sidecar", "sidecars", `{"name":"../bot","region":"ie"}`, "ie"},
288 {"sidecar", "sidecars", `{"name":"bot","region":"mars"}`, "mars"},
289 }
290 for _, c := range cases {
291 t.Run(c.door+" "+c.body, func(t *testing.T) {
292 fx := leftoverFixture(t, "code-left-order")
293 fx.srv.cfg.Region = c.consoleRegion
294 fx.srv.dbReplica = func(context.Context, string) (replicaReader, error) {
295 fx.opens.Add(1)
296 return failingReplica{}, nil
297 }
298 u := fx.ts.URL + "/api/cli/apps/bab/titogo/" + c.url
299 status, _, out := rawReq(t, http.MethodPost, u, fx.token, c.body)
300 if status != http.StatusBadRequest {
301 t.Fatalf("create = %d %s, want the door's own 400", status, out)
302 }
303 if n := fx.opens.Load(); n != 0 {
304 t.Fatalf("an invalid create reached for the bucket %d time(s)", n)
305 }
306 })
307 }
308}
309
310// Paths that create no record are untouched: deleting a record whose host
311// holds a replica (the ordinary case) never consults the rail.
312func TestDeleteNeverConsultsTheLeftoverCheck(t *testing.T) {
313 for _, d := range leftoverDoors {
314 t.Run(d.name, func(t *testing.T) {
315 fx := leftoverFixture(t, "code-left-del-"+d.name)
316 if status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body); status != http.StatusCreated {
317 t.Fatalf("create = %d %s", status, out)
318 }
319 fx.leaveReplica(t, d.host)
320 fx.opens.Store(0)
321 target := d.url(fx.consoleFixture) + "/" + d.host
322 if d.name == "sidecar" {
323 target = d.url(fx.consoleFixture) + "/bot"
324 }
325 if status, _, out := rawReq(t, http.MethodDelete, target, fx.token, ""); status >= 300 {
326 t.Fatalf("delete = %d %s, want success", status, out)
327 }
328 if n := fx.opens.Load(); n != 0 {
329 t.Fatalf("a delete reached for the bucket %d time(s)", n)
330 }
331 })
332 }
333}
334
335// The residency rules the rail applies are resolveReplica's: in-region
336// only, a pooled record's data is on the pool box, and a fleet placement is
337// never this console's to read, whatever its own region string says.
338func TestLeftoverCheckResidency(t *testing.T) {
339 s := &Server{}
340 s.cfg.HibernateBucket = "data-test"
341 s.cfg.Region = "fleet/acme"
342 for _, c := range []struct {
343 region, pool, console string
344 bucket string
345 want bool
346 }{
347 {"", "", "", "data-test", true},
348 {"ie", "", "ie", "data-test", true},
349 {"uk", "", "ie", "data-test", false},
350 {"", "", "ie", "data-test", false},
351 {"fleet/acme", "", "fleet/acme", "data-test", false},
352 {"ie", "", "ie", "", false},
353 {"ie", "tito", "ie", "data-test", false},
354 {"", "tito", "", "data-test", false},
355 } {
356 s.cfg.Region, s.cfg.HibernateBucket = c.console, c.bucket
357 if got := s.leftoverCheckApplies(c.region, c.pool); got != c.want {
358 t.Errorf("record %q pool %q on console %q (bucket %q) = %v, want %v", c.region, c.pool, c.console, c.bucket, got, c.want)
359 }
360 }
361}
362
363// ---- review round 1 (Astra, 6cc86d34) -----------------------------------------
364
365// fakeDataBucket is a ListObjectsV2 endpoint with real S3 prefix semantics,
366// for the PRODUCTION replica client (lease.ReadReplicaClient through
367// cachedReadReplica). The file replica the other tests use joins paths with
368// filepath.Join, which forgives a doubled slash that S3 does not; this is
369// where the exact keys the box writes, and the exact prefixes the console
370// lists, meet.
371type fakeDataBucket struct {
372 mu sync.Mutex
373 keys []string
374 prefixes []string
375}
376
377func (b *fakeDataBucket) ServeHTTP(w http.ResponseWriter, r *http.Request) {
378 q := r.URL.Query()
379 if r.Method != http.MethodGet || q.Get("list-type") != "2" {
380 http.Error(w, "the leftover check only lists", http.StatusMethodNotAllowed)
381 return
382 }
383 prefix := q.Get("prefix")
384 b.mu.Lock()
385 b.prefixes = append(b.prefixes, prefix)
386 var keys []string
387 for _, k := range b.keys {
388 if strings.HasPrefix(k, prefix) {
389 keys = append(keys, k)
390 }
391 }
392 b.mu.Unlock()
393 type obj struct {
394 Key string
395 Size int64
396 LastModified string
397 }
398 type result struct {
399 XMLName xml.Name `xml:"ListBucketResult"`
400 Name string
401 Prefix string
402 KeyCount int
403 IsTruncated bool
404 Contents []obj
405 }
406 res := result{Name: "data-test", Prefix: prefix, KeyCount: len(keys)}
407 for _, k := range keys {
408 res.Contents = append(res.Contents, obj{Key: k, Size: 100, LastModified: "2026-10-01T00:00:00Z"})
409 }
410 w.Header().Set("Content-Type", "application/xml")
411 _ = xml.NewEncoder(w).Encode(res)
412}
413
414func (b *fakeDataBucket) listed() []string {
415 b.mu.Lock()
416 defer b.mu.Unlock()
417 return append([]string(nil), b.prefixes...)
418}
419
420// ltxKey is the key litestream's S3 client writes one LTX file under:
421// <path>/<level as %04x>/<filename>.
422func ltxKey(prefix string, level int) string {
423 return fmt.Sprintf("%s/%04x/%s", strings.TrimSuffix(prefix, "/"), level, ltx.FormatFilename(1, 1))
424}
425
426// useFakeDataBucket points s at a fake S3 endpoint holding keys, through the
427// production replica client.
428func useFakeDataBucket(t *testing.T, s *Server, keys ...string) *fakeDataBucket {
429 t.Helper()
430 b := &fakeDataBucket{keys: keys}
431 srv := httptest.NewServer(b)
432 t.Cleanup(srv.Close)
433 t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
434 t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
435 t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
436 t.Setenv("AWS_SESSION_TOKEN", "")
437 t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
438 s.cfg.HibernateBucket = "data-test"
439 s.cfg.HibernateRegion = "us-east-1"
440 s.cfg.HibernatePrefix = "instances"
441 s.dbReplica = s.cachedReadReplica
442 return b
443}
444
445// The production client is asked for exactly the prefixes the box writes:
446// instances/<host>/<level>/ and moves/<host>/<level>/, levels 0 to 9, with
447// one slash between each segment. A doubled slash lists nothing on S3.
448func TestLeftoverCheckListsTheExactS3Prefixes(t *testing.T) {
449 s := &Server{}
450 b := useFakeDataBucket(t, s)
451 held, err := s.hostHoldsReplicaData(context.Background(), "app.example.com")
452 if err != nil || held {
453 t.Fatalf("empty bucket = %v, %v; want false, nil", held, err)
454 }
455 var want []string
456 for _, root := range []string{"instances/app.example.com", "moves/app.example.com"} {
457 for level := 0; level <= 9; level++ {
458 want = append(want, fmt.Sprintf("%s/%04x/", root, level))
459 }
460 }
461 if got := b.listed(); strings.Join(got, "\n") != strings.Join(want, "\n") {
462 t.Fatalf("listed prefixes:\n%s\nwant:\n%s", strings.Join(got, "\n"), strings.Join(want, "\n"))
463 }
464}
465
466// Data at any level counts, the snapshot level included, in either prefix;
467// and a host is never answered with another host's data, even one its name
468// is a string prefix of.
469func TestLeftoverCheckThroughTheS3Client(t *testing.T) {
470 const host = "app.example.co"
471 for _, c := range []struct {
472 name string
473 keys []string
474 want bool
475 }{
476 {"replica at L0", []string{ltxKey("instances/"+host, 0)}, true},
477 {"snapshot only (L9)", []string{ltxKey("instances/"+host, 9)}, true},
478 {"move staging at L0", []string{ltxKey(activator.MovesPrefix(host), 0)}, true},
479 {"move staging snapshot only (L9)", []string{ltxKey(activator.MovesPrefix(host), 9)}, true},
480 {"only a longer host's data", []string{
481 ltxKey("instances/app.example.com", 0), ltxKey("instances/app.example.com", 9),
482 ltxKey(activator.MovesPrefix("app.example.com"), 0),
483 }, false},
484 {"a lock file and nothing else", []string{"instances/" + host + "/lock.json"}, false},
485 } {
486 t.Run(c.name, func(t *testing.T) {
487 s := &Server{}
488 useFakeDataBucket(t, s, c.keys...)
489 held, err := s.hostHoldsReplicaData(context.Background(), host)
490 if err != nil || held != c.want {
491 t.Fatalf("hostHoldsReplicaData = %v, %v; want %v, nil", held, err, c.want)
492 }
493 })
494 }
495}
496
497// End to end through the production client: move staging the box wrote at
498// moves/<host>/0000/ refuses both doors.
499func TestCreateRefusesMoveStagingThroughTheS3Client(t *testing.T) {
500 for _, d := range leftoverDoors {
501 t.Run(d.name, func(t *testing.T) {
502 fx := leftoverFixture(t, "code-left-s3-"+d.name)
503 useFakeDataBucket(t, fx.srv, ltxKey(activator.MovesPrefix(d.host), 0))
504 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
505 if status != http.StatusConflict || !strings.Contains(out, "still holds data from a deleted") {
506 t.Fatalf("create = %d %s, want the leftover-data 409", status, out)
507 }
508 if d.exists(t, fx.consoleFixture) {
509 t.Fatal("a refused create wrote a record anyway")
510 }
511 })
512 }
513}
514
515// writeDeadlineStore records the deadline of the context the record write
516// ran under, and can fail reads of that record.
517type writeDeadlineStore struct {
518 release.Store
519 key string
520 failGet bool
521 mu sync.Mutex
522 deadline time.Time
523 puts int
524}
525
526func (s *writeDeadlineStore) Get(ctx context.Context, key string) ([]byte, string, error) {
527 if s.failGet && key == s.key {
528 return nil, "", errors.New("get secret-bucket: connection reset")
529 }
530 return s.Store.Get(ctx, key)
531}
532
533func (s *writeDeadlineStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
534 if key == s.key {
535 s.mu.Lock()
536 s.deadline, _ = ctx.Deadline()
537 s.puts++
538 s.mu.Unlock()
539 }
540 return s.Store.Put(ctx, key, body, ifMatch)
541}
542
543// recordKey is the bucket key a door's create writes.
544func (d createDoor) recordKey() string {
545 if d.name == "sidecar" {
546 return provision.SidecarRecordKey("bab", "titogo", "bot")
547 }
548 return provision.RecordKey("bab", "titogo", d.host)
549}
550
551// slowReplica delays every listing, and remembers when the last one ended.
552type slowReplica struct {
553 replicaReader
554 delay time.Duration
555 last *atomic.Int64
556}
557
558func (r slowReplica) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
559 time.Sleep(r.delay)
560 itr, err := r.replicaReader.LTXFiles(ctx, level, seek, useMetadata)
561 r.last.Store(time.Now().UnixNano())
562 return itr, err
563}
564
565// The record write gets a full budget of its own AFTER the check, rather
566// than whatever the request's original deadline has left once validation and
567// twenty listings have spent it.
568func TestCreateWritesUnderAFreshDeadlineAfterTheCheck(t *testing.T) {
569 for _, d := range leftoverDoors {
570 t.Run(d.name, func(t *testing.T) {
571 fx := leftoverFixture(t, "code-left-deadline-"+d.name)
572 last := new(atomic.Int64)
573 fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) {
574 return slowReplica{lsfile.NewReplicaClient(filepath.Join(fx.root, prefix)), 5 * time.Millisecond, last}, nil
575 }
576 st := &writeDeadlineStore{Store: fx.srv.cfg.Releases, key: d.recordKey()}
577 fx.srv.cfg.Releases = st
578 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
579 if status != http.StatusCreated {
580 t.Fatalf("create = %d %s, want 201", status, out)
581 }
582 checked := time.Unix(0, last.Load())
583 if st.deadline.IsZero() {
584 t.Fatal("the record write ran with no deadline")
585 }
586 if floor := checked.Add(instanceStoreTimeout); st.deadline.Before(floor) {
587 t.Fatalf("record write deadline %s is %s short of a fresh budget after the check ended",
588 st.deadline.Format(time.RFC3339Nano), floor.Sub(st.deadline))
589 }
590 })
591 }
592}
593
594// A retried create whose first attempt landed (a lost response) gets the
595// ordinary duplicate answer, unchanged, even though the live record's own
596// replica now fills the prefix, and even when the bucket cannot be listed.
597func TestCreateRetryOfAnExistingRecordIsTheOrdinaryDuplicate(t *testing.T) {
598 dup := map[string]string{"instance": "instance acme.go.tito.io already exists", "sidecar": "sidecar bot already exists"}
599 for _, d := range leftoverDoors {
600 for _, seam := range []string{"replica", "failing listing"} {
601 t.Run(d.name+"/"+seam, func(t *testing.T) {
602 fx := leftoverFixture(t, "code-left-retry-"+d.name)
603 if status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body); status != http.StatusCreated {
604 t.Fatalf("first create = %d %s", status, out)
605 }
606 if seam == "replica" {
607 fx.leaveReplica(t, d.host)
608 } else {
609 fx.srv.dbReplica = func(context.Context, string) (replicaReader, error) { return failingReplica{}, nil }
610 }
611 before, _, err := fx.rel.Get(context.Background(), d.recordKey())
612 if err != nil {
613 t.Fatal(err)
614 }
615 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
616 if status != http.StatusConflict || !strings.Contains(out, dup[d.name]) {
617 t.Fatalf("retry = %d %s, want the ordinary 409 %q", status, out, dup[d.name])
618 }
619 after, _, err := fx.rel.Get(context.Background(), d.recordKey())
620 if err != nil || string(after) != string(before) {
621 t.Fatalf("the retry changed the stored record (err %v)", err)
622 }
623 })
624 }
625 }
626}
627
628// A record read that fails is not "absent": the create is refused with the
629// door's own failure, and nothing is written.
630func TestCreateFailsClosedWhenTheRecordReadFails(t *testing.T) {
631 for _, d := range leftoverDoors {
632 t.Run(d.name, func(t *testing.T) {
633 fx := leftoverFixture(t, "code-left-getfail-"+d.name)
634 st := &writeDeadlineStore{Store: fx.srv.cfg.Releases, key: d.recordKey(), failGet: true}
635 fx.srv.cfg.Releases = st
636 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
637 if status != http.StatusBadGateway || strings.Contains(out, "secret-bucket") {
638 t.Fatalf("create = %d %s, want the door's own 502 with no bucket name", status, out)
639 }
640 if st.puts != 0 {
641 t.Fatalf("a create whose record read failed wrote the record %d time(s)", st.puts)
642 }
643 })
644 }
645}
646
647// A pooled record's data lives on the pool's own box, not in this bucket
648// (resolveReplica's rule), so the rail does not look, and leftovers here do
649// not refuse it.
650func TestCreateSkipsTheCheckForPooledRecords(t *testing.T) {
651 for _, d := range leftoverDoors {
652 t.Run(d.name, func(t *testing.T) {
653 fx := leftoverFixture(t, "code-left-pool-"+d.name)
654 enableInstances(t, fx.consoleFixture, "titogo", titoManifest())
655 enableSidecars(t, fx.consoleFixture, "titogo", titoSidecarManifest())
656 fx.leaveReplica(t, d.host)
657 status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body)
658 if status != http.StatusCreated {
659 t.Fatalf("pooled create = %d %s, want 201 (unchecked)", status, out)
660 }
661 if n := fx.opens.Load(); n != 0 {
662 t.Fatalf("replica clients opened = %d for a pooled record, want 0", n)
663 }
664 })
665 }
666}
667