| new file mode 100644 |
| index 000000000..0732fd7f7 |
| --- /dev/null |
| +++ b/internal/hostops/operations_test.go |
| @@ -0,0 +1,457 @@ |
| +package hostops |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +type authority struct{ err error } |
| + |
| +func (a *authority) Create(context.Context, hostclaim.Observed, Record, string) error { return a.err } |
| +func (a *authority) Executor(context.Context, hostclaim.Observed, Record) (string, error) { |
| + return "box-a", a.err |
| +} |
| +func (a *authority) StartDelete(context.Context, hostclaim.Observed, Record) error { return a.err } |
| +func (a *authority) FirstPhase(context.Context, hostclaim.Observed) error { return a.err } |
| +func (a *authority) Inventory(context.Context, hostclaim.Observed) ([]string, []string, error) { |
| + return []string{"ie", "us"}, []string{"box-a", "box-b"}, a.err |
| +} |
| +func (a *authority) Adoption(context.Context, Record) (string, string, []string, []string, error) { |
| + return "ie", "box-a", []string{"us"}, []string{"box-b"}, a.err |
| +} |
| +func (a *authority) PurgeRetirement(context.Context, hostclaim.Observed) error { return a.err } |
| + |
| +type faultStore struct { |
| + release.Store |
| + failGet string |
| + beforePut func(string) error |
| + beforeDelete func(string, string) error |
| + bareDeletes int |
| +} |
| + |
| +func (s *faultStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == s.failGet { |
| + return nil, "", errors.New("read failed") |
| + } |
| + return s.Store.Get(ctx, key) |
| +} |
| +func (s *faultStore) Put(ctx context.Context, key string, b []byte, e string) (string, error) { |
| + if s.beforePut != nil { |
| + if err := s.beforePut(key); err != nil { |
| + return "", err |
| + } |
| + } |
| + return s.Store.Put(ctx, key, b, e) |
| +} |
| +func (s *faultStore) Delete(context.Context, string) error { |
| + s.bareDeletes++ |
| + return errors.New("bare delete forbidden") |
| +} |
| +func (s *faultStore) DeleteIfMatch(ctx context.Context, key, e string) error { |
| + if s.beforeDelete != nil { |
| + if err := s.beforeDelete(key, e); err != nil { |
| + return err |
| + } |
| + } |
| + return release.DeleteIfMatch(ctx, s.Store, key, e) |
| +} |
| + |
| +type fixture struct { |
| + ctx context.Context |
| + store *faultStore |
| + c Controller |
| + auth *authority |
| + now time.Time |
| + serial int |
| +} |
| + |
| +func newFixture(t *testing.T) *fixture { |
| + t.Helper() |
| + f := &fixture{ctx: context.Background(), now: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC), auth: &authority{}} |
| + f.store = &faultStore{Store: release.NewDirStore(t.TempDir())} |
| + f.c = Controller{Store: f.store, Authority: f.auth, Now: func() time.Time { return f.now }, NewID: func() string { f.serial++; return fmt.Sprintf("id-%d", f.serial) }} |
| + return f |
| +} |
| +func (f *fixture) record(t *testing.T, host string) Record { |
| + t.Helper() |
| + r, err := PrepareInstance(provision.Record{Schema: 1, Account: "bab", App: "jam", Host: host, Slug: "shop", Channel: "edge", Region: "ie"}, f.now, "ie") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return r |
| +} |
| +func (f *fixture) serving(t *testing.T, host string) hostclaim.Observed { |
| + t.Helper() |
| + r := f.record(t, host) |
| + before, err := hostclaim.Read(f.ctx, f.store, host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + obs, err := f.c.Attempt(f.ctx, before, r, f.c.NewID(), "test") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return obs |
| +} |
| +func TestAttemptUsesPreRegistrationObservation(t *testing.T) { |
| + f := newFixture(t) |
| + for _, host := range []string{"one.bab.example.com", "two.bab.example.com"} { |
| + before, err := hostclaim.Read(f.ctx, f.store, host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + fenced, err := hostclaim.Update(f.ctx, f.store, before, FencePurge("purge", f.now)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err = hostclaim.Update(f.ctx, f.store, fenced, ReleasePurgeFence("purge")); err != nil { |
| + t.Fatal(err) |
| + } |
| + r := f.record(t, host) |
| + if _, err = f.c.Attempt(f.ctx, before, r, "paused", "test"); !errors.Is(err, hostclaim.ErrConflict) { |
| + t.Fatalf("late claim: %v", err) |
| + } |
| + if _, _, err = f.store.Get(f.ctx, r.Holder.Key); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("record landed: %v", err) |
| + } |
| + } |
| + if f.store.bareDeletes != 0 { |
| + t.Fatal("deleted claim") |
| + } |
| +} |
| +func TestRecoverCreateAbsentAndLateOrphan(t *testing.T) { |
| + f := newFixture(t) |
| + host := "shop.bab.example.com" |
| + r := f.record(t, host) |
| + before, _ := hostclaim.Read(f.ctx, f.store, host) |
| + if _, err := hostclaim.Update(f.ctx, f.store, before, hostclaim.BeginCreate(r.Holder, r.Body, "create", "test", f.now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + f.now = f.now.Add(5 * time.Minute) |
| + if err := f.c.RecoverCreate(f.ctx, host); !errors.Is(err, ErrPending) { |
| + t.Fatal(err) |
| + } |
| + f.now = f.now.Add(time.Nanosecond) |
| + if err := f.c.RecoverCreate(f.ctx, host); err != nil { |
| + t.Fatal(err) |
| + } |
| + obs, _ := hostclaim.Read(f.ctx, f.store, host) |
| + if !obs.Claim.Free() || len(obs.Claim.CancelledCreates) != 1 || len(obs.Claim.RetiredIncarnations) != 1 { |
| + t.Fatalf("cancel: %+v", obs.Claim) |
| + } |
| + f.now = f.now.Add(2 * time.Hour) |
| + if _, err := f.store.Put(f.ctx, r.Holder.Key, r.Body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + late, err := ReadRecord(f.ctx, f.store, r.Holder.Key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + removed, err := RemoveOrphan(f.ctx, f.store, late) |
| + if err != nil || !removed { |
| + t.Fatalf("late orphan: %t %v", removed, err) |
| + } |
| +} |
| +func TestCreateRecoveryReadErrorsNeverCancel(t *testing.T) { |
| + f := newFixture(t) |
| + r := f.record(t, "shop.bab.example.com") |
| + before, _ := hostclaim.Read(f.ctx, f.store, r.Holder.Host) |
| + obs, err := hostclaim.Update(f.ctx, f.store, before, hostclaim.BeginCreate(r.Holder, r.Body, "create", "test", f.now)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + f.now = f.now.Add(6 * time.Minute) |
| + f.store.failGet = r.Holder.Key |
| + if err := f.c.RecoverCreate(f.ctx, r.Holder.Host); err == nil { |
| + t.Fatal("read error accepted") |
| + } |
| + got, _ := hostclaim.Read(f.ctx, f.store, r.Holder.Host) |
| + if got.ETag != obs.ETag { |
| + t.Fatal("acted after read error") |
| + } |
| + f.store.failGet = "" |
| + if _, err := f.store.Put(f.ctx, r.Holder.Key, r.Body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := f.c.RecoverCreate(f.ctx, r.Holder.Host); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, _ = hostclaim.Read(f.ctx, f.store, r.Holder.Host) |
| + if got.Claim.Op.Kind != hostclaim.KindServing || got.Claim.Holder.Stamp != got.Claim.Gen { |
| + t.Fatal("not completed") |
| + } |
| +} |
| +func TestRetireWaitsForExactRecordAndAuthority(t *testing.T) { |
| + f := newFixture(t) |
| + obs := f.serving(t, "shop.bab.example.com") |
| + key := obs.Claim.Holder.Key |
| + deleting, err := f.c.BeginRetire(f.ctx, key, "test", obs.Claim.Holder.Incarnation) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + f.auth.err = ErrPending |
| + if err := f.c.Retire(f.ctx, obs.Host, deleting.Claim.Op.ID); !errors.Is(err, ErrPending) { |
| + t.Fatal(err) |
| + } |
| + if _, _, err := f.store.Get(f.ctx, key); err != nil { |
| + t.Fatal("record deleted before barrier") |
| + } |
| + f.auth.err = nil |
| + f.store.beforeDelete = func(k, e string) error { return release.ErrCASConflict } |
| + if err := f.c.Retire(f.ctx, obs.Host, deleting.Claim.Op.ID); !errors.Is(err, release.ErrCASConflict) { |
| + t.Fatal(err) |
| + } |
| + current, _ := hostclaim.Read(f.ctx, f.store, obs.Host) |
| + if current.Claim.Op.Kind != hostclaim.KindDeleting { |
| + t.Fatal("parked after 412") |
| + } |
| + f.store.beforeDelete = nil |
| + if err := f.c.Retire(f.ctx, obs.Host, deleting.Claim.Op.ID); err != nil { |
| + t.Fatal(err) |
| + } |
| + current, _ = hostclaim.Read(f.ctx, f.store, obs.Host) |
| + if current.Claim.Op.Kind != hostclaim.KindParking || len(current.Claim.Obligations) != 2 { |
| + t.Fatalf("parking: %+v", current.Claim) |
| + } |
| + if err := f.c.Free(f.ctx, obs.Host, deleting.Claim.Op.ID); err == nil { |
| + t.Fatal("freed with chores") |
| + } |
| + for _, ob := range current.Claim.Obligations { |
| + var err error |
| + current, err = hostclaim.Update(f.ctx, f.store, current, hostclaim.CompleteObligation(ob.Op, ob.ID)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := f.c.Free(f.ctx, obs.Host, deleting.Claim.Op.ID); err != nil { |
| + t.Fatal(err) |
| + } |
| + if f.store.bareDeletes != 0 { |
| + t.Fatal("bare delete") |
| + } |
| +} |
| +func TestAdoptionIsNotOrphanAndHasDiscovery(t *testing.T) { |
| + f := newFixture(t) |
| + r := f.record(t, "shop.bab.example.com") |
| + if _, err := f.store.Put(f.ctx, r.Holder.Key, r.Body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + read, err := ReadRecord(f.ctx, f.store, r.Holder.Key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if removed, err := RemoveOrphan(f.ctx, f.store, read); err != nil || removed { |
| + t.Fatal("legacy treated as orphan") |
| + } |
| + obs, err := f.c.Adopt(f.ctx, read) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if obs.Claim.Touched.Complete || len(obs.Claim.Obligations) != 2 || obs.Claim.Holder.Location != "instances/"+r.Holder.Host || obs.Claim.Holder.WriterBox != "box-a" { |
| + t.Fatalf("adopt: %+v", obs.Claim) |
| + } |
| + if _, err := f.c.BeginRetire(f.ctx, r.Holder.Key, "test", obs.Claim.Holder.Incarnation); err == nil { |
| + t.Fatal("delete overlapped discovery") |
| + } |
| +} |
| +func TestPurgeJournalPhasesAndFenceRetention(t *testing.T) { |
| + f := newFixture(t) |
| + p, _, err := ReservePurge(f.ctx, f.store, "bab", "jam", "purge") |
| + if err != nil || p.Phase != "reserved" { |
| + t.Fatal(err) |
| + } |
| + if _, _, err := ReservePurge(f.ctx, f.store, "bab", "jam", "other"); !errors.Is(err, ErrPending) { |
| + t.Fatal(err) |
| + } |
| + if err := RemovePurge(f.ctx, f.store, "bab", "jam", "purge"); !errors.Is(err, ErrPending) { |
| + t.Fatal(err) |
| + } |
| + if err := AdvancePurge(f.ctx, f.store, "bab", "jam", "purge", "reserved", "done"); err == nil { |
| + t.Fatal("skipped phases") |
| + } |
| + for i := 0; i < len(phases)-1; i++ { |
| + if err := AdvancePurge(f.ctx, f.store, "bab", "jam", "purge", phases[i], phases[i+1]); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := RemovePurge(f.ctx, f.store, "bab", "jam", "purge"); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| +func TestPurgeRetiresUnderOwnOperation(t *testing.T) { |
| + f := newFixture(t) |
| + obs := f.serving(t, "shop.bab.example.com") |
| + rec, err := ReadRecord(f.ctx, f.store, obs.Claim.Holder.Key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + obs, err = hostclaim.Update(f.ctx, f.store, obs, BeginPurge("purge", "test", f.now, []string{"ie"}, []string{"box-a"})) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + obs, err = hostclaim.Update(f.ctx, f.store, obs, PurgeRecordIntent("purge", rec.Holder.Key, rec.ETag)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + f.auth.err = ErrPending |
| + if err := f.c.Retire(f.ctx, obs.Host, "purge"); !errors.Is(err, ErrPending) { |
| + t.Fatal(err) |
| + } |
| + f.auth.err = nil |
| + if err := f.c.Retire(f.ctx, obs.Host, "purge"); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, _ := hostclaim.Read(f.ctx, f.store, obs.Host) |
| + if got.Claim.Op.Kind != hostclaim.KindParking || got.Claim.Op.ID != "purge" { |
| + t.Fatal("purge identity lost") |
| + } |
| +} |
| +func TestPrepareRejectsCallerTimeAndSidecarIdentity(t *testing.T) { |
| + f := newFixture(t) |
| + if _, err := PrepareInstance(provision.Record{CreatedAt: f.now}, f.now, "ie"); err == nil { |
| + t.Fatal("accepted caller time") |
| + } |
| + r, err := PrepareSidecar(provision.SidecarRecord{Account: "bab", App: "jam", Name: "job", Host: provision.SidecarHost("job", "jam", "bab"), Region: "ie"}, f.now, "ie") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + before, _ := hostclaim.Read(f.ctx, f.store, r.Holder.Host) |
| + obs, err := f.c.Attempt(f.ctx, before, r, "sidecar", "test") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if obs.Claim.Holder.Family != "sidecar" { |
| + t.Fatal("sidecar became instance") |
| + } |
| +} |
| +func TestRecoveryFreshReadsAndRequiredHooks(t *testing.T) { |
| + f := newFixture(t) |
| + if err := (Recovery{Controller: f.c}).Pass(f.ctx); !errors.Is(err, ErrContract) { |
| + t.Fatal(err) |
| + } |
| + calls := 0 |
| + r := Recovery{Controller: f.c, Purge: func(context.Context, hostclaim.Observed) error { return nil }, Placement: func(context.Context, hostclaim.Observed) error { return nil }, SweepRecords: func(context.Context) error { calls++; return nil }, SweepPurges: func(context.Context) error { calls++; return nil }, SweepAdmissions: func(context.Context) error { return nil }} |
| + if err := r.Pass(f.ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if calls != 2 { |
| + t.Fatal(calls) |
| + } |
| +} |
| + |
| +func TestAppsPreserveLegacyPrefixesAndFindClaimlessPurge(t *testing.T) { |
| + f := newFixture(t) |
| + for key, body := range map[string][]byte{ |
| + release.LegacyChannelKey("my-app", "stable"): []byte("legacy channel"), |
| + release.LegacyChannelKey("myapp", "stable"): []byte("legacy channel"), |
| + release.LegacyManifestKey("myapp", "v1"): []byte("legacy manifest"), |
| + release.LegacyArtifactKey("myapp", "v1", "app"): []byte("legacy artifact"), |
| + } { |
| + if _, err := f.store.Put(f.ctx, key, body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if _, _, err := ReservePurge(f.ctx, f.store, "bab", "doomed", "purge"); err != nil { |
| + t.Fatal(err) |
| + } |
| + hosts, err := ListClaims(f.ctx, f.store) |
| + if err != nil || len(hosts) != 0 { |
| + t.Fatalf("claims = %v, %v; want none", hosts, err) |
| + } |
| + if err := f.c.SweepAllRecords(f.ctx); err != nil { |
| + t.Fatalf("record sweep with legacy prefixes: %v", err) |
| + } |
| + var replayed []App |
| + err = SweepAllPurges(f.ctx, f.store, func(_ context.Context, app App, p Purge) error { |
| + if p.ID != "purge" { |
| + t.Fatalf("purge ID = %q", p.ID) |
| + } |
| + replayed = append(replayed, app) |
| + return nil |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(replayed) != 1 || replayed[0] != (App{Account: "bab", Name: "doomed"}) { |
| + t.Fatalf("replayed = %+v", replayed) |
| + } |
| +} |
| + |
| +func TestOrphanConditionalDeleteCannotEraseReplacement(t *testing.T) { |
| + f := newFixture(t) |
| + r := f.record(t, "shop.bab.example.com") |
| + if _, err := f.store.Put(f.ctx, r.Holder.Key, r.Body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + old, err := ReadRecord(f.ctx, f.store, r.Holder.Key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + absent, _ := hostclaim.Read(f.ctx, f.store, r.Holder.Host) |
| + if _, err := hostclaim.Update(f.ctx, f.store, absent, func(*hostclaim.Claim) error { return nil }); err != nil { |
| + t.Fatal(err) |
| + } |
| + replacement := append([]byte(nil), r.Body...) |
| + replacement = append(replacement, ' ') |
| + f.store.beforeDelete = func(key, etag string) error { |
| + _, err := f.store.Store.Put(f.ctx, key, replacement, etag) |
| + return err |
| + } |
| + if removed, err := RemoveOrphan(f.ctx, f.store, old); removed || !errors.Is(err, release.ErrCASConflict) { |
| + t.Fatalf("removed=%t err=%v", removed, err) |
| + } |
| + if _, _, err := f.store.Get(f.ctx, r.Holder.Key); err != nil { |
| + t.Fatal("replacement lost") |
| + } |
| +} |
| + |
| +func TestPlacementStartKeepsInstallationAndExcludesDelete(t *testing.T) { |
| + f := newFixture(t) |
| + obs := f.serving(t, "shop.bab.example.com") |
| + stamp := obs.Claim.Holder.Stamp |
| + placed, err := hostclaim.Update(f.ctx, f.store, obs, BeginPlacement("place", "test", "fleet/customer", f.now, []string{"ie", "us"}, []string{"box-a", "box-b"})) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if placed.Claim.Op.Exec != obs.Claim.Op.Exec || placed.Claim.Holder.Stamp != stamp || placed.Claim.Op.Intent["target"] != "fleet/customer" { |
| + t.Fatal("placement changed authority") |
| + } |
| + rec, err := ReadRecord(f.ctx, f.store, obs.Claim.Holder.Key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := hostclaim.Update(f.ctx, f.store, placed, hostclaim.BeginDelete("delete", "test", rec.ETag, f.now, nil, nil)); err == nil { |
| + t.Fatal("delete overlapped placement") |
| + } |
| +} |
| + |
| +func TestFenceReleasePreservesHistoryAndRefusesAnotherPurge(t *testing.T) { |
| + f := newFixture(t) |
| + before, _ := hostclaim.Read(f.ctx, f.store, "shop.bab.example.com") |
| + obs, err := hostclaim.Update(f.ctx, f.store, before, func(c *hostclaim.Claim) error { |
| + c.Lost = []string{"lost-box"} |
| + c.QuietAfter = map[string]time.Time{"ie": f.now.Add(time.Hour)} |
| + return FencePurge("purge", f.now)(c) |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := hostclaim.Update(f.ctx, f.store, obs, ReleasePurgeFence("other")); err == nil { |
| + t.Fatal("released other's fence") |
| + } |
| + obs, err = hostclaim.Update(f.ctx, f.store, obs, ReleasePurgeFence("purge")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !obs.Claim.Free() || len(obs.Claim.Lost) != 1 || obs.Claim.QuietAfter["ie"].IsZero() { |
| + t.Fatal("lost history") |
| + } |
| +} |