amadan

carlos / platform Public

moves-phase1-6 Task 3 fix round 5 (2): purge an incarnation's move log rows when it retires

The instance delete drops the incarnation's move_log_outbox rows once its
claim is free, beside the incarnation's other cleanup; the app purge drops
those of every instance incarnation its journal records before it is done.
The drainer's drop stays the backstop. Another incarnation's rows stay.
TestRetiringAnInstancePurgesItsMoveLogRows (the delete door, end to end) and
TestPurgingAnAppPurgesItsInstancesMoveLogRows (the purge's stage, on a
journal: the fixture cannot drive an app-with-instances purge to completion).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 90a8ede1862c1f3f24d991c0bd24974eb694e56a parent d4f8123
4 files changed, +153 −0
  • internal/console/claim_purge.go +20 −0
  • internal/console/claim_retire.go +6 −0
  • internal/console/movelog_retire_test.go +120 −0
  • internal/console/store.go +7 −0
diff --git a/internal/console/claim_purge.go b/internal/console/claim_purge.go
index 2237da65e..f64d6a5e5 100644
--- a/internal/console/claim_purge.go
+++ b/internal/console/claim_purge.go
@@ -383,6 +383,11 @@ func (p *ClaimPurgeCoordinator) Resume(ctx context.Context, account, app, id, ap
return err
}
}
+ // The purge retired its instances' incarnations (retire-hosts); their
+ // move log outbox rows go with them, before the purge is done.
+ if err = s.purgeMoveLogOfPurgedHosts(j); err != nil {
+ return err
+ }
j.Phase = "done"
if _, err = w.SavePurgeJournal(ctx, j, tag); err != nil {
return err
@@ -390,6 +395,21 @@ func (p *ClaimPurgeCoordinator) Resume(ctx context.Context, account, app, id, ap
return p.Controller.CompletePurgeFences(ctx, h)
}
+// purgeMoveLogOfPurgedHosts drops the move log outbox rows of every instance
+// incarnation the purge journal records (Plan 6 Task 3). It is local and
+// idempotent, so a replayed stage repeats it harmlessly.
+func (s *Server) purgeMoveLogOfPurgedHosts(j hostops.PurgeJournal) error {
+ for host, ph := range j.Hosts {
+ if ph.Holder == nil || ph.Holder.Incarnation == "" {
+ continue
+ }
+ if err := s.store.PurgeMoveLogIncarnation(host, ph.Holder.Incarnation); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
func (p *ClaimPurgeCoordinator) SweepRequests(ctx context.Context) error {
w, err := p.control()
if err != nil {
diff --git a/internal/console/claim_retire.go b/internal/console/claim_retire.go
index f3b9a2061..48714f96a 100644
--- a/internal/console/claim_retire.go
+++ b/internal/console/claim_retire.go
@@ -124,6 +124,12 @@ func (s *Server) retireClaimHostRecord(ctx context.Context, ref provision.Retire
if err = s.advanceClaimRetire(ctx, c, h, ref, key); err != nil {
return err
}
+ // The claim is free: the incarnation is retired, and its move log
+ // outbox rows go with it (Plan 6 Task 3; the drainer's drop of a row
+ // whose claim is gone is the backstop).
+ if err = s.store.PurgeMoveLogIncarnation(ref.Host, ref.Incarnation); err != nil {
+ return claimRetireWaiting(h, "releasing", ref.Host, err)
+ }
if err = s.releaseClaimRetireMarkers(ctx, h, ref, appID, op.Intent["acquisition"]); err != nil {
return claimRetireWaiting(h, "releasing", ref.Host, err)
}
diff --git a/internal/console/movelog_retire_test.go b/internal/console/movelog_retire_test.go
new file mode 100644
index 000000000..f169d3528
--- /dev/null
+++ b/internal/console/movelog_retire_test.go
@@ -0,0 +1,120 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostops"
+ "github.com/carlosframework/platform/internal/moves"
+)
+
+// Plan 6 Task 3 fix round 5, ruling 2: an instance incarnation's move log
+// outbox rows go when the incarnation is retired or purged, in that step,
+// not when the drainer next finds the claim gone (its drop stays a
+// backstop). Another incarnation's rows stay.
+
+// queueTestMoveLog queues one confirmed row for host, op and incarnation.
+func queueTestMoveLog(t *testing.T, st *Store, host, op, name, inc string) {
+ t.Helper()
+ body, err := json.Marshal(moves.LogEvent{V: 1, Op: op, Host: host, Event: "x", Identity: name, Actor: "console", At: time.Now().UTC()})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := st.QueueMoveLog(host, op, name, inc, body, true, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func moveLogRows(t *testing.T, st *Store, host, op string) int {
+ t.Helper()
+ got, err := st.MoveLogPending(host, op)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return len(got)
+}
+
+func holderIncarnation(t *testing.T, cx *claimCase, host string) string {
+ t.Helper()
+ o, err := hostclaim.Read(context.Background(), cx.rel, host)
+ if err != nil || o.Absent || o.Claim.Holder == nil {
+ t.Fatalf("the claim of %s: %+v %v", host, o, err)
+ }
+ return o.Claim.Holder.Incarnation
+}
+
+func TestRetiringAnInstancePurgesItsMoveLogRows(t *testing.T) {
+ cx := newClaimCaseWith(t)
+ cx.instanceWorld(t)
+ cx.instance(t, `{"host":"acme.go.tito.io","region":"ie"}`)
+ cx.quiesce()
+ inc := holderIncarnation(t, cx, "acme.go.tito.io")
+ queueTestMoveLog(t, cx.st, "acme.go.tito.io", "mv_retired", "done", inc)
+ queueTestMoveLog(t, cx.st, "acme.go.tito.io", "mv_earlier", "done", "an-earlier-incarnation")
+ if status := claimRetireDoor("/api/cli/apps/bab/titogo/instances/acme.go.tito.io", "acme.go.tito.io")(t, cx); status != http.StatusAccepted {
+ t.Fatalf("delete answered %d", status)
+ }
+ if n := moveLogRows(t, cx.st, "acme.go.tito.io", "mv_retired"); n != 1 {
+ t.Fatalf("the rows went before the incarnation was retired: %d", n)
+ }
+ // The boxes park the instance (world staging: its obligations are
+ // discharged), and the delete's retry frees the claim and retires the
+ // incarnation.
+ for i := 0; i < 3; i++ {
+ obs, err := hostclaim.Read(context.Background(), cx.rel, "acme.go.tito.io")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if obs.Absent || obs.Claim.Free() {
+ break
+ }
+ if obs.Claim.Op.Kind == hostclaim.KindParking && len(obs.Claim.Obligations) != 0 {
+ if _, err := hostclaim.Update(context.Background(), cx.rel, obs, func(c *hostclaim.Claim) error { c.Obligations = nil; return nil }); err != nil {
+ t.Fatal(err)
+ }
+ }
+ rawReq(t, http.MethodDelete, cx.ts.URL+"/api/cli/apps/bab/titogo/instances/acme.go.tito.io", cx.bearer, "")
+ }
+ if o, err := hostclaim.Read(context.Background(), cx.rel, "acme.go.tito.io"); err != nil || !(o.Absent || o.Claim.Free()) {
+ t.Fatalf("the instance was not retired: op %+v holder %+v obligations %+v %v", o.Claim.Op, o.Claim.Holder, o.Claim.Obligations, err)
+ }
+ if n := moveLogRows(t, cx.st, "acme.go.tito.io", "mv_retired"); n != 0 {
+ t.Fatalf("%d rows of the retired incarnation remain", n)
+ }
+ if n := moveLogRows(t, cx.st, "acme.go.tito.io", "mv_earlier"); n != 1 {
+ t.Fatalf("another incarnation's rows: %d, want 1", n)
+ }
+}
+
+// The app purge retires its instances' incarnations itself (RetirePurgeHosts,
+// not this door), and records each in its journal; its last stage drops their
+// rows. The fixture world cannot drive a purge of an app with instances to
+// completion (its hosts' purge needs the boxes), so the stage is tested on a
+// journal.
+func TestPurgingAnAppPurgesItsInstancesMoveLogRows(t *testing.T) {
+ fx := newConsoleFixture(t)
+ j := hostops.PurgeJournal{Hosts: map[string]hostops.PurgeHost{
+ "a.example.com": {Holder: &hostclaim.Holder{Incarnation: "inc-a"}},
+ "b.example.com": {Holder: &hostclaim.Holder{Incarnation: "inc-b"}},
+ "c.example.com": {},
+ }}
+ queueTestMoveLog(t, fx.st, "a.example.com", "mv_1", "done", "inc-a")
+ queueTestMoveLog(t, fx.st, "b.example.com", "mv_2", "done", "inc-b")
+ queueTestMoveLog(t, fx.st, "b.example.com", "mv_3", "done", "inc-b-earlier")
+ queueTestMoveLog(t, fx.st, "d.example.com", "mv_4", "done", "inc-d")
+ if err := fx.srv.purgeMoveLogOfPurgedHosts(j); err != nil {
+ t.Fatal(err)
+ }
+ for _, c := range []struct {
+ host, op string
+ want int
+ }{{"a.example.com", "mv_1", 0}, {"b.example.com", "mv_2", 0}, {"b.example.com", "mv_3", 1}, {"d.example.com", "mv_4", 1}} {
+ if n := moveLogRows(t, fx.st, c.host, c.op); n != c.want {
+ t.Errorf("%s %s: %d rows, want %d", c.host, c.op, n, c.want)
+ }
+ }
+}
diff --git a/internal/console/store.go b/internal/console/store.go
index efa71a212..5d955744c 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -2304,6 +2304,13 @@ func (s *Store) RetryMoveLog(host, op, name string, attempts int, next time.Time
return err
}
+// PurgeMoveLogIncarnation drops every row of one instance incarnation: the
+// incarnation is retired or purged, so none of them can ever be written.
+func (s *Store) PurgeMoveLogIncarnation(host, incarnation string) error {
+ _, err := s.db.Exec(`DELETE FROM move_log_outbox WHERE host = ? AND incarnation = ?`, host, incarnation)
+ return err
+}
+
// SettleMoveLog drops an entry once it is written, found written, or can
// never be.
func (s *Store) SettleMoveLog(host, op, name string) error {