| index 0e818c6ba..68f08c4dd 100644 |
| --- a/internal/console/createleftovers_test.go |
| +++ b/internal/console/createleftovers_test.go |
| @@ -12,14 +12,19 @@ package console |
| |
| 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" |
| @@ -27,6 +32,7 @@ import ( |
| "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 { |
| @@ -37,12 +43,18 @@ type leftoverFx struct { |
| } |
| |
| // leftoverFixture is a console in region ie with a data bucket configured, |
| -// and titogo enabled for both instances and sidecars in ie. |
| +// 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) |
| - enableInstances(t, fx, "titogo", titoManifest()) |
| - enableSidecars(t, fx, "titogo", titoSidecarManifest()) |
| + 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" |
| @@ -320,28 +332,335 @@ func TestDeleteNeverConsultsTheLeftoverCheck(t *testing.T) { |
| } |
| } |
| |
| -// The residency rule the rail applies is resolveReplica's: in-region only, |
| -// and a fleet placement is never this console's to read, whatever its own |
| -// region string says. |
| +// 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, console string |
| - bucket string |
| - want bool |
| + 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}, |
| + {"", "", "", "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); got != c.want { |
| - t.Errorf("record %q on console %q (bucket %q) = %v, want %v", c.region, c.console, c.bucket, got, c.want) |
| + 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