amadan

carlos / platform Public

registry: the source's row and its timers rows deleted in one transaction, only under the holding op (§7.9 step 3, H5(d))

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 397df23793faa6aec239c4ab09a9684d6cfa3a69 parent 118031d
2 files changed, +94 −0
  • internal/registry/hold_test.go +64 −0
  • internal/registry/registry.go +30 −0
diff --git a/internal/registry/hold_test.go b/internal/registry/hold_test.go
index 34d016f56..688f87076 100644
--- a/internal/registry/hold_test.go
+++ b/internal/registry/hold_test.go
@@ -185,6 +185,70 @@ func TestDeleteIfHold(t *testing.T) {
}
}
+// The source's finalization (moves spec §7.9 step 3, ruling H5(d)): the row
+// and the host's timer rows go in one transaction, and only while the op
+// holds the row. Any refusal leaves both, and the neighbour's timers stay.
+func TestDeleteIfHoldWithTimers(t *testing.T) {
+ r := holdFixture(t, "h")
+ if err := r.Upsert(Route{Host: "n", Kind: "instance", Socket: "/n", App: "a", Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+ for _, tm := range []Timer{{Host: "h", Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000},
+ {Host: "h", Name: "hourly", Kind: "cron", Spec: "0 * * * *", NextAt: 2000},
+ {Host: "n", Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}} {
+ if err := r.UpsertTimer(tm); err != nil {
+ t.Fatal(err)
+ }
+ }
+ timers := func(host string) int {
+ t.Helper()
+ got, err := r.TimersFor(host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return len(got)
+ }
+ // Not held, held by another op, an empty op: nothing goes.
+ for _, c := range []struct{ name, hold, op string }{{"unheld", "", holdOpA}, {"another op", holdOpB, holdOpA}, {"empty op", holdOpA, ""}} {
+ if c.hold != "" {
+ if err := r.SetHold("h", "moving", c.hold); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if ok, err := r.DeleteIfHoldWithTimers("h", c.op); err != nil || ok {
+ t.Fatalf("%s: %v %v", c.name, ok, err)
+ }
+ if rt, _ := r.Get("h"); rt == nil || timers("h") != 2 {
+ t.Fatalf("%s: the row or its timers changed: %+v, %d timers", c.name, rt, timers("h"))
+ }
+ if c.hold != "" {
+ if ok, err := r.ReleaseHold("h", c.hold); err != nil || !ok {
+ t.Fatal(ok, err)
+ }
+ }
+ }
+ if err := r.SetHold("h", "moving", holdOpA); err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := r.DeleteIfHoldWithTimers("h", holdOpA); err != nil || !ok {
+ t.Fatalf("delete under the holding op: %v %v", ok, err)
+ }
+ if rt, _ := r.Get("h"); rt != nil || timers("h") != 0 {
+ t.Fatalf("after the delete: %+v, %d timers", rt, timers("h"))
+ }
+ if rt, _ := r.Get("n"); rt == nil || timers("n") != 1 {
+ t.Fatalf("the neighbour changed: %+v, %d timers", rt, timers("n"))
+ }
+ // A replay finds no row held by the op and deletes nothing, timers
+ // written since included.
+ if err := r.UpsertTimer(Timer{Host: "h", Name: "late", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}); err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := r.DeleteIfHoldWithTimers("h", holdOpA); err != nil || ok || timers("h") != 1 {
+ t.Fatalf("replay: %v %v, %d timers", ok, err, timers("h"))
+ }
+}
+
func TestReleaseHoldEmptyOpMatchesNothing(t *testing.T) {
r := holdFixture(t, "h")
if ok, err := r.ReleaseHold("h", ""); err != nil || ok {
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index f8642d645..db75d890e 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -1148,6 +1148,36 @@ func (r *Registry) DeleteIfHold(host, op string) (bool, error) {
return n > 0, err
}
+// DeleteIfHoldWithTimers is the source's finalization (moves spec §7.9 step
+// 3, ruling H5(d)): host's row and every timer row of host are deleted in one
+// transaction, and only while op holds the row. false means nothing matched
+// and nothing was deleted, timers included, so a timers clear never runs
+// ahead of the row it belongs to or lands on a successor's rows.
+func (r *Registry) DeleteIfHoldWithTimers(host, op string) (bool, error) {
+ if op == "" {
+ return false, nil
+ }
+ tx, err := r.sql.Begin()
+ if err != nil {
+ return false, err
+ }
+ defer tx.Rollback()
+ res, err := tx.Exec(`DELETE FROM routes WHERE host = ? AND hold_op = ?`, host, op)
+ if err != nil {
+ return false, err
+ }
+ if n, err := res.RowsAffected(); err != nil || n == 0 {
+ return false, err
+ }
+ if _, err := tx.Exec(`DELETE FROM timers WHERE host = ?`, host); err != nil {
+ return false, err
+ }
+ if err := tx.Commit(); err != nil {
+ return false, err
+ }
+ return true, nil
+}
+
// HoldRows returns every row held by an operation (a non-empty hold_op), ordered by
// host. Over a fresh connection, like List, so a reader in another process
// sees the latest commit.