| new file mode 100644 |
| index 000000000..68f08c4dd |
| --- /dev/null |
| +++ b/internal/console/createleftovers_test.go |
| @@ -0,0 +1,666 @@ |
| +package console |
| + |
| +// The leftover-replica rail on the two create doors (discussion #199). A |
| +// deleted instance or sidecar leaves its replica at instances/<host> and any |
| +// move staging at moves/<host>/, and both are keyed by HOST alone, so the next |
| +// record created on that host would inherit the old database on its first |
| +// wake or through `carlos db pull`. Until the full fix lands, both doors |
| +// refuse a create on a host whose replica or staging prefix in this |
| +// console's data bucket still holds an LTX file. These tests use real |
| +// litestream replicas on disk behind the db door's replica seam, so "holds an |
| +// LTX file" means exactly what the activator would restore from. |
| + |
| +import ( |
| + "context" |
| + "encoding/xml" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "net/http" |
| + "net/http/httptest" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + lsfile "github.com/benbjohnson/litestream/file" |
| + "github.com/superfly/ltx" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +type leftoverFx struct { |
| + *consoleFixture |
| + token string |
| + root string // the fake data bucket: one file replica per prefix |
| + opens *atomic.Int32 // how many times a door reached for a replica client |
| +} |
| + |
| +// leftoverFixture is a console in region ie with a data bucket configured, |
| +// and titogo enabled for both instances and sidecars in ie. Neither manifest |
| +// names a pool: a pooled record's data is on the pool box, which the rail |
| +// skips (TestCreateSkipsTheCheckForPooledRecords). |
| +func leftoverFixture(t *testing.T, code string) *leftoverFx { |
| + t.Helper() |
| + fx, token := instanceFixture(t, code) |
| + im := titoManifest() |
| + im.Pool = "" |
| + enableInstances(t, fx, "titogo", im) |
| + sm := titoSidecarManifest() |
| + sm.Pool = "" |
| + enableSidecars(t, fx, "titogo", sm) |
| + root := t.TempDir() |
| + fx.srv.cfg.HibernateBucket = "data-test" |
| + fx.srv.cfg.HibernatePrefix = "instances" |
| + fx.srv.cfg.Region = "ie" |
| + opens := new(atomic.Int32) |
| + fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) { |
| + opens.Add(1) |
| + return lsfile.NewReplicaClient(filepath.Join(root, prefix)), nil |
| + } |
| + return &leftoverFx{consoleFixture: fx, token: token, root: root, opens: opens} |
| +} |
| + |
| +// leaveReplica writes a real litestream replica where a deleted record's |
| +// activator left it. |
| +func (fx *leftoverFx) leaveReplica(t *testing.T, host string) { |
| + t.Helper() |
| + dbdoortest.Replicate(t, filepath.Join(fx.root, activator.HibernatePrefix("instances", host)), 1) |
| +} |
| + |
| +// leaveStaging writes one where an abandoned move staged it. |
| +func (fx *leftoverFx) leaveStaging(t *testing.T, host string) { |
| + t.Helper() |
| + dbdoortest.Replicate(t, filepath.Join(fx.root, activator.MovesPrefix(host)), 1) |
| +} |
| + |
| +// createDoor is one of the two doors, reduced to what the matrix needs. |
| +type createDoor struct { |
| + name string |
| + url func(fx *consoleFixture) string |
| + body string // a create the door accepts on a clean host, in region ie |
| + host string // the host that create claims |
| + exists func(t *testing.T, fx *consoleFixture) bool |
| +} |
| + |
| +var leftoverDoors = []createDoor{ |
| + { |
| + name: "instance", |
| + url: func(fx *consoleFixture) string { return instancesURL(fx, "bab", "titogo") }, |
| + body: `{"host":"acme.go.tito.io","region":"ie"}`, |
| + host: "acme.go.tito.io", |
| + exists: func(t *testing.T, fx *consoleFixture) bool { |
| + t.Helper() |
| + _, err := provision.GetRecord(context.Background(), fx.rel, "bab", "titogo", "acme.go.tito.io") |
| + if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + t.Fatal(err) |
| + } |
| + return err == nil |
| + }, |
| + }, |
| + { |
| + name: "sidecar", |
| + url: func(fx *consoleFixture) string { return sidecarsURL(fx, "bab", "titogo") }, |
| + body: `{"name":"bot","region":"ie"}`, |
| + host: botHost, |
| + exists: func(t *testing.T, fx *consoleFixture) bool { |
| + t.Helper() |
| + _, err := provision.GetSidecarRecord(context.Background(), fx.rel, "bab", "titogo", "bot") |
| + if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + t.Fatal(err) |
| + } |
| + return err == nil |
| + }, |
| + }, |
| +} |
| + |
| +// A host whose replica prefix still holds LTX files is refused with 409, |
| +// and the refusal writes no record. |
| +func TestCreateRefusesLeftoverReplica(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-rep-"+d.name) |
| + fx.leaveReplica(t, d.host) |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusConflict || !strings.Contains(out, "still holds data from a deleted") { |
| + t.Fatalf("create = %d %s, want the leftover-data 409", status, out) |
| + } |
| + if d.exists(t, fx.consoleFixture) { |
| + t.Fatal("a refused create wrote a record anyway") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// Move staging counts too: a host whose only leftover is an abandoned |
| +// moves/<host>/ is refused the same way, because a landing promotes that |
| +// staging into the instance's prefix. |
| +func TestCreateRefusesLeftoverMoveStaging(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-mov-"+d.name) |
| + fx.leaveStaging(t, d.host) |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusConflict || !strings.Contains(out, "still holds data from a deleted") { |
| + t.Fatalf("create = %d %s, want the leftover-data 409", status, out) |
| + } |
| + if d.exists(t, fx.consoleFixture) { |
| + t.Fatal("a refused create wrote a record anyway") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A clean host is created as before, and the check really looked: both |
| +// prefixes were opened. |
| +func TestCreateAllowsACleanHost(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-clean-"+d.name) |
| + // Another host's leftovers are not this host's business. |
| + fx.leaveReplica(t, "other.go.tito.io") |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusCreated { |
| + t.Fatalf("create = %d %s, want 201", status, out) |
| + } |
| + if !d.exists(t, fx.consoleFixture) { |
| + t.Fatal("the create answered 201 and wrote no record") |
| + } |
| + if n := fx.opens.Load(); n != 2 { |
| + t.Fatalf("replica clients opened = %d, want 2 (the replica prefix and the staging prefix)", n) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A record homed in another region is not checked: its data lives in that |
| +// region's bucket, which this console cannot read. The gap is documented at |
| +// the rail; this pins that the rail does not refuse what it cannot see. |
| +func TestCreateSkipsTheCheckInAnotherRegion(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-region-"+d.name) |
| + fx.srv.cfg.Region = "uk" |
| + fx.leaveReplica(t, d.host) |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusCreated { |
| + t.Fatalf("create = %d %s, want 201 (unchecked)", status, out) |
| + } |
| + if n := fx.opens.Load(); n != 0 { |
| + t.Fatalf("replica clients opened = %d for an out-of-region record, want 0", n) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A console with no data bucket creates as it always did, and never reaches |
| +// for a replica client. |
| +func TestCreateSkipsTheCheckWithNoDataBucket(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-nobucket-"+d.name) |
| + fx.srv.cfg.HibernateBucket = "" |
| + fx.leaveReplica(t, d.host) |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusCreated { |
| + t.Fatalf("create = %d %s, want 201 (unchecked)", status, out) |
| + } |
| + if n := fx.opens.Load(); n != 0 { |
| + t.Fatalf("replica clients opened = %d with no bucket, want 0", n) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// failingReplica fails every listing, with a message carrying a bucket name |
| +// the way a real S3 error does. |
| +type failingReplica struct{} |
| + |
| +func (failingReplica) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) { |
| + return nil, errors.New("list secret-bucket: connection reset") |
| +} |
| + |
| +func (failingReplica) OpenLTXFile(context.Context, int, ltx.TXID, ltx.TXID, int64, int64) (io.ReadCloser, error) { |
| + return nil, errors.New("not used") |
| +} |
| + |
| +// erroringIterator fails on the first Next, the way the S3 client surfaces |
| +// a paging error. |
| +type erroringIterator struct{} |
| + |
| +func (erroringIterator) Close() error { return nil } |
| +func (erroringIterator) Next() bool { return false } |
| +func (erroringIterator) Err() error { return errors.New("page secret-bucket: timeout") } |
| +func (erroringIterator) Item() *ltx.FileInfo { return nil } |
| + |
| +type pagingFailReplica struct{ failingReplica } |
| + |
| +func (pagingFailReplica) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) { |
| + return erroringIterator{}, nil |
| +} |
| + |
| +// A check that cannot complete refuses with 503: never create on |
| +// uncertainty. The store's own error, which names the bucket, stays in the |
| +// log. |
| +func TestCreateRefusesWhenTheCheckFails(t *testing.T) { |
| + seams := map[string]func(context.Context, string) (replicaReader, error){ |
| + "client": func(context.Context, string) (replicaReader, error) { |
| + return nil, errors.New("resolve region for secret-bucket: no credentials") |
| + }, |
| + "listing": func(context.Context, string) (replicaReader, error) { return failingReplica{}, nil }, |
| + "paging": func(context.Context, string) (replicaReader, error) { return pagingFailReplica{}, nil }, |
| + } |
| + for _, d := range leftoverDoors { |
| + for seam, open := range seams { |
| + t.Run(d.name+"/"+seam, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-fail-"+d.name+"-"+seam) |
| + fx.srv.dbReplica = open |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusServiceUnavailable || !strings.Contains(out, "could not check whether") { |
| + t.Fatalf("create = %d %s, want the 503 that refuses on uncertainty", status, out) |
| + } |
| + if strings.Contains(out, "secret-bucket") || strings.Contains(out, "data-test") { |
| + t.Fatalf("the refusal leaked a bucket name: %s", out) |
| + } |
| + if d.exists(t, fx.consoleFixture) { |
| + t.Fatal("a create that could not be checked wrote a record anyway") |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +// Every existing validation answers first: a create the door would refuse |
| +// anyway never reaches for the bucket, and keeps its own answer. |
| +func TestCreateValidatesBeforeTheCheck(t *testing.T) { |
| + // consoleRegion is the region the rail would check, so a check run |
| + // too early would reach for the bucket rather than skip. |
| + cases := []struct{ door, url, body, consoleRegion string }{ |
| + {"instance", "instances", `{"host":"acme.elsewhere.io","region":"ie"}`, "ie"}, |
| + {"instance", "instances", `{"host":"acme.go.tito.io","region":"mars"}`, "mars"}, |
| + {"sidecar", "sidecars", `{"name":"../bot","region":"ie"}`, "ie"}, |
| + {"sidecar", "sidecars", `{"name":"bot","region":"mars"}`, "mars"}, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.door+" "+c.body, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-order") |
| + fx.srv.cfg.Region = c.consoleRegion |
| + fx.srv.dbReplica = func(context.Context, string) (replicaReader, error) { |
| + fx.opens.Add(1) |
| + return failingReplica{}, nil |
| + } |
| + u := fx.ts.URL + "/api/cli/apps/bab/titogo/" + c.url |
| + status, _, out := rawReq(t, http.MethodPost, u, fx.token, c.body) |
| + if status != http.StatusBadRequest { |
| + t.Fatalf("create = %d %s, want the door's own 400", status, out) |
| + } |
| + if n := fx.opens.Load(); n != 0 { |
| + t.Fatalf("an invalid create reached for the bucket %d time(s)", n) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// Paths that create no record are untouched: deleting a record whose host |
| +// holds a replica (the ordinary case) never consults the rail. |
| +func TestDeleteNeverConsultsTheLeftoverCheck(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-del-"+d.name) |
| + if status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body); status != http.StatusCreated { |
| + t.Fatalf("create = %d %s", status, out) |
| + } |
| + fx.leaveReplica(t, d.host) |
| + fx.opens.Store(0) |
| + target := d.url(fx.consoleFixture) + "/" + d.host |
| + if d.name == "sidecar" { |
| + target = d.url(fx.consoleFixture) + "/bot" |
| + } |
| + if status, _, out := rawReq(t, http.MethodDelete, target, fx.token, ""); status >= 300 { |
| + t.Fatalf("delete = %d %s, want success", status, out) |
| + } |
| + if n := fx.opens.Load(); n != 0 { |
| + t.Fatalf("a delete reached for the bucket %d time(s)", n) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// The residency rules the rail applies are resolveReplica's: in-region |
| +// only, a pooled record's data is on the pool box, and a fleet placement is |
| +// never this console's to read, whatever its own region string says. |
| +func TestLeftoverCheckResidency(t *testing.T) { |
| + s := &Server{} |
| + s.cfg.HibernateBucket = "data-test" |
| + s.cfg.Region = "fleet/acme" |
| + for _, c := range []struct { |
| + region, pool, console string |
| + bucket string |
| + want bool |
| + }{ |
| + {"", "", "", "data-test", true}, |
| + {"ie", "", "ie", "data-test", true}, |
| + {"uk", "", "ie", "data-test", false}, |
| + {"", "", "ie", "data-test", false}, |
| + {"fleet/acme", "", "fleet/acme", "data-test", false}, |
| + {"ie", "", "ie", "", false}, |
| + {"ie", "tito", "ie", "data-test", false}, |
| + {"", "tito", "", "data-test", false}, |
| + } { |
| + s.cfg.Region, s.cfg.HibernateBucket = c.console, c.bucket |
| + if got := s.leftoverCheckApplies(c.region, c.pool); got != c.want { |
| + 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) |
| + } |
| + } |
| +} |
| + |
| +// ---- review round 1 (Astra, 6cc86d34) ----------------------------------------- |
| + |
| +// fakeDataBucket is a ListObjectsV2 endpoint with real S3 prefix semantics, |
| +// for the PRODUCTION replica client (lease.ReadReplicaClient through |
| +// cachedReadReplica). The file replica the other tests use joins paths with |
| +// filepath.Join, which forgives a doubled slash that S3 does not; this is |
| +// where the exact keys the box writes, and the exact prefixes the console |
| +// lists, meet. |
| +type fakeDataBucket struct { |
| + mu sync.Mutex |
| + keys []string |
| + prefixes []string |
| +} |
| + |
| +func (b *fakeDataBucket) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| + q := r.URL.Query() |
| + if r.Method != http.MethodGet || q.Get("list-type") != "2" { |
| + http.Error(w, "the leftover check only lists", http.StatusMethodNotAllowed) |
| + return |
| + } |
| + prefix := q.Get("prefix") |
| + b.mu.Lock() |
| + b.prefixes = append(b.prefixes, prefix) |
| + var keys []string |
| + for _, k := range b.keys { |
| + if strings.HasPrefix(k, prefix) { |
| + keys = append(keys, k) |
| + } |
| + } |
| + b.mu.Unlock() |
| + type obj struct { |
| + Key string |
| + Size int64 |
| + LastModified string |
| + } |
| + type result struct { |
| + XMLName xml.Name `xml:"ListBucketResult"` |
| + Name string |
| + Prefix string |
| + KeyCount int |
| + IsTruncated bool |
| + Contents []obj |
| + } |
| + res := result{Name: "data-test", Prefix: prefix, KeyCount: len(keys)} |
| + for _, k := range keys { |
| + res.Contents = append(res.Contents, obj{Key: k, Size: 100, LastModified: "2026-10-01T00:00:00Z"}) |
| + } |
| + w.Header().Set("Content-Type", "application/xml") |
| + _ = xml.NewEncoder(w).Encode(res) |
| +} |
| + |
| +func (b *fakeDataBucket) listed() []string { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + return append([]string(nil), b.prefixes...) |
| +} |
| + |
| +// ltxKey is the key litestream's S3 client writes one LTX file under: |
| +// <path>/<level as %04x>/<filename>. |
| +func ltxKey(prefix string, level int) string { |
| + return fmt.Sprintf("%s/%04x/%s", strings.TrimSuffix(prefix, "/"), level, ltx.FormatFilename(1, 1)) |
| +} |
| + |
| +// useFakeDataBucket points s at a fake S3 endpoint holding keys, through the |
| +// production replica client. |
| +func useFakeDataBucket(t *testing.T, s *Server, keys ...string) *fakeDataBucket { |
| + t.Helper() |
| + b := &fakeDataBucket{keys: keys} |
| + srv := httptest.NewServer(b) |
| + t.Cleanup(srv.Close) |
| + t.Setenv("CARLOS_S3_ENDPOINT", srv.URL) |
| + t.Setenv("AWS_ACCESS_KEY_ID", "test-key") |
| + t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret") |
| + t.Setenv("AWS_SESSION_TOKEN", "") |
| + t.Setenv("AWS_EC2_METADATA_DISABLED", "true") |
| + s.cfg.HibernateBucket = "data-test" |
| + s.cfg.HibernateRegion = "us-east-1" |
| + s.cfg.HibernatePrefix = "instances" |
| + s.dbReplica = s.cachedReadReplica |
| + return b |
| +} |
| + |
| +// The production client is asked for exactly the prefixes the box writes: |
| +// instances/<host>/<level>/ and moves/<host>/<level>/, levels 0 to 9, with |
| +// one slash between each segment. A doubled slash lists nothing on S3. |
| +func TestLeftoverCheckListsTheExactS3Prefixes(t *testing.T) { |
| + s := &Server{} |
| + b := useFakeDataBucket(t, s) |
| + held, err := s.hostHoldsReplicaData(context.Background(), "app.example.com") |
| + if err != nil || held { |
| + t.Fatalf("empty bucket = %v, %v; want false, nil", held, err) |
| + } |
| + var want []string |
| + for _, root := range []string{"instances/app.example.com", "moves/app.example.com"} { |
| + for level := 0; level <= 9; level++ { |
| + want = append(want, fmt.Sprintf("%s/%04x/", root, level)) |
| + } |
| + } |
| + if got := b.listed(); strings.Join(got, "\n") != strings.Join(want, "\n") { |
| + t.Fatalf("listed prefixes:\n%s\nwant:\n%s", strings.Join(got, "\n"), strings.Join(want, "\n")) |
| + } |
| +} |
| + |
| +// Data at any level counts, the snapshot level included, in either prefix; |
| +// and a host is never answered with another host's data, even one its name |
| +// is a string prefix of. |
| +func TestLeftoverCheckThroughTheS3Client(t *testing.T) { |
| + const host = "app.example.co" |
| + for _, c := range []struct { |
| + name string |
| + keys []string |
| + want bool |
| + }{ |
| + {"replica at L0", []string{ltxKey("instances/"+host, 0)}, true}, |
| + {"snapshot only (L9)", []string{ltxKey("instances/"+host, 9)}, true}, |
| + {"move staging at L0", []string{ltxKey(activator.MovesPrefix(host), 0)}, true}, |
| + {"move staging snapshot only (L9)", []string{ltxKey(activator.MovesPrefix(host), 9)}, true}, |
| + {"only a longer host's data", []string{ |
| + ltxKey("instances/app.example.com", 0), ltxKey("instances/app.example.com", 9), |
| + ltxKey(activator.MovesPrefix("app.example.com"), 0), |
| + }, false}, |
| + {"a lock file and nothing else", []string{"instances/" + host + "/lock.json"}, false}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + s := &Server{} |
| + useFakeDataBucket(t, s, c.keys...) |
| + held, err := s.hostHoldsReplicaData(context.Background(), host) |
| + if err != nil || held != c.want { |
| + t.Fatalf("hostHoldsReplicaData = %v, %v; want %v, nil", held, err, c.want) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// End to end through the production client: move staging the box wrote at |
| +// moves/<host>/0000/ refuses both doors. |
| +func TestCreateRefusesMoveStagingThroughTheS3Client(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-s3-"+d.name) |
| + useFakeDataBucket(t, fx.srv, ltxKey(activator.MovesPrefix(d.host), 0)) |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusConflict || !strings.Contains(out, "still holds data from a deleted") { |
| + t.Fatalf("create = %d %s, want the leftover-data 409", status, out) |
| + } |
| + if d.exists(t, fx.consoleFixture) { |
| + t.Fatal("a refused create wrote a record anyway") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// writeDeadlineStore records the deadline of the context the record write |
| +// ran under, and can fail reads of that record. |
| +type writeDeadlineStore struct { |
| + release.Store |
| + key string |
| + failGet bool |
| + mu sync.Mutex |
| + deadline time.Time |
| + puts int |
| +} |
| + |
| +func (s *writeDeadlineStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if s.failGet && key == s.key { |
| + return nil, "", errors.New("get secret-bucket: connection reset") |
| + } |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +func (s *writeDeadlineStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if key == s.key { |
| + s.mu.Lock() |
| + s.deadline, _ = ctx.Deadline() |
| + s.puts++ |
| + s.mu.Unlock() |
| + } |
| + return s.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// recordKey is the bucket key a door's create writes. |
| +func (d createDoor) recordKey() string { |
| + if d.name == "sidecar" { |
| + return provision.SidecarRecordKey("bab", "titogo", "bot") |
| + } |
| + return provision.RecordKey("bab", "titogo", d.host) |
| +} |
| + |
| +// slowReplica delays every listing, and remembers when the last one ended. |
| +type slowReplica struct { |
| + replicaReader |
| + delay time.Duration |
| + last *atomic.Int64 |
| +} |
| + |
| +func (r slowReplica) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) { |
| + time.Sleep(r.delay) |
| + itr, err := r.replicaReader.LTXFiles(ctx, level, seek, useMetadata) |
| + r.last.Store(time.Now().UnixNano()) |
| + return itr, err |
| +} |
| + |
| +// The record write gets a full budget of its own AFTER the check, rather |
| +// than whatever the request's original deadline has left once validation and |
| +// twenty listings have spent it. |
| +func TestCreateWritesUnderAFreshDeadlineAfterTheCheck(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-deadline-"+d.name) |
| + last := new(atomic.Int64) |
| + fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) { |
| + return slowReplica{lsfile.NewReplicaClient(filepath.Join(fx.root, prefix)), 5 * time.Millisecond, last}, nil |
| + } |
| + st := &writeDeadlineStore{Store: fx.srv.cfg.Releases, key: d.recordKey()} |
| + fx.srv.cfg.Releases = st |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusCreated { |
| + t.Fatalf("create = %d %s, want 201", status, out) |
| + } |
| + checked := time.Unix(0, last.Load()) |
| + if st.deadline.IsZero() { |
| + t.Fatal("the record write ran with no deadline") |
| + } |
| + if floor := checked.Add(instanceStoreTimeout); st.deadline.Before(floor) { |
| + t.Fatalf("record write deadline %s is %s short of a fresh budget after the check ended", |
| + st.deadline.Format(time.RFC3339Nano), floor.Sub(st.deadline)) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A retried create whose first attempt landed (a lost response) gets the |
| +// ordinary duplicate answer, unchanged, even though the live record's own |
| +// replica now fills the prefix, and even when the bucket cannot be listed. |
| +func TestCreateRetryOfAnExistingRecordIsTheOrdinaryDuplicate(t *testing.T) { |
| + dup := map[string]string{"instance": "instance acme.go.tito.io already exists", "sidecar": "sidecar bot already exists"} |
| + for _, d := range leftoverDoors { |
| + for _, seam := range []string{"replica", "failing listing"} { |
| + t.Run(d.name+"/"+seam, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-retry-"+d.name) |
| + if status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body); status != http.StatusCreated { |
| + t.Fatalf("first create = %d %s", status, out) |
| + } |
| + if seam == "replica" { |
| + fx.leaveReplica(t, d.host) |
| + } else { |
| + fx.srv.dbReplica = func(context.Context, string) (replicaReader, error) { return failingReplica{}, nil } |
| + } |
| + before, _, err := fx.rel.Get(context.Background(), d.recordKey()) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusConflict || !strings.Contains(out, dup[d.name]) { |
| + t.Fatalf("retry = %d %s, want the ordinary 409 %q", status, out, dup[d.name]) |
| + } |
| + after, _, err := fx.rel.Get(context.Background(), d.recordKey()) |
| + if err != nil || string(after) != string(before) { |
| + t.Fatalf("the retry changed the stored record (err %v)", err) |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +// A record read that fails is not "absent": the create is refused with the |
| +// door's own failure, and nothing is written. |
| +func TestCreateFailsClosedWhenTheRecordReadFails(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-getfail-"+d.name) |
| + st := &writeDeadlineStore{Store: fx.srv.cfg.Releases, key: d.recordKey(), failGet: true} |
| + fx.srv.cfg.Releases = st |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusBadGateway || strings.Contains(out, "secret-bucket") { |
| + t.Fatalf("create = %d %s, want the door's own 502 with no bucket name", status, out) |
| + } |
| + if st.puts != 0 { |
| + t.Fatalf("a create whose record read failed wrote the record %d time(s)", st.puts) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A pooled record's data lives on the pool's own box, not in this bucket |
| +// (resolveReplica's rule), so the rail does not look, and leftovers here do |
| +// not refuse it. |
| +func TestCreateSkipsTheCheckForPooledRecords(t *testing.T) { |
| + for _, d := range leftoverDoors { |
| + t.Run(d.name, func(t *testing.T) { |
| + fx := leftoverFixture(t, "code-left-pool-"+d.name) |
| + enableInstances(t, fx.consoleFixture, "titogo", titoManifest()) |
| + enableSidecars(t, fx.consoleFixture, "titogo", titoSidecarManifest()) |
| + fx.leaveReplica(t, d.host) |
| + status, _, out := rawReq(t, http.MethodPost, d.url(fx.consoleFixture), fx.token, d.body) |
| + if status != http.StatusCreated { |
| + t.Fatalf("pooled create = %d %s, want 201 (unchecked)", status, out) |
| + } |
| + if n := fx.opens.Load(); n != 0 { |
| + t.Fatalf("replica clients opened = %d for a pooled record, want 0", n) |
| + } |
| + }) |
| + } |
| +} |
amadan