amadan

carlos / platform Public

lease: the cleaning mark, a state with no expiry written only over this move's lifted tombstone and lifted back with a raised cleanup count, every lock write of ours carrying its own write id so that no earlier lock body recurs, its lock writes single attempts sent again until answered, refused by every acquirer whatever its age, and answered to a wake with a 30-second retry; every hibernate-prefix site inventoried

MarkCleaning and UnmarkCleaning take one typed owner pair (lease.CleaningOwners,
ruling D10). LiftTombstone, MarkCleaning and UnmarkCleaning write lock.json
only through putLockUntilAnswered: one PutObject per send with the SDK's
retryer off, a fresh crypto/rand write_id per send, resent on
offlinecleanup.RetrySchedule (ruling D2) until answered or the operation's
10 s ends (ErrLockWriteUncertain). A lifted tombstone without a write_id is
rewritten as ours before it is marked. AcquireTTL reads the lock first and
refuses a cleaning: owner whatever its expiry; Tombstone, LockState and the
landing's lease step do the same. The activator answers a wake refused by a
mark with errLeaseCleaning: Retry-After 30 and its own sentence (MV-O268,
pending), and the landing waits with MV-O269 (pending). The lease gets a
clock hook and a lock attempt timeout, both registered in the seam audit
with their one setter (M37). leasetest.Holder holds a PUT in the network.
TestHibernatePrefixSitesAreInventoried pins every hibernate-prefix call site
with its class.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 2aefe5cce5040339633a54cfc4896a1c8a94bbae parent 8c689f8
18 files changed, +2093 −51
  • approved_moves_copy_test.go +1 −0
  • docs/superpowers/plans/2026-10-06-moves-phase1-copy.md +7 −0
  • internal/activator/activator.go +29 −1
  • internal/activator/backend_exec.go +7 −1
  • internal/activator/cleaning_wake_test.go +104 −0
  • internal/host/hibernateprefix_inventory_test.go +245 −0
  • internal/host/movesoffline_land.go +8 −2
  • internal/host/movesoffline_land_cleaning_test.go +52 −0
  • internal/host/schedule_wiring_test.go +1 −0
  • internal/lease/cleaning.go +316 −0
  • internal/lease/cleaning_test.go +1055 −0
  • internal/lease/export_test.go +21 −1
  • internal/lease/lease.go +10 −0
  • internal/lease/leasetest/hold.go +154 −0
  • internal/lease/lockstate.go +3 −1
  • internal/lease/tombstone.go +63 −42
  • internal/lease/tombstone_test.go +10 −0
  • internal/offlinecleanup/seams_test.go +7 −3

CI failed — run details

CI log 
=== gate ===
sigwho: pid 2 ppid 1 pgid 0 sid 0; child pid 3
.amadan/test-gate.sh
gate: 2026-10-24T15:21:16Z on aarch64, 32 cpus, 62110 MB available, go1.26.8 linux/arm64
gate: +0s start launchctx
gate: +0s start vet
gate: +0s start gofmt
gate: +2s start rest
gate: +23s start internal_console@0
gate: +23s start internal_console@1
gate: +23s start internal_console@2
gate: +23s start internal_console@3
gate: +23s start internal_console@4
gate: +23s start internal_console@5
gate: +23s start internal_console@6
gate: +23s start internal_console@7
gate: +23s waiting for internal_console's shards before the next split package
gate: +199s start root@0
gate: +199s start root@1
gate: +199s start root@2
gate: +199s waiting for root's shards before the next split package

amadan: step gate failed: signal: terminated

amadan: job failed at step gate — its output is above, not at the tail of this log
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index bd02f45ca..24f2f5b71 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -115,6 +115,7 @@ var movesCopyIDs = []string{
"MV-O257", "MV-O258",
"MV-O259", "MV-O260", "MV-O261", "MV-O262", "MV-O263",
"MV-O264", "MV-O265", "MV-O266", "MV-O267",
+ "MV-O268", "MV-O269",
}
// copyRow is one row of the record. The File cell is the file the sentence is
diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
index 48d4b4262..29fee2957 100644
--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md
@@ -828,3 +828,10 @@ Task 1 fix round 1 (pending Paul's review): the pages of a gone that records no
| MV-O265 | internal/host/movescleanup_retired.go (self-serve moves plan 4 Task 1 fix round 1) | cleanup of <host> skipped: its evidence does not validate | pending |
| MV-O266 | internal/host/movescleanup_retired.go (self-serve moves plan 4 Task 1 fix round 1) | cleanup of <host> skipped: its damaged record cannot be moved aside | pending |
| MV-O267 | internal/host/movescleanup_retired.go (self-serve moves plan 4 Task 1 fix round 1) | the move <id> finished, but nothing will clean up its source replica or what it left on <box>. clean them up by hand. | pending |
+
+Task 3 (pending Paul's review): MV-O268 is the activator's answer to a wake refused by a cleanup's mark on the host's lock (a 503 with `Retry-After: 30`; it begins with `instance`, which the scheduler's `deliverVia` reads as a wake refusal, `internal/host/schedule_wiring.go`), and MV-O269 the landing's lease item while a cleanup's mark holds the target's lock (a wait, beside MV-O111).
+
+| id | file | text | status |
+|---|---|---|---|
+| MV-O268 | internal/activator/activator.go (self-serve moves plan 4 Task 3) | instance is being cleaned up after a move. retry shortly | pending |
+| MV-O269 | internal/host/movesoffline_land.go (self-serve moves plan 4 Task 3) | a cleanup holds the target's lock | pending |
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index a0d56891b..81e6f9700 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1089,6 +1089,25 @@ func (e *errLeaseHeld) Error() string {
return fmt.Sprintf("instance lease held by %q (retry in %s)", e.holder, e.retryAfter)
}
+// errLeaseCleaning: the per-host lease is a cleanup's mark (self-serve moves
+// plan 4): a move finished and its source replica is being retired. The
+// mark's expiry is a sentinel no clock reaches, so the wait is not read from
+// it: the edge answers 503 with cleaningRetryAfter.
+type errLeaseCleaning struct{ holder string }
+
+// newErrLeaseCleaning is the wake's refusal for a lease marked by holder.
+func newErrLeaseCleaning(holder string) *errLeaseCleaning { return &errLeaseCleaning{holder: holder} }
+
+func (e *errLeaseCleaning) Error() string {
+ return fmt.Sprintf("instance replica marked by %q: a cleanup is running", e.holder)
+}
+
+// cleaningRetryAfter is the Retry-After (seconds) a wake refused by a
+// cleanup's mark sends. Fixed: a cleanup's run is at most 60 s
+// (cleanupRunDeadline), so a client back within half a minute finds a
+// finished cleanup promptly.
+const cleaningRetryAfter = 30
+
// errConfigRefused: this box cannot provide the instance's sealed config
// (no key, the wrong key, or it could not tell), so it will not take the
// lease (amadan #213). The condition outlives one request and a keyed
@@ -2242,9 +2261,13 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
}
var held *errLeaseHeld
- if errors.As(err, &held) {
+ var cleaning *errLeaseCleaning
+ switch {
+ case errors.As(err, &held):
log.Printf("activator: LEASE HELD for %s by %q — a previous incarnation hasn't released; refusing to wake (retry after %s)",
host, held.holder, held.retryAfter)
+ case errors.As(err, &cleaning):
+ log.Printf("activator: %s: a cleanup holds the replica (%q); refusing to wake", host, cleaning.holder)
}
return w, err
}
@@ -5412,6 +5435,7 @@ func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Rout
func (a *Service) serveWakeError(w http.ResponseWriter, r *http.Request, rt registry.Route, err error) bool {
var held *errLeaseHeld
+ var cleaning *errLeaseCleaning
var refused *errConfigRefused
switch {
case errors.Is(err, ErrFenced):
@@ -5423,6 +5447,10 @@ func (a *Service) serveWakeError(w http.ResponseWriter, r *http.Request, rt regi
w.Header().Set("Retry-After", strconv.Itoa(configRefusedRetryAfter))
edge.MarkAnswered(r)
http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
+ case errors.As(err, &cleaning):
+ w.Header().Set("Retry-After", strconv.Itoa(cleaningRetryAfter))
+ edge.MarkAnswered(r)
+ http.Error(w, "instance is being cleaned up after a move. retry shortly", http.StatusServiceUnavailable)
case errors.As(err, &held):
retry := int(held.retryAfter/time.Second) + 1
w.Header().Set("Retry-After", strconv.Itoa(retry))
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 738ff0ccc..03ec153e7 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -688,7 +688,8 @@ func (b *ExecBackend) LiveVersion(host string) (string, error) {
// Acquire claims the per-host single-writer lease. A contested lease is
// re-surfaced as errLeaseHeld (never a litestream type — the lease package's
// *HeldError is this package's only view of it) so the edge can 503 with a
-// Retry-After that tracks the holder's expiry.
+// Retry-After that tracks the holder's expiry; a cleanup's mark is
+// errLeaseCleaning, answered with a fixed Retry-After.
func (b *ExecBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
host := rt.Host
// Admission before the lease: the instance's config has to be
@@ -701,6 +702,11 @@ func (b *ExecBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, er
held, err := lease.Acquire(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region)
if err != nil {
var he *lease.HeldError
+ if errors.As(err, &he) && lease.IsCleaningOwner(he.Owner) {
+ // A cleanup's mark: its expiry is a sentinel, never a distance
+ // to wait, so it is not read (self-serve moves plan 4).
+ return nil, newErrLeaseCleaning(he.Owner)
+ }
if errors.As(err, &he) {
retry := time.Until(he.ExpiresAt)
if retry < time.Second {
diff --git a/internal/activator/cleaning_wake_test.go b/internal/activator/cleaning_wake_test.go
new file mode 100644
index 000000000..4806d8282
--- /dev/null
+++ b/internal/activator/cleaning_wake_test.go
@@ -0,0 +1,104 @@
+package activator
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// A wake refused by a cleanup's mark (self-serve moves plan 4, Task 3; review
+// 7's finding 7): the mark's expiry is a sentinel no clock reaches, so the
+// distance to it is no Retry-After. The answer is a fixed 30 seconds and its
+// own sentence, whatever the mark's expiry says; an ordinary holder's answer
+// is unchanged.
+func TestCleaningMarkAnswersAFiniteRetry(t *testing.T) {
+ const (
+ bucket = "hib"
+ host = "shop.example.com"
+ region = "test-region"
+ )
+ f := leasetest.NewVersionedBucket()
+ f.AddBucket(bucket)
+ srv, _ := leasetest.NewS3Server(f)
+ ts := httptest.NewServer(srv)
+ t.Cleanup(ts.Close)
+ 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")
+ t.Setenv("AWS_CONFIG_FILE", t.TempDir()+"/absent")
+ t.Setenv("AWS_SHARED_CREDENTIALS_FILE", t.TempDir()+"/absent")
+ t.Setenv("CARLOS_S3_ENDPOINT", ts.URL)
+
+ cfg := Config{Bucket: bucket, Prefix: "instances", Region: region}
+ loc := HibernatePrefix(cfg.Prefix, host)
+ lockKey := loc + "/lock.json"
+ move := lease.TombstonePrefix + "mv_01AAAAAAAAAAAAAAAAAAAAAAAA"
+ owners := lease.CleaningOwners{Move: move, Cleaning: lease.CleaningOwner("cl_01CCCCCCCCCCCCCCCCCCCCCCCC")}
+ ctx := context.Background()
+ if err := lease.Tombstone(ctx, bucket, loc, region, move, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ if lifted, err := lease.LiftTombstone(ctx, bucket, loc, region, func(o string) bool { return o == move }); err != nil || !lifted {
+ t.Fatalf("LiftTombstone = %v %v", lifted, err)
+ }
+ if err := lease.MarkCleaning(ctx, bucket, loc, region, owners); err != nil {
+ t.Fatal(err)
+ }
+
+ b := &ExecBackend{Cfg: cfg}
+ a, _, _, _ := testActivator(t, time.Minute)
+ rt := registry.Route{Host: host, Kind: "instance", Hibernate: true}
+ answer := func(t *testing.T) *httptest.ResponseRecorder {
+ t.Helper()
+ _, err := b.Acquire(ctx, rt)
+ if err == nil {
+ t.Fatal("Acquire took the lease")
+ }
+ rec := httptest.NewRecorder()
+ req := httptest.NewRequest("GET", "http://"+host+"/", nil)
+ a.serveWakeError(rec, req, rt, err)
+ return rec
+ }
+ wantCleaning := func(t *testing.T) {
+ t.Helper()
+ rec := answer(t)
+ first, _, _ := strings.Cut(rec.Body.String(), "\n")
+ if rec.Code != http.StatusServiceUnavailable || rec.Header().Get("Retry-After") != "30" || first != "instance is being cleaned up after a move. retry shortly" {
+ t.Fatalf("answer %d Retry-After %q body %q", rec.Code, rec.Header().Get("Retry-After"), rec.Body.String())
+ }
+ }
+
+ t.Run("the mark as written", func(t *testing.T) { wantCleaning(t) })
+
+ t.Run("the mark's expiry edited by hand to 2000-01-01", func(t *testing.T) {
+ raw, _ := f.Current(bucket, lockKey)
+ var m map[string]any
+ if err := json.Unmarshal(raw, &m); err != nil {
+ t.Fatal(err)
+ }
+ m["expires_at"] = "2000-01-01T00:00:00Z"
+ raw, _ = json.Marshal(m)
+ f.Put(bucket, lockKey, raw)
+ wantCleaning(t)
+ })
+
+ t.Run("an ordinary lease of another box", func(t *testing.T) {
+ raw, _ := json.Marshal(map[string]any{"generation": 9, "expires_at": time.Now().Add(40 * time.Second).UTC(), "owner": "box-b:1234"})
+ f.Put(bucket, lockKey, raw)
+ rec := answer(t)
+ // time.Until(expiry) is read a moment after the write, a little
+ // under 40 s, and serveWakeError rounds it down and adds one.
+ if rec.Code != http.StatusServiceUnavailable || rec.Header().Get("Retry-After") != "40" || rec.Body.String() != "instance is waking elsewhere; retry shortly\n" {
+ t.Fatalf("answer %d Retry-After %q body %q", rec.Code, rec.Header().Get("Retry-After"), rec.Body.String())
+ }
+ })
+}
diff --git a/internal/host/hibernateprefix_inventory_test.go b/internal/host/hibernateprefix_inventory_test.go
new file mode 100644
index 000000000..6f3b76dca
--- /dev/null
+++ b/internal/host/hibernateprefix_inventory_test.go
@@ -0,0 +1,245 @@
+package host
+
+import (
+ "fmt"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "go/types"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+ "testing"
+
+ "golang.org/x/tools/go/packages"
+)
+
+// The writer inventory of the hibernate prefix (self-serve moves plan 4,
+// Task 3; Global Constraint "The writers the mark excludes"). Every call of
+// activator.HibernatePrefix and of (activator.Config).hibernatePrefix in a
+// non-test file under internal/ is listed here with its class, so a new
+// site, or a removed one, fails until the list is edited and the new site's
+// relation to the cleaning mark is stated:
+//
+// - reader: reads under the prefix and writes nothing there;
+// - lease-gated writer: writes under the prefix only while it holds the
+// host's lease, which every acquirer refuses while a cleanup's mark is
+// on the lock (lease.AcquireTTL's pre-read, litestream's own read);
+// - tombstone-gated writer: writes the lock only through lease.Tombstone,
+// which refuses a mark, or lifts only this move's tombstone;
+// - build-state only: writes only build-state.json, which no cleanup
+// inventories;
+// - delete only: deletes by key and never writes a version a cleanup
+// could adopt (the seed rollback; the seed finish refuses a held lock,
+// and LockState answers a mark held);
+// - a move into the region (hostname key): the target landing's, which
+// holds host/<hostname> and whose lease step waits on a mark;
+// - excluded: <reason>.
+//
+// A site is "<file under internal/> <function>#<n>", n counting that
+// function's calls of either name in source order.
+var hibernatePrefixSites = map[string]string{
+ "activator/activator.go Config.hibernatePrefix#1": "excluded: the method's own body",
+ // The wake: ExecBackend.Acquire takes the host's lease (step 1 of
+ // doWake); the restore, the staged-replica promotion, BeforeExec's
+ // build-state write and litestream replication all run under it.
+ "activator/backend_exec.go ExecBackend.Acquire#1": "lease-gated writer",
+ "activator/backend_exec.go ExecBackend.restorePlain#1": "reader",
+ "activator/backend_exec.go ExecBackend.restorePlain#2": "excluded: a log line",
+ "activator/backend_exec.go ExecBackend.restorePlain#3": "excluded: a log line",
+ "activator/backend_exec.go ExecBackend.restorePlain#4": "excluded: a log line",
+ "activator/backend_exec.go ExecBackend.promoteStagedReplica#1": "lease-gated writer",
+ "activator/beforeexec.go Service.recordBuild#1": "build-state only",
+ "activator/seedplace.go ExecBackend.seededRestore#1": "reader",
+ "console/apidb.go Server.resolveReplica#1": "reader",
+ "console/apiinstancemove.go Server.handleAPIInstanceMoveRequest#1": "excluded: compares a claim holder's location; reads and writes nothing",
+ "console/createleftovers.go Server.hostHoldsReplicaData#1": "reader",
+ "console/seedfinish.go Server.deleteSeedReplica#1": "delete only",
+ "host/buildstate.go newBuildStates#1": "build-state only",
+ "host/e2e_bridge.go sealWorld.arm#1": "lease-gated writer",
+ "host/e2e_bridge.go sealWorld.writeRows#1": "lease-gated writer",
+ "host/movescleanup_retired.go movesWorker.recordRetired#1": "reader",
+ "host/movescut.go writerLease#1": "excluded: names the writer's lease location for the cut's admission; reads and writes nothing",
+ "host/movesoffline_land.go landRun.lease#1": "a move into the region (hostname key)",
+ "host/movesoffline_land.go movesWorker.offlineTargetReady#1": "a move into the region (hostname key)",
+ "host/movesoffline_reports.go movesWorker.listReplicas#1": "reader",
+ "host/movesoffline_seal.go moveCfg.tombstoneAt#1": "tombstone-gated writer",
+ "host/movespass.go armMovesPass#1": "lease-gated writer",
+ "host/movespass.go armPhase1#1": "reader",
+ "host/movespass.go legacyMoveRound#1": "reader",
+ "host/seedwiring.go deleteReplica#1": "delete only",
+ "ops/hibernate_verify.go VerifyHibernating#1": "reader",
+}
+
+var hibernatePrefixClasses = map[string]bool{
+ "reader": true, "lease-gated writer": true, "tombstone-gated writer": true,
+ "build-state only": true, "delete only": true, "a move into the region (hostname key)": true,
+}
+
+const activatorPkg = "github.com/carlosframework/platform/internal/activator"
+
+func TestHibernatePrefixSitesAreInventoried(t *testing.T) {
+ root, err := filepath.Abs("..")
+ if err != nil {
+ t.Fatal(err)
+ }
+ // The syntactic pass: every non-test file under internal/, whatever its
+ // build constraints, and the directories that name either function.
+ syntactic := map[string]bool{}
+ dirs := map[string]bool{}
+ fset := token.NewFileSet()
+ err = filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
+ if d.IsDir() {
+ if n := d.Name(); p != root && (n == "testdata" || strings.HasPrefix(n, ".") || strings.HasPrefix(n, "_")) {
+ return filepath.SkipDir
+ }
+ return nil
+ }
+ if !strings.HasSuffix(p, ".go") || strings.HasSuffix(p, "_test.go") {
+ return nil
+ }
+ src, err := os.ReadFile(p)
+ if err != nil {
+ return err
+ }
+ if !strings.Contains(string(src), "ibernatePrefix") {
+ return nil
+ }
+ f, err := parser.ParseFile(fset, p, src, parser.SkipObjectResolution)
+ if err != nil {
+ return err
+ }
+ rel, _ := filepath.Rel(root, p)
+ rel = filepath.ToSlash(rel)
+ for _, s := range prefixSites(rel, f, func(call *ast.CallExpr) bool {
+ sel, ok := call.Fun.(*ast.SelectorExpr)
+ if ok {
+ return sel.Sel.Name == "HibernatePrefix" || sel.Sel.Name == "hibernatePrefix"
+ }
+ id, ok := call.Fun.(*ast.Ident)
+ return ok && id.Name == "HibernatePrefix" && f.Name.Name == "activator"
+ }) {
+ syntactic[s] = true
+ }
+ dirs["./"+filepath.ToSlash(filepath.Dir(rel))] = true
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ // The typed pass: the calls whose callee is one of the two functions.
+ var patterns []string
+ for d := range dirs {
+ patterns = append(patterns, d)
+ }
+ sort.Strings(patterns)
+ cfg := &packages.Config{Mode: packages.NeedName | packages.NeedFiles | packages.NeedSyntax | packages.NeedTypes | packages.NeedTypesInfo | packages.NeedImports, Dir: root}
+ pkgs, err := packages.Load(cfg, patterns...)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if packages.PrintErrors(pkgs) > 0 {
+ t.Fatal("the packages naming the hibernate prefix do not load")
+ }
+ typed := map[string]bool{}
+ for _, p := range pkgs {
+ for _, f := range p.Syntax {
+ name := p.Fset.Position(f.Pos()).Filename
+ rel, _ := filepath.Rel(root, name)
+ info := p.TypesInfo
+ for _, s := range prefixSites(filepath.ToSlash(rel), f, func(call *ast.CallExpr) bool {
+ var id *ast.Ident
+ switch fn := call.Fun.(type) {
+ case *ast.Ident:
+ id = fn
+ case *ast.SelectorExpr:
+ id = fn.Sel
+ }
+ obj, _ := info.Uses[id].(*types.Func)
+ if obj == nil || obj.Pkg() == nil || obj.Pkg().Path() != activatorPkg {
+ return false
+ }
+ if obj.Name() == "HibernatePrefix" {
+ return obj.Type().(*types.Signature).Recv() == nil
+ }
+ if obj.Name() == "hibernatePrefix" {
+ recv := obj.Type().(*types.Signature).Recv()
+ return recv != nil && types.TypeString(recv.Type(), nil) == activatorPkg+".Config"
+ }
+ return false
+ }) {
+ typed[s] = true
+ }
+ }
+ }
+ for s := range syntactic {
+ if !typed[s] {
+ t.Errorf("%s names the hibernate prefix where the type check did not confirm it (a build-constrained file, or another function of that name): inventory it by hand", s)
+ }
+ }
+
+ var unlisted []string
+ for s := range typed {
+ class, ok := hibernatePrefixSites[s]
+ switch {
+ case !ok:
+ unlisted = append(unlisted, s)
+ case !hibernatePrefixClasses[class] && !strings.HasPrefix(class, "excluded: "):
+ t.Errorf("%s: class %q is not one the Global Constraint lists", s, class)
+ }
+ }
+ sort.Strings(unlisted)
+ for _, s := range unlisted {
+ t.Errorf("a new hibernate-prefix site %s: add it to hibernatePrefixSites with its class", s)
+ }
+ for s := range hibernatePrefixSites {
+ if !typed[s] {
+ t.Errorf("hibernatePrefixSites lists %s, which no longer calls the hibernate prefix", s)
+ }
+ }
+}
+
+// prefixSites names the calls match accepts in f, by enclosing function.
+func prefixSites(rel string, f *ast.File, match func(*ast.CallExpr) bool) []string {
+ var out []string
+ for _, d := range f.Decls {
+ fd, ok := d.(*ast.FuncDecl)
+ if !ok || fd.Body == nil {
+ if gd, ok := d.(*ast.GenDecl); ok {
+ ast.Inspect(gd, func(n ast.Node) bool {
+ if call, ok := n.(*ast.CallExpr); ok && match(call) {
+ out = append(out, rel+" <package scope>#"+strconv.Itoa(len(out)+1))
+ }
+ return true
+ })
+ }
+ continue
+ }
+ name := fd.Name.Name
+ if fd.Recv != nil && len(fd.Recv.List) > 0 {
+ e := fd.Recv.List[0].Type
+ if s, ok := e.(*ast.StarExpr); ok {
+ e = s.X
+ }
+ if id, ok := e.(*ast.Ident); ok {
+ name = id.Name + "." + name
+ }
+ }
+ n := 0
+ ast.Inspect(fd.Body, func(x ast.Node) bool {
+ if call, ok := x.(*ast.CallExpr); ok && match(call) {
+ n++
+ out = append(out, fmt.Sprintf("%s %s#%d", rel, name, n))
+ }
+ return true
+ })
+ }
+ return out
+}
diff --git a/internal/host/movesoffline_land.go b/internal/host/movesoffline_land.go
index 39aa6c120..1f6e63b3a 100644
--- a/internal/host/movesoffline_land.go
+++ b/internal/host/movesoffline_land.go
@@ -604,8 +604,9 @@ func (r *landRun) keyAdoption() {
// lease: the target bucket's lock for the host is absent, a lifted
// tombstone, or an ordinary lease that has expired. A tombstone of another
-// move is lifted. Nothing here ever takes the lease. Expiry is read on the
-// wall clock, as the lease package reads it.
+// move is lifted; a cleanup's mark is waited on whatever its expiry. Nothing
+// here ever takes the lease. Expiry is read on the wall clock, as the lease
+// package reads it.
func (r *landRun) lease(ctx context.Context) {
cfg := r.cfg()
loc := activator.HibernatePrefix(cfg.OwnPrefix, r.host)
@@ -624,6 +625,11 @@ func (r *landRun) lease(ctx context.Context) {
case owner == moves.TombstoneOwner(r.m.ID):
r.landFail("lease", "the target's lock is this move's own tombstone")
return
+ case lease.IsCleaningOwner(owner):
+ // A cleanup's mark: held whatever its expiry says, and lifted only
+ // by the cleanup (self-serve moves plan 4).
+ r.landWait("lease", "a cleanup holds the target's lock")
+ return
case !strings.HasPrefix(owner, lease.TombstonePrefix):
if time.Now().After(expires) {
r.ok("lease")
diff --git a/internal/host/movesoffline_land_cleaning_test.go b/internal/host/movesoffline_land_cleaning_test.go
new file mode 100644
index 000000000..7ed38ec00
--- /dev/null
+++ b/internal/host/movesoffline_land_cleaning_test.go
@@ -0,0 +1,52 @@
+//go:build linux
+
+package host
+
+import (
+ "encoding/json"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// A cleanup's mark on the target's lock (self-serve moves plan 4, Task 3)
+// holds the landing's lease step whatever its expiry says: the step waits
+// (a landed report with the lease item not OK), never passes it as an
+// expired ordinary lease and never fails the move.
+func TestLandingWaitsOnACleaningLockWhateverItsExpiry(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ expires time.Time
+ }{
+ {"the sentinel expiry", lease.CleaningExpiry},
+ {"an expiry edited to the past", time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ lw := newLandWorld(t)
+ raw, err := json.Marshal(map[string]any{"generation": 3, "expires_at": c.expires, "owner": lease.CleaningOwner("cl_01CCCCCCCCCCCCCCCCCCCCCCCC"), "write_id": "0123456789abcdef0123456789abcdef"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ lockKey := activator.HibernatePrefix("instances", offHost) + "/lock.json"
+ lw.s3.f.Put("data-uk", lockKey, raw)
+ lw.flipAndLand(2)
+ if st := lw.state(); st != provision.OfflineLanding {
+ t.Fatalf("state %s", st)
+ }
+ r := lw.report()
+ if it := r.Items["lease"]; it.OK || it.Reason != "a cleanup holds the target's lock" {
+ t.Fatalf("lease item %+v", it)
+ }
+ if r.Phase != moves.OfflinePhaseLanded {
+ t.Fatalf("phase %s, want %s (a wait, not a failure)", r.Phase, moves.OfflinePhaseLanded)
+ }
+ if now, _ := lw.s3.f.Current("data-uk", lockKey); string(now) != string(raw) {
+ t.Fatalf("the landing wrote the lock: %s", now)
+ }
+ })
+ }
+}
diff --git a/internal/host/schedule_wiring_test.go b/internal/host/schedule_wiring_test.go
index aa221a854..d8c6aa0ac 100644
--- a/internal/host/schedule_wiring_test.go
+++ b/internal/host/schedule_wiring_test.go
@@ -30,6 +30,7 @@ func TestDeliverViaHandlerMapsWakeRefusals(t *testing.T) {
{503, "instance failed to wake\n", 0, "instance failed to wake"},
{503, "instance is waking; retry shortly\n", 0, "instance is waking; retry shortly"},
{503, "instance is waking elsewhere; retry shortly\n", 0, "instance is waking elsewhere"},
+ {503, "instance is being cleaned up after a move. retry shortly\n", 0, "instance is being cleaned up"},
{503, "app says no", 503, ""},
{500, "instance blew up", 500, ""}, // only a 503 is a wake refusal
}
diff --git a/internal/lease/cleaning.go b/internal/lease/cleaning.go
new file mode 100644
index 000000000..2744ff072
--- /dev/null
+++ b/internal/lease/cleaning.go
@@ -0,0 +1,316 @@
+package lease
+
+// The cleaning mark (self-serve moves plan 4, Task 3): the state of a host's
+// lock.json while a cleanup retires a finished move's source replica.
+//
+// A mark is owned by "cleaning:<cleanup id>" and expires at CleaningExpiry, a
+// time no clock reaches. It is a state, not a timed lease: MarkCleaning
+// writes it only over this move's lifted tombstone, nothing renews it, and
+// UnmarkCleaning alone lifts it, by writing this move's lifted tombstone
+// back with its cleanup count raised. Every acquirer refuses it whatever its
+// expiry says (AcquireTTL's pre-read, Tombstone, LockState's held).
+//
+// No lock body recurs. A conditional write that read ETag E lands whenever E
+// is current, and a single-part PUT's ETag is a digest of its bytes, so a
+// body current twice would make every outstanding write that read it valid
+// again. Every lock body this package writes itself (the lift, the mark and
+// the unmark) carries a write_id drawn afresh for each HTTP attempt, and
+// every attempt is a single request with the SDK's retryer off
+// (putLockUntilAnswered), so each body reaches the bucket at most once. The
+// mark and the unmark are conditioned only on bodies this package wrote:
+// MarkCleaning first rewrites a lifted tombstone that has no write_id (a lift
+// from before this plan, or a tombstone that expired unlifted) as one of
+// ours. Once such a body is replaced it never exists again, and no write
+// that named it can land.
+
+import (
+ "bytes"
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "strings"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ lss3 "github.com/benbjohnson/litestream/s3"
+
+ "github.com/carlosframework/platform/internal/offlinecleanup"
+)
+
+// CleaningPrefix starts the owner of every cleaning mark.
+const CleaningPrefix = "cleaning:"
+
+// CleaningExpiry is the expiry every mark is written with: a time no clock
+// reaches, so litestream's own AcquireLease refuses a mark on the read that
+// conditions its write. litestream's lease needs an expiry, and readLock
+// refuses a zero one.
+var CleaningExpiry = time.Date(9999, 12, 31, 0, 0, 0, 0, time.UTC)
+
+// CleaningOwner is the mark's owner for the cleanup id.
+func CleaningOwner(id string) string { return CleaningPrefix + id }
+
+// IsCleaningOwner reports whether owner is a cleaning mark's: the prefix and
+// a non-empty id.
+func IsCleaningOwner(owner string) bool {
+ return strings.HasPrefix(owner, CleaningPrefix) && len(owner) > len(CleaningPrefix)
+}
+
+var (
+ // ErrNotOurLock is MarkCleaning's refusal: the lock is absent, another
+ // owner's, or this move's tombstone still live. Nothing was written.
+ ErrNotOurLock = errors.New("lease: the lock is not this move's lifted tombstone; nothing was written")
+ // ErrLockWriteUncertain reports a lock write no attempt of which was
+ // answered before the context ended: one may still land. The caller's
+ // next visit reads the lock afresh.
+ ErrLockWriteUncertain = errors.New("lease: the lock write was not answered; an attempt may still land")
+ // ErrCleaningOwners refuses an owner pair whose Move is not moved:<id> or
+ // whose Cleaning is not cleaning:<id>.
+ ErrCleaningOwners = errors.New("lease: a cleaning mark names a moved:<id> and a cleaning:<id> owner")
+)
+
+// CleaningOwners is the one owner pair the mark and the unmark take: the
+// move whose lifted tombstone the mark replaces, and the cleanup's mark.
+type CleaningOwners struct {
+ Move string // moved:<move id>
+ Cleaning string // cleaning:<cleanup id>
+}
+
+func (o CleaningOwners) valid() bool { return tombstoneOwnerOK(o.Move) && IsCleaningOwner(o.Cleaning) }
+
+// leaseNow is what this package reads as now when it decides whether a lock
+// has expired. Tests move it with SetClockForTest (its only setter, audited
+// by offlinecleanup's TestEveryE2ESeamHasOneSetter). litestream's own reads
+// keep the wall clock.
+var leaseNow = time.Now
+
+// lockAttemptTimeout bounds one HTTP attempt of a lock write.
+// SetLockAttemptTimeoutForTest is its only setter.
+var lockAttemptTimeout = 2 * time.Second
+
+// lockWriteBudget bounds a whole lock write, every resend included: the
+// operation's 10 s. A caller's earlier deadline ends it sooner.
+const lockWriteBudget = 10 * time.Second
+
+// lockBody is a lock as this package writes it: litestream's Lease fields
+// with their tags (litestream@v0.5.14/leaser.go:31-36), the cleanup count,
+// and the per-attempt write id. Older readers decode it into litestream's
+// Lease and ignore the last two.
+type lockBody struct {
+ Generation int64 `json:"generation"`
+ ExpiresAt time.Time `json:"expires_at"`
+ Owner string `json:"owner,omitempty"`
+ Cleanups uint64 `json:"cleanups,omitempty"`
+ WriteID string `json:"write_id,omitempty"`
+}
+
+// newWriteID is 16 bytes from crypto/rand as 32 lower-case hex digits.
+func newWriteID() string {
+ var b [16]byte
+ if _, err := rand.Read(b[:]); err != nil {
+ panic("lease: crypto/rand: " + err.Error())
+ }
+ return hex.EncodeToString(b[:])
+}
+
+// putLockUntilAnswered writes body to key with If-Match ifMatch, one HTTP
+// attempt per send under lockAttemptTimeout and with the SDK's retryer off,
+// each send carrying a write_id of its own drawn just before it is
+// marshalled. An answer is returned: success 200, a lost condition 412, an
+// absent key 404 (AWS answers a conditional PUT on a missing key 404, some
+// stores 412). Anything else (a timeout, a transport error, a 5xx) sends the
+// same fields under the same condition again, with a new write_id, on
+// offlinecleanup.RetrySchedule, until ctx or lockWriteBudget ends:
+// ErrLockWriteUncertain.
+func putLockUntilAnswered(ctx context.Context, leaser *lss3.Leaser, key string, body lockBody, ifMatch string) (status int, err error) {
+ ctx, cancel := context.WithTimeout(ctx, lockWriteBudget)
+ defer cancel()
+ var last error
+ for n := 0; ; n++ {
+ send := body
+ send.WriteID = newWriteID()
+ data, err := json.Marshal(&send)
+ if err != nil {
+ return 0, fmt.Errorf("lease: encode %s: %w", key, err)
+ }
+ actx, acancel := context.WithTimeout(ctx, lockAttemptTimeout)
+ _, err = leaser.Client().PutObject(actx, &awss3.PutObjectInput{
+ Bucket: aws.String(leaser.Bucket),
+ Key: aws.String(key),
+ Body: bytes.NewReader(data),
+ ContentType: aws.String("application/json"),
+ IfMatch: aws.String(ifMatch),
+ }, func(o *awss3.Options) { o.Retryer = aws.NopRetryer{} })
+ acancel()
+ if err == nil {
+ return http.StatusOK, nil
+ }
+ switch code, st := apiCode(err); {
+ case code == "PreconditionFailed" || st == http.StatusPreconditionFailed:
+ return http.StatusPreconditionFailed, nil
+ case code == "NoSuchKey" || (st == http.StatusNotFound && code != "NoSuchBucket"):
+ return http.StatusNotFound, nil
+ }
+ last = err
+ t := time.NewTimer(offlinecleanup.RetrySchedule.Wait(n))
+ select {
+ case <-ctx.Done():
+ t.Stop()
+ return 0, fmt.Errorf("lease: write %s: %w (last attempt: %v)", key, ErrLockWriteUncertain, last)
+ case <-t.C:
+ }
+ if ctx.Err() != nil {
+ return 0, fmt.Errorf("lease: write %s: %w (last attempt: %v)", key, ErrLockWriteUncertain, last)
+ }
+ }
+}
+
+// MarkCleaning writes the cleanup's mark over this move's lifted tombstone at
+// s3://bucket/<location>/lock.json: {generation: as read, expires_at:
+// CleaningExpiry, owner: o.Cleaning, cleanups: as read}, with If-Match the
+// ETag read.
+//
+// - the mark is already there: nil, nothing written (a resumed mark);
+// - this move's lifted tombstone without a write_id: first rewritten as
+// ours (the same fields, a write_id), whatever that write answers, and
+// read again, at most twice (still none: ErrTombstoneChanged), so the
+// mark is only ever conditioned on a body this package wrote;
+// - this move's lifted tombstone with a write_id: marked. A lost condition
+// is read back: our mark (an earlier attempt landed) is nil, anything
+// else ErrTombstoneChanged;
+// - no attempt answered: ErrLockWriteUncertain;
+// - absent, another owner's, or this move's tombstone still live:
+// ErrNotOurLock, nothing written.
+func MarkCleaning(ctx context.Context, bucket, location, region string, o CleaningOwners) error {
+ if !o.valid() {
+ return ErrCleaningOwners
+ }
+ leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, "", 0)
+ if err != nil {
+ return err
+ }
+ cur, etag, absent, err := readLockBody(ctx, leaser, key)
+ if err != nil {
+ return err
+ }
+ for normalised := 0; ; normalised++ {
+ switch {
+ case !absent && cur.Owner == o.Cleaning:
+ return nil
+ case absent || cur.Owner != o.Move || !IsLiftedTombstone(cur.Owner, cur.ExpiresAt):
+ return fmt.Errorf("mark cleaning: %w (%s)", ErrNotOurLock, describeLock(cur, absent))
+ }
+ if cur.WriteID != "" {
+ break
+ }
+ if normalised == 2 {
+ return fmt.Errorf("mark cleaning: %w (the lifted tombstone still has no write_id)", ErrTombstoneChanged)
+ }
+ if _, err := putLockUntilAnswered(ctx, leaser, key, lockBody{Generation: cur.Generation, ExpiresAt: liftedExpiry, Owner: o.Move, Cleanups: cur.Cleanups}, etag); err != nil {
+ return fmt.Errorf("mark cleaning: %w", err)
+ }
+ if cur, etag, absent, err = readLockBody(ctx, leaser, key); err != nil {
+ return err
+ }
+ }
+ status, err := putLockUntilAnswered(ctx, leaser, key, lockBody{Generation: cur.Generation, ExpiresAt: CleaningExpiry, Owner: o.Cleaning, Cleanups: cur.Cleanups}, etag)
+ if err != nil {
+ return fmt.Errorf("mark cleaning: %w", err)
+ }
+ if status == http.StatusOK {
+ return nil
+ }
+ now, _, gone, err := readLockBody(ctx, leaser, key)
+ if err != nil {
+ return err
+ }
+ if !gone && now.Owner == o.Cleaning {
+ return nil // an earlier attempt of ours landed: the mark is settled
+ }
+ return fmt.Errorf("mark cleaning: %w", ErrTombstoneChanged)
+}
+
+// UnmarkCleaning lifts the cleanup's mark by writing this move's lifted
+// tombstone with the count raised: {generation: as read, expires_at:
+// 1970-01-01, owner: o.Move, cleanups: as read + 1}, If-Match the ETag read.
+// It writes over our mark, and also over this move's lifted tombstone when
+// no mark is there (never sent, still travelling, or lifted by an earlier
+// visit): the new body has a write_id no other write has, so a mark attempt
+// conditioned on an earlier ETag can never land after it.
+//
+// - written: nil; the key absent at the write: nil (no conditional write
+// of ours can land on a missing key);
+// - a lost condition is read back: this move's lifted tombstone with a
+// higher count than read (an earlier attempt landed) is nil, anything
+// else ErrTombstoneChanged (a late mark of ours landed in between, say:
+// the next visit unmarks it);
+// - no attempt answered: ErrLockWriteUncertain;
+// - absent, another owner's, or this move's tombstone still live: nil and
+// nothing written. No attempt of ours can land there: each names the
+// ETag of our mark or of this move's lifted tombstone, and none of
+// those bodies recurs.
+func UnmarkCleaning(ctx context.Context, bucket, location, region string, o CleaningOwners) error {
+ if !o.valid() {
+ return ErrCleaningOwners
+ }
+ leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, "", 0)
+ if err != nil {
+ return err
+ }
+ cur, etag, absent, err := readLockBody(ctx, leaser, key)
+ if err != nil {
+ return err
+ }
+ if absent || !(cur.Owner == o.Cleaning || (cur.Owner == o.Move && IsLiftedTombstone(cur.Owner, cur.ExpiresAt))) {
+ return nil
+ }
+ status, err := putLockUntilAnswered(ctx, leaser, key, lockBody{Generation: cur.Generation, ExpiresAt: liftedExpiry, Owner: o.Move, Cleanups: cur.Cleanups + 1}, etag)
+ if err != nil {
+ return fmt.Errorf("unmark cleaning: %w", err)
+ }
+ switch status {
+ case http.StatusOK, http.StatusNotFound:
+ return nil
+ }
+ now, _, gone, err := readLockBody(ctx, leaser, key)
+ if err != nil {
+ return err
+ }
+ if !gone && now.Owner == o.Move && IsLiftedTombstone(now.Owner, now.ExpiresAt) && now.Cleanups > cur.Cleanups {
+ return nil // an earlier attempt of ours landed
+ }
+ return fmt.Errorf("unmark cleaning: %w", ErrTombstoneChanged)
+}
+
+// refuseCleaningMark is AcquireTTL's pre-read of the lock litestream's
+// leaser is about to take (its key, lss3.Leaser.lockKey): a cleaning: owner
+// is a *HeldError naming the mark, whatever its expiry, with nothing
+// written. A lock that cannot be read or decoded is an error, as
+// AcquireLease's own read would make it; any other lock, or none, is nil and
+// AcquireLease decides as before (a lock with a zero expiry or no ETag
+// included: this read applies no checks of its own beyond the owner).
+func refuseCleaningMark(ctx context.Context, leaser *lss3.Leaser) error {
+ key := "lock.json"
+ if leaser.Path != "" {
+ key = leaser.Path + "/lock.json"
+ }
+ cur, _, absent, err := fetchLock(ctx, leaser, key)
+ if err != nil {
+ return fmt.Errorf("acquire lease: %w", err)
+ }
+ if !absent && IsCleaningOwner(cur.Owner) {
+ return fmt.Errorf("acquire lease: %w", &HeldError{Owner: cur.Owner, ExpiresAt: cur.ExpiresAt})
+ }
+ return nil
+}
+
+func describeLock(l *lockBody, absent bool) string {
+ if absent || l == nil {
+ return "absent"
+ }
+ return fmt.Sprintf("owner %q, expires %s", l.Owner, l.ExpiresAt.UTC().Format(time.RFC3339))
+}
diff --git a/internal/lease/cleaning_test.go b/internal/lease/cleaning_test.go
new file mode 100644
index 000000000..7a62d261b
--- /dev/null
+++ b/internal/lease/cleaning_test.go
@@ -0,0 +1,1055 @@
+package lease_test
+
+import (
+ "bytes"
+ "context"
+ "crypto/md5"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "go/ast"
+ "go/token"
+ "go/types"
+ "net/http"
+ "path/filepath"
+ "regexp"
+ "sort"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ "golang.org/x/tools/go/packages"
+
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/moves"
+)
+
+// The cleaning mark (self-serve moves plan 4, Task 3): a state with no
+// expiry, written only over this move's lifted tombstone and lifted back
+// with a raised cleanup count; every lock write of ours carries its own
+// write_id, so no body of ours is current twice and no conditional write
+// that read one can land later.
+
+const (
+ cleanID = "cl_01CCCCCCCCCCCCCCCCCCCCCCCC"
+ cleanID2 = "cl_01DDDDDDDDDDDDDDDDDDDDDDDD"
+ tombLockKy = tombLoc + "/lock.json"
+ // attempt is the lock write attempt deadline the tests run under.
+ attempt = 200 * time.Millisecond
+)
+
+var owners = lease.CleaningOwners{Move: tombMove, Cleaning: lease.CleaningOwner(cleanID)}
+
+// body is a lock as this plan writes it, every field.
+type body struct {
+ Generation int64 `json:"generation"`
+ ExpiresAt time.Time `json:"expires_at"`
+ Owner string `json:"owner,omitempty"`
+ Cleanups uint64 `json:"cleanups,omitempty"`
+ WriteID string `json:"write_id,omitempty"`
+}
+
+var writeIDPattern = regexp.MustCompile(`^[0-9a-f]{32}$`)
+
+func etagOf(b []byte) string { s := md5.Sum(b); return hex.EncodeToString(s[:]) }
+
+func decodeBody(t *testing.T, raw []byte) body {
+ t.Helper()
+ var b body
+ if err := json.Unmarshal(raw, &b); err != nil {
+ t.Fatalf("not a lock: %v (%s)", err, raw)
+ }
+ return b
+}
+
+// current is the lock's bytes, decoded, and its ETag.
+func (s *tombStore) current(t *testing.T) (body, []byte, string) {
+ t.Helper()
+ raw, ok := s.Current("data", tombLockKy)
+ if !ok {
+ t.Fatal("lock.json is absent")
+ }
+ return decodeBody(t, raw), raw, etagOf(raw)
+}
+
+func (s *tombStore) absent(t *testing.T) bool {
+ t.Helper()
+ _, ok := s.Current("data", tombLockKy)
+ return !ok
+}
+
+// holdFixture is tombFixture with a leasetest.Holder in front of the store
+// (behind the write hook), and the lock write attempt deadline shortened.
+func holdFixture(t *testing.T) (*tombStore, *leasetest.Holder) {
+ t.Helper()
+ var h *leasetest.Holder
+ s := tombFixtureWrap(t, func(next http.Handler) http.Handler {
+ h = leasetest.NewHolder(next)
+ return h
+ })
+ t.Cleanup(lease.SetLockAttemptTimeoutForTest(attempt))
+ return s, h
+}
+
+// hold holds the next n PUTs of the lock and releases them at cleanup too, so
+// the test server can close.
+func hold(t *testing.T, h *leasetest.Holder, n int) func() {
+ rel := h.HoldPuts(tombLockKy, n)
+ t.Cleanup(rel)
+ return rel
+}
+
+// liftedAt writes a live tombstone of this move at gen and lifts it with the
+// production call: this move's lifted tombstone as this plan writes it.
+func liftedAt(t *testing.T, s *tombStore, gen int64) (body, []byte, string) {
+ t.Helper()
+ s.putLock(t, tombLoc, tombMove, gen, time.Now().Add(time.Hour))
+ lifted, err := lease.LiftTombstone(context.Background(), "data", tombLoc, tombRegion, isMine(tombMove))
+ if err != nil || !lifted {
+ t.Fatalf("LiftTombstone = %v, %v", lifted, err)
+ }
+ return s.current(t)
+}
+
+func requireUntouched(t *testing.T, s *tombStore, writes int, raw []byte) {
+ t.Helper()
+ if s.writeCount() != writes {
+ t.Fatalf("%d writes reached lock.json, want none", s.writeCount()-writes)
+ }
+ now, ok := s.Current("data", tombLockKy)
+ if ok != (raw != nil) || !bytes.Equal(now, raw) {
+ t.Fatalf("lock.json changed: %s -> %s", raw, now)
+ }
+}
+
+func TestMarkOnlyFromThisMovesLiftedTombstone(t *testing.T) {
+ ctx := context.Background()
+ t.Run("this move's lifted tombstone", func(t *testing.T) {
+ s, _ := holdFixture(t)
+ liftedAt(t, s, 4)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ l, _, _ := s.current(t)
+ if l.Owner != lease.CleaningOwner(cleanID) || !l.ExpiresAt.Equal(lease.CleaningExpiry) || l.Generation != 4 {
+ t.Fatalf("mark = %+v", l)
+ }
+ })
+ refused := []struct {
+ name string
+ set func(t *testing.T, s *tombStore)
+ }{
+ {"another move's lifted tombstone", func(t *testing.T, s *tombStore) {
+ s.putLock(t, tombLoc, tombOther, 3, time.Unix(0, 0))
+ }},
+ {"this move's live tombstone", func(t *testing.T, s *tombStore) {
+ s.putLock(t, tombLoc, tombMove, 3, time.Now().Add(time.Hour))
+ }},
+ {"an ordinary lease", func(t *testing.T, s *tombStore) {
+ held, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = held.Release(context.Background()) })
+ }},
+ {"absent", func(t *testing.T, s *tombStore) {}},
+ }
+ for _, c := range refused {
+ t.Run(c.name, func(t *testing.T) {
+ s, _ := holdFixture(t)
+ c.set(t, s)
+ raw, _ := s.Current("data", tombLockKy)
+ writes := s.writeCount()
+ err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ if !errors.Is(err, lease.ErrNotOurLock) {
+ t.Fatalf("MarkCleaning = %v, want ErrNotOurLock", err)
+ }
+ requireUntouched(t, s, writes, raw)
+ })
+ }
+}
+
+// An Acquire that lands between the mark's read and its PUT takes the
+// lifted tombstone; the mark's condition then fails and the holder keeps
+// its lease.
+func TestMarkRacesAnAcquire(t *testing.T) {
+ s, _ := holdFixture(t)
+ ctx := context.Background()
+ liftedAt(t, s, 4)
+ var held *lease.Held
+ var acqErr error
+ s.beforeWrite(func(string) { held, acqErr = lease.Acquire(ctx, "data", tombLoc, tombRegion) })
+ err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ if acqErr != nil || held == nil {
+ t.Fatalf("the racing Acquire = %v (MarkCleaning %v)", acqErr, err)
+ }
+ t.Cleanup(func() { _ = held.Release(context.Background()) })
+ if !errors.Is(err, lease.ErrTombstoneChanged) {
+ t.Fatalf("MarkCleaning = %v, want ErrTombstoneChanged", err)
+ }
+ if l, _, _ := s.current(t); l.Owner == lease.CleaningOwner(cleanID) || l.Generation != 5 {
+ t.Fatalf("lock = %+v, want the acquirer's at generation 5", l)
+ }
+ if err := held.Err(); err != nil {
+ t.Fatalf("the acquirer lost its lease: %v", err)
+ }
+}
+
+func TestCleaningMarkIsRefusedWhateverItsAge(t *testing.T) {
+ ctx := context.Background()
+ check := func(t *testing.T, s *tombStore) {
+ t.Helper()
+ _, raw, _ := s.current(t)
+ writes := s.writeCount()
+ wantMark := func(err error) {
+ t.Helper()
+ var he *lease.HeldError
+ if !errors.As(err, &he) || he.Owner != lease.CleaningOwner(cleanID) {
+ t.Fatalf("err = %v, want a *HeldError naming the mark", err)
+ }
+ requireUntouched(t, s, writes, raw)
+ }
+ _, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ wantMark(err)
+ _, err = lease.AcquireTTL(ctx, "data", tombLoc, tombRegion, moves.StagingLeaseTTL)
+ wantMark(err)
+ _, err = lease.ReplicateWithLease(ctx, filepath.Join(t.TempDir(), "x.db"), "data", tombLoc, tombRegion)
+ wantMark(err)
+ wantMark(lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour))
+ lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove))
+ if lifted {
+ t.Fatal("LiftTombstone lifted a mark")
+ }
+ wantMark(err)
+ held, owner, _, err := lease.LockState(ctx, "data", tombLoc, tombRegion)
+ if err != nil || !held || owner != lease.CleaningOwner(cleanID) {
+ t.Fatalf("LockState = %v %q %v, want held by the mark", held, owner, err)
+ }
+ requireUntouched(t, s, writes, raw)
+ }
+ t.Run("the clock advanced 25 h", func(t *testing.T) {
+ s, _ := holdFixture(t)
+ liftedAt(t, s, 4)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(lease.SetClockForTest(func() time.Time { return time.Now().Add(25 * time.Hour) }))
+ check(t, s)
+ })
+ t.Run("its expiry edited by hand to 2000-01-01", func(t *testing.T) {
+ s, _ := holdFixture(t)
+ liftedAt(t, s, 4)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ l, _, _ := s.current(t)
+ l.ExpiresAt = time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)
+ raw, _ := json.Marshal(l)
+ s.Put("data", tombLockKy, raw)
+ check(t, s)
+ })
+}
+
+func TestMarkIsNeverRewritten(t *testing.T) {
+ s, _ := holdFixture(t)
+ ctx := context.Background()
+ liftedAt(t, s, 4)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ _, raw, etag := s.current(t)
+ writes := s.writeCount()
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatalf("a resumed mark = %v", err)
+ }
+ requireUntouched(t, s, writes, raw)
+ if _, _, now := s.current(t); now != etag {
+ t.Fatal("the mark's ETag changed")
+ }
+
+ // In the code: the sentinel expiry is written by MarkCleaning alone, and
+ // the only exported functions that reach a lock write of ours are the
+ // lift, the mark and the unmark. Nothing renews a mark.
+ a := loadLeaseAudit(t)
+ if got := a.refsOutsideDecl("CleaningExpiry"); strings.Join(got, ",") != "MarkCleaning" {
+ t.Errorf("CleaningExpiry is used by %v, want MarkCleaning alone", got)
+ }
+ if got := a.exportedReaching("putLockUntilAnswered"); strings.Join(got, ",") != "LiftTombstone,MarkCleaning,UnmarkCleaning" {
+ t.Errorf("the exported lock writers are %v", got)
+ }
+ renews := regexp.MustCompile(`(?i)(reassert|renew|extend|refresh).*clean|clean.*(reassert|renew|extend|refresh)`)
+ for _, name := range a.funcNames() {
+ if renews.MatchString(name) {
+ t.Errorf("%s: nothing renews a mark", name)
+ }
+ }
+}
+
+func TestUnmarkWritesALiftedTombstoneThatNeverRecurs(t *testing.T) {
+ ctx := context.Background()
+ s, _ := holdFixture(t)
+ before, beforeRaw, beforeETag := liftedAt(t, s, 4)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ l, raw, etag := s.current(t)
+ if !lease.IsLiftedTombstone(l.Owner, l.ExpiresAt) || l.Owner != tombMove || l.Generation != 4 || l.Cleanups != before.Cleanups+1 {
+ t.Fatalf("after the unmark: %+v", l)
+ }
+ if len(l.WriteID) != 32 || l.WriteID == before.WriteID || bytes.Equal(raw, beforeRaw) || etag == beforeETag {
+ t.Fatalf("the unmark's body recurs: %s (before %s)", raw, beforeRaw)
+ }
+ // A second unmark finds no mark and writes anyway.
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ if l, _, _ := s.current(t); l.Cleanups != before.Cleanups+2 || !lease.IsLiftedTombstone(l.Owner, l.ExpiresAt) {
+ t.Fatalf("after a resumed unmark: %+v", l)
+ }
+ second := lease.CleaningOwners{Move: tombMove, Cleaning: lease.CleaningOwner(cleanID2)}
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, second); err != nil {
+ t.Fatal(err)
+ }
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, second); err != nil {
+ t.Fatal(err)
+ }
+ vs, err := s.ListVersions(ctx, "data", tombLockKy)
+ if err != nil {
+ t.Fatal(err)
+ }
+ seen := map[string]bool{}
+ for _, v := range vs {
+ if v.DeleteMarker || v.Key != tombLockKy {
+ continue
+ }
+ if seen[v.ETag] {
+ t.Fatalf("lock.json's ETag %s recurs across its versions", v.ETag)
+ }
+ seen[v.ETag] = true
+ }
+ if len(seen) < 6 {
+ t.Fatalf("%d versions of lock.json, want the live tombstone, the lift and four of ours", len(seen))
+ }
+
+ others := []struct {
+ name string
+ set func(t *testing.T, s *tombStore)
+ }{
+ {"absent", func(t *testing.T, s *tombStore) {}},
+ {"a box's ordinary lease", func(t *testing.T, s *tombStore) {
+ held, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = held.Release(context.Background()) })
+ }},
+ {"another move's tombstone", func(t *testing.T, s *tombStore) {
+ s.putLock(t, tombLoc, tombOther, 3, time.Unix(0, 0))
+ }},
+ {"this move's tombstone still live", func(t *testing.T, s *tombStore) {
+ s.putLock(t, tombLoc, tombMove, 3, time.Now().Add(time.Hour))
+ }},
+ }
+ for _, c := range others {
+ t.Run(c.name, func(t *testing.T) {
+ s, _ := holdFixture(t)
+ c.set(t, s)
+ raw, _ := s.Current("data", tombLockKy)
+ writes := s.writeCount()
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatalf("UnmarkCleaning = %v, want nil", err)
+ }
+ requireUntouched(t, s, writes, raw)
+ })
+ }
+}
+
+// W1's witness (review 9's interleaving): the mark's first attempt is held
+// in the network while its resend marks; the cleanup ends and unmarks; then
+// the first attempt reaches the store. It names the lifted tombstone's ETag,
+// which the unmark's body (a raised count and a fresh write_id) never
+// restores, so it fails, and the lock stays free.
+func TestDelayedMarkCannotLandAfterTheUnmark(t *testing.T) {
+ ctx := context.Background()
+ s, h := holdFixture(t)
+ _, tBody, tETag := liftedAt(t, s, 4)
+ release := hold(t, h, 1)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatalf("MarkCleaning = %v", err)
+ }
+ if l, _, _ := s.current(t); l.Owner != lease.CleaningOwner(cleanID) || l.Cleanups != 0 {
+ t.Fatalf("lock = %+v, want the mark", l)
+ }
+ reqs := h.Requests(tombLockKy)[1:] // [0] is the lift
+ if len(reqs) != 2 {
+ t.Fatalf("%d mark attempts, want 2", len(reqs))
+ }
+ a1, a2 := decodeBody(t, reqs[0].Body), decodeBody(t, reqs[1].Body)
+ if reqs[0].IfMatch != tETag || reqs[1].IfMatch != tETag {
+ t.Fatalf("If-Match %q and %q, want both %q", reqs[0].IfMatch, reqs[1].IfMatch, tETag)
+ }
+ if a1.WriteID == a2.WriteID || a1.WriteID == "" {
+ t.Fatalf("the attempts share write_id %q", a1.WriteID)
+ }
+ a1.WriteID, a2.WriteID = "", ""
+ if a1 != a2 {
+ t.Fatalf("the resend's fields differ: %+v and %+v", a1, a2)
+ }
+
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ l, unmarked, etag := s.current(t)
+ if !lease.IsLiftedTombstone(l.Owner, l.ExpiresAt) || l.Generation != 4 || l.Cleanups != 1 || bytes.Equal(unmarked, tBody) || etag == tETag {
+ t.Fatalf("after the unmark: %+v %s", l, unmarked)
+ }
+
+ release()
+ if st := h.Requests(tombLockKy)[1].Status; st != http.StatusPreconditionFailed {
+ t.Fatalf("the delayed attempt answered %d, want 412", st)
+ }
+ if _, now, _ := s.current(t); !bytes.Equal(now, unmarked) {
+ t.Fatalf("the delayed attempt landed: %s", now)
+ }
+ held, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ if err != nil {
+ t.Fatalf("Acquire after the cleanup = %v", err)
+ }
+ _ = held.Release(ctx)
+}
+
+func TestUnmarkSettlesAnUncertainMark(t *testing.T) {
+ ctx := context.Background()
+ uncertainMark := func(t *testing.T, s *tombStore, h *leasetest.Holder) func() {
+ t.Helper()
+ release := hold(t, h, 1)
+ mctx, cancel := context.WithTimeout(ctx, attempt+500*time.Millisecond)
+ defer cancel()
+ if err := lease.MarkCleaning(mctx, "data", tombLoc, tombRegion, owners); !errors.Is(err, lease.ErrLockWriteUncertain) {
+ t.Fatalf("MarkCleaning = %v, want ErrLockWriteUncertain", err)
+ }
+ return release
+ }
+ t.Run("the held attempt arrives after the unmark", func(t *testing.T) {
+ s, h := holdFixture(t)
+ _, tBody, _ := liftedAt(t, s, 4)
+ release := uncertainMark(t, s, h)
+ if _, now, _ := s.current(t); !bytes.Equal(now, tBody) {
+ t.Fatalf("lock = %s, want the lifted tombstone", now)
+ }
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ l, after, _ := s.current(t)
+ if l.Owner != tombMove || !lease.IsLiftedTombstone(l.Owner, l.ExpiresAt) || l.Cleanups != 1 {
+ t.Fatalf("after the unmark: %+v", l)
+ }
+ release()
+ if st := h.Requests(tombLockKy)[1].Status; st != http.StatusPreconditionFailed {
+ t.Fatalf("the held attempt answered %d, want 412", st)
+ }
+ if _, now, _ := s.current(t); !bytes.Equal(now, after) {
+ t.Fatalf("the held attempt landed: %s", now)
+ }
+ })
+ t.Run("the held attempt lands between the unmark's read and its write", func(t *testing.T) {
+ s, h := holdFixture(t)
+ liftedAt(t, s, 4)
+ release := uncertainMark(t, s, h)
+ s.beforeWrite(func(string) { release() })
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); !errors.Is(err, lease.ErrTombstoneChanged) {
+ t.Fatalf("UnmarkCleaning = %v, want ErrTombstoneChanged", err)
+ }
+ if st := h.Requests(tombLockKy)[1].Status; st != http.StatusOK {
+ t.Fatalf("the held attempt answered %d, want 200", st)
+ }
+ if l, _, _ := s.current(t); l.Owner != lease.CleaningOwner(cleanID) {
+ t.Fatalf("lock = %+v, want the late mark", l)
+ }
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ if l, _, _ := s.current(t); l.Owner != tombMove || !lease.IsLiftedTombstone(l.Owner, l.ExpiresAt) || l.Cleanups != 1 {
+ t.Fatalf("after the second unmark: %+v", l)
+ }
+ })
+}
+
+func TestLockWritesAreSingleAttemptsEachWithItsOwnWriteID(t *testing.T) {
+ ctx := context.Background()
+ t.Run("every send reaches the store once", func(t *testing.T) {
+ s, h := holdFixture(t)
+ liftedAt(t, s, 4)
+ s.FailOn = func(op, key string) error {
+ if op == "put" && key == tombLockKy {
+ return errors.New("injected 500")
+ }
+ return nil
+ }
+ mctx, cancel := context.WithTimeout(ctx, 1500*time.Millisecond)
+ defer cancel()
+ writes := s.writeCount()
+ if err := lease.MarkCleaning(mctx, "data", tombLoc, tombRegion, owners); !errors.Is(err, lease.ErrLockWriteUncertain) {
+ t.Fatalf("MarkCleaning = %v, want ErrLockWriteUncertain", err)
+ }
+ reqs := h.Requests(tombLockKy)[1:]
+ // Sent at 0 s and after 1 s, then the context ends during the 2 s
+ // wait. The SDK's default retryer would have sent the first three
+ // times.
+ if len(reqs) != 2 || s.writeCount()-writes != 2 {
+ t.Fatalf("%d sends (%d writes), want 2", len(reqs), s.writeCount()-writes)
+ }
+ first := decodeBody(t, reqs[0].Body)
+ for i, r := range reqs {
+ if r.Status != http.StatusInternalServerError {
+ t.Fatalf("send %d answered %d", i, r.Status)
+ }
+ b := decodeBody(t, r.Body)
+ if r.IfMatch != reqs[0].IfMatch || !writeIDPattern.MatchString(b.WriteID) {
+ t.Fatalf("send %d: If-Match %q write_id %q", i, r.IfMatch, b.WriteID)
+ }
+ b.WriteID = first.WriteID
+ if b != first {
+ t.Fatalf("send %d's fields differ: %+v", i, b)
+ }
+ for j := 0; j < i; j++ {
+ if bytes.Equal(r.Body, reqs[j].Body) {
+ t.Fatalf("sends %d and %d carry the same body", j, i)
+ }
+ }
+ }
+ })
+ t.Run("a 412 after a timed-out attempt with our mark current", func(t *testing.T) {
+ s, h := holdFixture(t)
+ liftedAt(t, s, 4)
+ var release func()
+ // The hook runs as the first attempt arrives and arms the next: the
+ // resend's arrival lets the held first attempt land before it.
+ s.beforeWrite(func(string) { s.beforeWrite(func(string) { release() }) })
+ release = hold(t, h, 1)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatalf("MarkCleaning = %v, want nil", err)
+ }
+ reqs := h.Requests(tombLockKy)[1:]
+ if len(reqs) != 2 || reqs[0].Status != http.StatusOK || reqs[1].Status != http.StatusPreconditionFailed {
+ t.Fatalf("attempts %+v", reqs)
+ }
+ if l, _, _ := s.current(t); l.Owner != lease.CleaningOwner(cleanID) {
+ t.Fatalf("lock = %+v", l)
+ }
+ })
+}
+
+func TestOlderReadersIgnoreTheNewFields(t *testing.T) {
+ ctx := context.Background()
+ s, _ := holdFixture(t)
+ s.Put("data", tombLockKy, []byte(`{"generation":5,"expires_at":"1970-01-01T00:00:00Z","owner":"`+tombMove+`","cleanups":3,"write_id":"0123456789abcdef0123456789abcdef"}`))
+ held, owner, expires, err := lease.LockState(ctx, "data", tombLoc, tombRegion)
+ if err != nil || held || owner != tombMove || !expires.Equal(time.Unix(0, 0)) {
+ t.Fatalf("LockState = %v %q %s %v", held, owner, expires, err)
+ }
+ if !lease.IsLiftedTombstone(owner, expires) {
+ t.Fatal("not read as a lifted tombstone")
+ }
+ h, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ if err != nil {
+ t.Fatal(err)
+ }
+ raw, _ := s.Current("data", tombLockKy)
+ var m map[string]any
+ if err := json.Unmarshal(raw, &m); err != nil {
+ t.Fatal(err)
+ }
+ if m["generation"] != float64(6) || m["cleanups"] != nil || m["write_id"] != nil {
+ t.Fatalf("Acquire wrote %s", raw)
+ }
+ _ = h.Release(ctx)
+
+ s.Put("data", tombLockKy, []byte(`{"generation":5,"expires_at":"9999-12-31T00:00:00Z","owner":"`+lease.CleaningOwner(cleanID)+`","cleanups":3,"write_id":"0123456789abcdef0123456789abcdef"}`))
+ _, err = lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ wantHeld(t, err, lease.CleaningOwner(cleanID))
+}
+
+func TestLiftTombstoneKeepsLitestreamsFields(t *testing.T) {
+ ctx := context.Background()
+ s, h := holdFixture(t)
+ s.putLock(t, tombLoc, tombMove, 7, time.Now().Add(time.Hour))
+ if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove)); err != nil || !lifted {
+ t.Fatalf("LiftTombstone = %v %v", lifted, err)
+ }
+ _, raw, first := s.current(t)
+ var l litestream.Lease
+ if err := json.Unmarshal(raw, &l); err != nil {
+ t.Fatal(err)
+ }
+ if l.Generation != 7 || !l.ExpiresAt.Equal(time.Unix(0, 0).UTC()) || l.Owner != tombMove {
+ t.Fatalf("lifted = %+v", l)
+ }
+ var m map[string]any
+ _ = json.Unmarshal(raw, &m)
+ if _, ok := m["cleanups"]; ok {
+ t.Fatalf("a lift wrote a cleanups key: %s", raw)
+ }
+ if id, _ := m["write_id"].(string); !writeIDPattern.MatchString(id) {
+ t.Fatalf("write_id %q", m["write_id"])
+ }
+
+ // The same move's tombstone recreated at the same generation and lifted
+ // again: another write_id, so another ETag.
+ s.putLock(t, tombLoc, tombMove, 7, time.Now().Add(time.Hour))
+ if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove)); err != nil || !lifted {
+ t.Fatalf("second LiftTombstone = %v %v", lifted, err)
+ }
+ if _, _, second := s.current(t); second == first {
+ t.Fatal("the second lift's ETag is the first's")
+ }
+
+ // An attempt answered 500 is sent again by the package, with a new
+ // write_id, and never by the SDK.
+ s.putLock(t, tombLoc, tombMove, 7, time.Now().Add(time.Hour))
+ var failed atomic.Bool
+ s.FailOn = func(op, key string) error {
+ if op == "put" && key == tombLockKy && failed.CompareAndSwap(false, true) {
+ return errors.New("injected 500")
+ }
+ return nil
+ }
+ before := len(h.Requests(tombLockKy))
+ if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove)); err != nil || !lifted {
+ t.Fatalf("LiftTombstone through a 500 = %v %v", lifted, err)
+ }
+ reqs := h.Requests(tombLockKy)[before:]
+ if len(reqs) != 2 || reqs[0].Status != http.StatusInternalServerError || reqs[1].Status != http.StatusOK {
+ t.Fatalf("attempts %+v", reqs)
+ }
+ if decodeBody(t, reqs[0].Body).WriteID == decodeBody(t, reqs[1].Body).WriteID {
+ t.Fatal("the resend reused the write_id")
+ }
+}
+
+// X1's witness (review 10's interleaving), every step a production call: a
+// delayed tombstone creation lands on an absent lock long after the move
+// that wrote it finished, and a later landing lifts that stale tombstone.
+// Without a write_id the lift would be T0 byte for byte and a mark attempt
+// still in the network that named T0 would land.
+func TestDelayedTombstoneCreationCannotRestoreAMarkCondition(t *testing.T) {
+ ctx := context.Background()
+ s, h := holdFixture(t)
+ x := lease.TombstonePrefix + "mv_x"
+ o := lease.CleaningOwners{Move: x, Cleaning: lease.CleaningOwner(cleanID)}
+ if !s.absent(t) {
+ t.Fatal("fixture: the lock exists")
+ }
+
+ relS1 := hold(t, h, 1)
+ sctx, cancel := context.WithTimeout(ctx, 300*time.Millisecond)
+ if err := lease.Tombstone(sctx, "data", tombLoc, tombRegion, x, time.Hour); err == nil {
+ t.Fatal("Tombstone with its creation held succeeded")
+ }
+ cancel()
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, x, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ if l, _, _ := s.current(t); l.Owner != x || l.Generation != 1 {
+ t.Fatalf("tombstone = %+v", l)
+ }
+ if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(x)); err != nil || !lifted {
+ t.Fatalf("LiftTombstone = %v %v", lifted, err)
+ }
+ _, _, t0 := s.current(t)
+
+ relM1 := hold(t, h, 1)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, o); err != nil {
+ t.Fatalf("MarkCleaning = %v", err)
+ }
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, o); err != nil {
+ t.Fatal(err)
+ }
+ if l, _, _ := s.current(t); l.Cleanups != 1 {
+ t.Fatalf("after the unmark: %+v", l)
+ }
+ held, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := held.Release(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if _, owner, exp, err := lease.LockState(ctx, "data", tombLoc, tombRegion); err != nil || owner != "" || !exp.IsZero() {
+ t.Fatalf("after the release: %q %s %v", owner, exp, err)
+ }
+
+ relS1()
+ reqs := h.Requests(tombLockKy)
+ if reqs[0].Status != http.StatusOK {
+ t.Fatalf("the delayed creation answered %d", reqs[0].Status)
+ }
+ if l, _, _ := s.current(t); l.Owner != x || l.Generation != 1 || l.WriteID != "" || lease.IsLiftedTombstone(l.Owner, l.ExpiresAt) {
+ t.Fatalf("after the delayed creation: %+v", l)
+ }
+ stale := x
+ if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, func(o string) bool { return o == stale }); err != nil || !lifted {
+ t.Fatalf("the landing's lift = %v %v", lifted, err)
+ }
+ _, _, t1 := s.current(t)
+ if t1 == t0 {
+ t.Fatal("the stale tombstone's lift recurs as T0")
+ }
+
+ relM1()
+ var m1 *leasetest.HeldRequest
+ for i, r := range h.Requests(tombLockKy) {
+ if b := decodeBody(t, r.Body); b.Owner == o.Cleaning && r.IfMatch == t0 {
+ m1 = &h.Requests(tombLockKy)[i]
+ break
+ }
+ }
+ if m1 == nil || m1.Status != http.StatusPreconditionFailed {
+ t.Fatalf("the delayed mark attempt: %+v, want 412", m1)
+ }
+ if _, _, now := s.current(t); now != t1 {
+ t.Fatal("the delayed mark attempt changed the lock")
+ }
+ held, err = lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ if err != nil {
+ t.Fatalf("Acquire = %v", err)
+ }
+ _ = held.Release(ctx)
+}
+
+func TestMarkNormalisesALiftWithoutAWriteID(t *testing.T) {
+ ctx := context.Background()
+ for _, c := range []struct {
+ name string
+ expires time.Time
+ }{
+ {"a lift written before this plan", time.Unix(0, 0)},
+ {"this move's tombstone expired unlifted", time.Now().Add(-time.Hour)},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ s, h := holdFixture(t)
+ s.putLock(t, tombLoc, tombMove, 3, c.expires)
+ _, _, e0 := s.current(t)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ reqs := h.Requests(tombLockKy)
+ if len(reqs) != 2 {
+ t.Fatalf("%d PUTs, want the normalising write and the mark", len(reqs))
+ }
+ n := decodeBody(t, reqs[0].Body)
+ if reqs[0].IfMatch != e0 || n.Owner != tombMove || !n.ExpiresAt.Equal(time.Unix(0, 0)) || n.Generation != 3 || !writeIDPattern.MatchString(n.WriteID) {
+ t.Fatalf("normalising write: If-Match %q %+v", reqs[0].IfMatch, n)
+ }
+ if reqs[1].IfMatch != etagOf(reqs[0].Body) || reqs[1].IfMatch == e0 || decodeBody(t, reqs[1].Body).Owner != lease.CleaningOwner(cleanID) {
+ t.Fatalf("mark: If-Match %q %s", reqs[1].IfMatch, reqs[1].Body)
+ }
+ if l, _, _ := s.current(t); l.Owner != lease.CleaningOwner(cleanID) {
+ t.Fatalf("lock = %+v", l)
+ }
+ })
+ }
+}
+
+func TestEveryLockWriteOfOursCarriesItsOwnWriteID(t *testing.T) {
+ a := loadLeaseAudit(t)
+ // Every PutObject in the package, by its enclosing function. The lock is
+ // written only inside putLockUntilAnswered; the others write objects
+ // their callers name and are never given a lock key by this package (no
+ // function that names the lock calls PutObject itself, checked below).
+ puts := a.callsTo(func(fn *types.Func) bool {
+ return fn.Name() == "PutObject" // the SDK client's, or litestream's S3API's
+ })
+ want := map[string]bool{
+ "putLockUntilAnswered": true, // the lock
+ "versionedS3.PutIfAbsent": true, // VersionedStore: an object its caller names
+ "versionedS3.PutFileIfAbsent": true,
+ "s3Mirror.Put": true, // the mirror's write of a replica object
+ }
+ var got []string
+ for fn := range puts {
+ got = append(got, fn)
+ if !want[fn] {
+ t.Errorf("PutObject in %s, which is not inventoried", fn)
+ }
+ }
+ sort.Strings(got)
+ for fn := range want {
+ if !puts[fn] {
+ t.Errorf("the inventory names %s, which no longer calls PutObject (have %v)", fn, got)
+ }
+ }
+ for fn := range a.namesLock() {
+ if puts[fn] && fn != "putLockUntilAnswered" {
+ t.Errorf("%s names the lock and calls PutObject itself", fn)
+ }
+ }
+ if callers := a.callersOf("putLockUntilAnswered"); strings.Join(callers, ",") != "LiftTombstone,MarkCleaning,UnmarkCleaning" {
+ t.Errorf("putLockUntilAnswered is called by %v", callers)
+ }
+ if !a.writeIDDrawnPerSend() {
+ t.Error("putLockUntilAnswered does not draw WriteID from newWriteID() inside its loop before it marshals")
+ }
+
+ // In behaviour: a lift, a mark with one held attempt and an unmark send
+ // pairwise distinct bodies, each with its own write_id.
+ ctx := context.Background()
+ s, h := holdFixture(t)
+ liftedAt(t, s, 2)
+ release := hold(t, h, 1)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ release()
+ reqs := h.Requests(tombLockKy)
+ if len(reqs) != 4 {
+ t.Fatalf("%d lock writes, want 4", len(reqs))
+ }
+ ids := map[string]bool{}
+ for i, r := range reqs {
+ id := decodeBody(t, r.Body).WriteID
+ if !writeIDPattern.MatchString(id) || ids[id] {
+ t.Fatalf("write %d: write_id %q", i, id)
+ }
+ ids[id] = true
+ for j := 0; j < i; j++ {
+ if bytes.Equal(r.Body, reqs[j].Body) {
+ t.Fatalf("writes %d and %d carry the same body", j, i)
+ }
+ }
+ }
+}
+
+// --- the package audit -------------------------------------------------------
+
+type leaseAudit struct {
+ fset *token.FileSet
+ info *types.Info
+ decls map[string]*ast.FuncDecl // "Name" or "Recv.Name"
+}
+
+func loadLeaseAudit(t *testing.T) *leaseAudit {
+ t.Helper()
+ cfg := &packages.Config{Mode: packages.NeedName | packages.NeedFiles | packages.NeedSyntax | packages.NeedTypes | packages.NeedTypesInfo | packages.NeedImports, Dir: "."}
+ pkgs, err := packages.Load(cfg, ".")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if packages.PrintErrors(pkgs) > 0 || len(pkgs) != 1 {
+ t.Fatal("could not load internal/lease")
+ }
+ p := pkgs[0]
+ a := &leaseAudit{fset: p.Fset, info: p.TypesInfo, decls: map[string]*ast.FuncDecl{}}
+ for _, f := range p.Syntax {
+ if strings.HasSuffix(p.Fset.Position(f.Pos()).Filename, "_test.go") {
+ continue
+ }
+ for _, d := range f.Decls {
+ if fd, ok := d.(*ast.FuncDecl); ok && fd.Body != nil {
+ a.decls[funcKey(fd)] = fd
+ }
+ }
+ }
+ return a
+}
+
+func funcKey(fd *ast.FuncDecl) string {
+ if fd.Recv == nil || len(fd.Recv.List) == 0 {
+ return fd.Name.Name
+ }
+ e := fd.Recv.List[0].Type
+ if s, ok := e.(*ast.StarExpr); ok {
+ e = s.X
+ }
+ if id, ok := e.(*ast.Ident); ok {
+ return id.Name + "." + fd.Name.Name
+ }
+ return "?." + fd.Name.Name
+}
+
+func (a *leaseAudit) funcNames() []string {
+ var out []string
+ for k := range a.decls {
+ out = append(out, k)
+ }
+ sort.Strings(out)
+ return out
+}
+
+// callee is the function a call names, or nil.
+func (a *leaseAudit) callee(call *ast.CallExpr) *types.Func {
+ var id *ast.Ident
+ switch f := call.Fun.(type) {
+ case *ast.Ident:
+ id = f
+ case *ast.SelectorExpr:
+ id = f.Sel
+ default:
+ return nil
+ }
+ fn, _ := a.info.Uses[id].(*types.Func)
+ return fn
+}
+
+func (a *leaseAudit) callsTo(match func(*types.Func) bool) map[string]bool {
+ out := map[string]bool{}
+ for name, fd := range a.decls {
+ ast.Inspect(fd.Body, func(n ast.Node) bool {
+ if call, ok := n.(*ast.CallExpr); ok {
+ if fn := a.callee(call); fn != nil && match(fn) {
+ out[name] = true
+ }
+ }
+ return true
+ })
+ }
+ return out
+}
+
+func (a *leaseAudit) callersOf(name string) []string {
+ var out []string
+ for fn := range a.callsTo(func(f *types.Func) bool {
+ return f.Name() == name && f.Pkg() != nil && f.Pkg().Path() == "github.com/carlosframework/platform/internal/lease"
+ }) {
+ out = append(out, fn)
+ }
+ sort.Strings(out)
+ return out
+}
+
+// exportedReaching lists the exported functions of the package from which a
+// chain of calls inside the package reaches target.
+func (a *leaseAudit) exportedReaching(target string) []string {
+ calls := map[string]map[string]bool{}
+ for name, fd := range a.decls {
+ calls[name] = map[string]bool{}
+ ast.Inspect(fd.Body, func(n ast.Node) bool {
+ if call, ok := n.(*ast.CallExpr); ok {
+ if fn := a.callee(call); fn != nil && fn.Pkg() != nil && fn.Pkg().Path() == "github.com/carlosframework/platform/internal/lease" {
+ calls[name][fn.Name()] = true
+ }
+ }
+ return true
+ })
+ }
+ var reaches func(from string, seen map[string]bool) bool
+ reaches = func(from string, seen map[string]bool) bool {
+ if seen[from] {
+ return false
+ }
+ seen[from] = true
+ for c := range calls[from] {
+ if c == target || reaches(c, seen) {
+ return true
+ }
+ }
+ return false
+ }
+ var out []string
+ for name := range a.decls {
+ if ast.IsExported(name) && reaches(name, map[string]bool{}) {
+ out = append(out, name)
+ }
+ }
+ sort.Strings(out)
+ return out
+}
+
+// refsOutsideDecl lists the functions that use the package-level name.
+func (a *leaseAudit) refsOutsideDecl(name string) []string {
+ var out []string
+ for fn, fd := range a.decls {
+ found := false
+ ast.Inspect(fd.Body, func(n ast.Node) bool {
+ if id, ok := n.(*ast.Ident); ok && id.Name == name {
+ if obj := a.info.Uses[id]; obj != nil && obj.Parent() == obj.Pkg().Scope() {
+ found = true
+ }
+ }
+ return true
+ })
+ if found {
+ out = append(out, fn)
+ }
+ }
+ sort.Strings(out)
+ return out
+}
+
+// namesLock lists the functions that name the lock: they call lockKey or
+// tombstoneLeaser, or hold the literal "lock.json".
+func (a *leaseAudit) namesLock() map[string]bool {
+ out := map[string]bool{}
+ for fn, fd := range a.decls {
+ ast.Inspect(fd.Body, func(n ast.Node) bool {
+ switch x := n.(type) {
+ case *ast.CallExpr:
+ if f := a.callee(x); f != nil && (f.Name() == "lockKey" || f.Name() == "tombstoneLeaser" || f.Name() == "leaserLockKey") {
+ out[fn] = true
+ }
+ case *ast.BasicLit:
+ if x.Kind == token.STRING && strings.Contains(x.Value, "lock.json") {
+ out[fn] = true
+ }
+ }
+ return true
+ })
+ }
+ return out
+}
+
+// writeIDDrawnPerSend checks putLockUntilAnswered's shape: inside its loop,
+// an assignment of newWriteID() to a WriteID field comes before the
+// json.Marshal and the PutObject.
+func (a *leaseAudit) writeIDDrawnPerSend() bool {
+ fd := a.decls["putLockUntilAnswered"]
+ if fd == nil {
+ return false
+ }
+ ok := false
+ ast.Inspect(fd.Body, func(n ast.Node) bool {
+ loop, isLoop := n.(*ast.ForStmt)
+ if !isLoop {
+ return true
+ }
+ var drawn, marshal, put token.Pos
+ ast.Inspect(loop.Body, func(n ast.Node) bool {
+ switch x := n.(type) {
+ case *ast.AssignStmt:
+ for i, l := range x.Lhs {
+ if sel, isSel := l.(*ast.SelectorExpr); isSel && sel.Sel.Name == "WriteID" && i < len(x.Rhs) {
+ if call, isCall := x.Rhs[i].(*ast.CallExpr); isCall {
+ if f := a.callee(call); f != nil && f.Name() == "newWriteID" && drawn == token.NoPos {
+ drawn = x.Pos()
+ }
+ }
+ }
+ }
+ case *ast.CallExpr:
+ if f := a.callee(x); f != nil {
+ if f.Name() == "Marshal" && f.Pkg() != nil && f.Pkg().Path() == "encoding/json" && marshal == token.NoPos {
+ marshal = x.Pos()
+ }
+ if f.Name() == "PutObject" && put == token.NoPos {
+ put = x.Pos()
+ }
+ }
+ }
+ return true
+ })
+ if drawn != token.NoPos && marshal != token.NoPos && put != token.NoPos && drawn < marshal && marshal < put {
+ ok = true
+ }
+ return false
+ })
+ return ok
+}
diff --git a/internal/lease/export_test.go b/internal/lease/export_test.go
index c6c589e69..554219449 100644
--- a/internal/lease/export_test.go
+++ b/internal/lease/export_test.go
@@ -1,9 +1,29 @@
package lease
-import awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+import (
+ "time"
+
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+)
// NewVersionedS3WithClients is the test-only constructor: tests bind
// buckets to clients pointed at leasetest.NewS3Server.
func NewVersionedS3WithClients(clients map[string]*awss3.Client) VersionedStore {
return &versionedS3{clients: clients}
}
+
+// SetLockAttemptTimeoutForTest shortens one lock write attempt's deadline
+// (lockAttemptTimeout) and returns the restore.
+func SetLockAttemptTimeoutForTest(d time.Duration) (restore func()) {
+ old := lockAttemptTimeout
+ lockAttemptTimeout = d
+ return func() { lockAttemptTimeout = old }
+}
+
+// SetClockForTest is the lease's clock hook: what this package reads as now
+// when it decides whether a lock has expired. It returns the restore.
+func SetClockForTest(now func() time.Time) (restore func()) {
+ old := leaseNow
+ leaseNow = now
+ return func() { leaseNow = old }
+}
diff --git a/internal/lease/lease.go b/internal/lease/lease.go
index 3da80af54..05e7644d6 100644
--- a/internal/lease/lease.go
+++ b/internal/lease/lease.go
@@ -186,6 +186,16 @@ func AcquireTTL(ctx context.Context, bucket, prefix, region string, ttl time.Dur
leaser.Path = prefix // lock lives at <prefix>/lock.json, colocated with the replica
leaser.TTL = ttl
+ // A cleanup's mark is refused whatever its expiry says (self-serve moves
+ // plan 4): read first, and refuse a cleaning: owner with nothing
+ // written. The sentinel expiry already makes AcquireLease refuse a mark
+ // on its own read; this read also refuses a mark whose expiry was
+ // edited by hand. It costs one GET per acquisition. Any other lock, or
+ // none, goes on to AcquireLease as before.
+ if err := refuseCleaningMark(ctx, leaser); err != nil {
+ return nil, err
+ }
+
started := time.Now()
lease, err := leaser.AcquireLease(ctx)
if err != nil {
diff --git a/internal/lease/leasetest/hold.go b/internal/lease/leasetest/hold.go
new file mode 100644
index 000000000..b47dcf7b9
--- /dev/null
+++ b/internal/lease/leasetest/hold.go
@@ -0,0 +1,154 @@
+package leasetest
+
+import (
+ "bytes"
+ "io"
+ "net/http"
+ "strings"
+ "sync"
+)
+
+// Holder stands in front of NewS3Server's handler and holds chosen PUTs of a
+// key inside the server, as a request delayed in the network would be: the
+// client's attempt ends first and gets no answer, and the request reaches the
+// store later, when the test releases it. The store then applies its
+// condition against whatever is current at that moment.
+//
+// Keys are object keys, the request path after the bucket (path-style, as
+// NewS3Server speaks it).
+type Holder struct {
+ next http.Handler
+
+ mu sync.Mutex
+ holds []*hold
+ reqs map[string][]*HeldRequest
+}
+
+// HeldRequest is one PUT of a key, in arrival order: its body, its If-Match
+// header (without quotes, as the store compares it) and the status next
+// gave it, 0 while it is held.
+type HeldRequest struct {
+ Body []byte
+ IfMatch string
+ Status int
+}
+
+type hold struct {
+ key string
+ left int
+ waiting []*heldCall
+}
+
+type heldCall struct {
+ proceed chan struct{}
+ done chan struct{}
+}
+
+// NewHolder wraps next (NewS3Server's handler).
+func NewHolder(next http.Handler) *Holder {
+ return &Holder{next: next, reqs: map[string][]*HeldRequest{}}
+}
+
+// HoldPuts holds the next n PUTs of key. release lets the requests this call
+// held through to next in arrival order, and returns once next has answered
+// each; from then on this call holds nothing more. release may be called
+// more than once.
+func (h *Holder) HoldPuts(key string, n int) (release func()) {
+ hd := &hold{key: key, left: n}
+ h.mu.Lock()
+ h.holds = append(h.holds, hd)
+ h.mu.Unlock()
+ var once sync.Once
+ return func() {
+ once.Do(func() {
+ h.mu.Lock()
+ hd.left = 0
+ waiting := hd.waiting
+ hd.waiting = nil
+ for i, x := range h.holds {
+ if x == hd {
+ h.holds = append(h.holds[:i], h.holds[i+1:]...)
+ break
+ }
+ }
+ h.mu.Unlock()
+ for _, c := range waiting {
+ close(c.proceed)
+ <-c.done
+ }
+ })
+ }
+}
+
+// Requests lists every PUT of key in arrival order.
+func (h *Holder) Requests(key string) []HeldRequest {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ out := make([]HeldRequest, 0, len(h.reqs[key]))
+ for _, r := range h.reqs[key] {
+ out = append(out, HeldRequest{Body: append([]byte(nil), r.Body...), IfMatch: r.IfMatch, Status: r.Status})
+ }
+ return out
+}
+
+func (h *Holder) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodPut {
+ h.next.ServeHTTP(w, r)
+ return
+ }
+ _, key, _ := strings.Cut(strings.TrimPrefix(r.URL.Path, "/"), "/")
+ // The body is read in full before anything waits: a held request's
+ // client has gone by the time it is let through.
+ body, err := io.ReadAll(r.Body)
+ if err != nil {
+ http.Error(w, "leasetest: holder could not read the body", http.StatusBadRequest)
+ return
+ }
+ rec := &HeldRequest{Body: body, IfMatch: strings.Trim(r.Header.Get("If-Match"), `"`)}
+ var call *heldCall
+ h.mu.Lock()
+ h.reqs[key] = append(h.reqs[key], rec)
+ for _, hd := range h.holds {
+ if hd.key == key && hd.left > 0 {
+ hd.left--
+ call = &heldCall{proceed: make(chan struct{}), done: make(chan struct{})}
+ hd.waiting = append(hd.waiting, call)
+ break
+ }
+ }
+ h.mu.Unlock()
+ if call != nil {
+ <-call.proceed
+ defer close(call.done)
+ }
+ r2 := r.Clone(r.Context())
+ r2.Body = io.NopCloser(bytes.NewReader(body))
+ r2.ContentLength = int64(len(body))
+ sw := &statusWriter{ResponseWriter: w}
+ h.next.ServeHTTP(sw, r2)
+ if sw.status == 0 {
+ sw.status = http.StatusOK
+ }
+ h.mu.Lock()
+ rec.Status = sw.status
+ h.mu.Unlock()
+}
+
+type statusWriter struct {
+ http.ResponseWriter
+ status int
+}
+
+func (s *statusWriter) WriteHeader(code int) {
+ if s.status == 0 {
+ s.status = code
+ }
+ s.ResponseWriter.WriteHeader(code)
+}
+
+func (s *statusWriter) Write(b []byte) (int, error) {
+ if s.status == 0 {
+ s.status = http.StatusOK
+ }
+ return s.ResponseWriter.Write(b)
+}
diff --git a/internal/lease/lockstate.go b/internal/lease/lockstate.go
index 95db5c0cf..f5b7a6269 100644
--- a/internal/lease/lockstate.go
+++ b/internal/lease/lockstate.go
@@ -75,7 +75,9 @@ func LockState(ctx context.Context, bucket, location, region string) (held bool,
if lock.ExpiresAt.IsZero() {
return false, "", time.Time{}, fmt.Errorf("lease: %s has no expiry", key)
}
- return lockHeld(lock.ExpiresAt, time.Now()), lock.Owner, lock.ExpiresAt, nil
+ // A cleanup's mark is held whatever its expiry says (self-serve moves
+ // plan 4): the cut's stop evidence and the seed finish read it so.
+ return lockHeld(lock.ExpiresAt, leaseNow()) || IsCleaningOwner(lock.Owner), lock.Owner, lock.ExpiresAt, nil
}
// lockAbsent reports an explicit S3 object-absence answer: HTTP 404 with an
diff --git a/internal/lease/tombstone.go b/internal/lease/tombstone.go
index 7f9969a21..ef6e0c626 100644
--- a/internal/lease/tombstone.go
+++ b/internal/lease/tombstone.go
@@ -26,7 +26,6 @@ package lease
// wake is free to the next Tombstone, and the two cannot drift apart.
import (
- "bytes"
"context"
"encoding/json"
"errors"
@@ -84,7 +83,7 @@ var liftedExpiry = time.Unix(0, 0).UTC()
// that is still unexpired. An absent lock is not "lifted" here: callers read
// absence from LockState's zero owner and expiry and treat it the same way.
func IsLiftedTombstone(owner string, expires time.Time) bool {
- return tombstoneOwnerOK(owner) && !expires.IsZero() && leaseFree(expires, time.Now())
+ return tombstoneOwnerOK(owner) && !expires.IsZero() && leaseFree(expires, leaseNow())
}
func tombstoneOwnerOK(owner string) bool {
@@ -114,37 +113,61 @@ func tombstoneLeaser(ctx context.Context, bucket, location, region, owner string
return leaser, key, nil
}
-// readLock reads the lock with the ETag it was read at. Only S3's explicit
-// NoSuchKey is absence (lockAbsent); a garbled body, an oversized one or a
-// lock with no expiry is an error, never "free".
+// readLock reads the lock with the ETag it was read at, as litestream's
+// Lease: readLockBody's answer without the fields only this package writes.
func readLock(ctx context.Context, leaser *lss3.Leaser, key string) (lock *litestream.Lease, absent bool, err error) {
+ b, etag, absent, err := readLockBody(ctx, leaser, key)
+ if err != nil || absent {
+ return nil, absent, err
+ }
+ return &litestream.Lease{Generation: b.Generation, ExpiresAt: b.ExpiresAt, Owner: b.Owner, ETag: etag}, false, nil
+}
+
+// readLockBody reads the lock with the ETag it was read at. Only S3's
+// explicit NoSuchKey is absence (lockAbsent); a garbled body, an oversized
+// one, a lock with no expiry or one read back without an ETag is an error,
+// never "free".
+func readLockBody(ctx context.Context, leaser *lss3.Leaser, key string) (body *lockBody, etag string, absent bool, err error) {
+ l, etag, absent, err := fetchLock(ctx, leaser, key)
+ if err != nil || absent {
+ return nil, "", absent, err
+ }
+ if l.ExpiresAt.IsZero() {
+ return nil, "", false, fmt.Errorf("lease: %s has no expiry", key)
+ }
+ if etag == "" {
+ return nil, "", false, fmt.Errorf("lease: %s came back without an ETag", key)
+ }
+ return l, etag, false, nil
+}
+
+// fetchLock is readLockBody's read and decode, before its checks of the
+// expiry and the ETag: absent only on S3's explicit NoSuchKey, an error for a
+// failed read, an oversized body or one that does not decode.
+func fetchLock(ctx context.Context, leaser *lss3.Leaser, key string) (body *lockBody, etag string, absent bool, err error) {
out, err := leaser.Client().GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(leaser.Bucket), Key: aws.String(key)})
if err != nil {
if lockAbsent(err) {
- return nil, true, nil
+ return nil, "", true, nil
}
- return nil, false, fmt.Errorf("lease: read %s: %w", key, err)
+ return nil, "", false, fmt.Errorf("lease: read %s: %w", key, err)
}
defer out.Body.Close()
- body, err := io.ReadAll(io.LimitReader(out.Body, lockBodyLimit+1))
+ raw, err := io.ReadAll(io.LimitReader(out.Body, lockBodyLimit+1))
if err != nil {
- return nil, false, fmt.Errorf("lease: read %s: %w", key, err)
- }
- if len(body) > lockBodyLimit {
- return nil, false, fmt.Errorf("lease: %s is larger than a lock", key)
+ return nil, "", false, fmt.Errorf("lease: read %s: %w", key, err)
}
- var l litestream.Lease
- if err := json.Unmarshal(body, &l); err != nil {
- return nil, false, fmt.Errorf("lease: decode %s: %w", key, err)
+ if len(raw) > lockBodyLimit {
+ return nil, "", false, fmt.Errorf("lease: %s is larger than a lock", key)
}
- if l.ExpiresAt.IsZero() {
- return nil, false, fmt.Errorf("lease: %s has no expiry", key)
+ var l lockBody
+ if err := json.Unmarshal(raw, &l); err != nil {
+ return nil, "", false, fmt.Errorf("lease: decode %s: %w", key, err)
}
- if out.ETag == nil || *out.ETag == "" {
- return nil, false, fmt.Errorf("lease: %s came back without an ETag", key)
+ if out.ETag != nil {
+ etag = *out.ETag
}
- l.ETag = *out.ETag
- return &l, false, nil
+ return &l, etag, false, nil
}
func heldErr(l *litestream.Lease) error {
@@ -180,7 +203,11 @@ func Tombstone(ctx context.Context, bucket, location, region, owner string, ttl
if err != nil {
return err
}
- if !absent && !leaseFree(cur.ExpiresAt, time.Now()) {
+ if !absent && IsCleaningOwner(cur.Owner) {
+ // A cleanup's mark is never free, whatever its expiry says.
+ return heldErr(cur)
+ }
+ if !absent && !leaseFree(cur.ExpiresAt, leaseNow()) {
if cur.Owner == owner {
return nil
}
@@ -239,7 +266,7 @@ func Reassert(ctx context.Context, bucket, location, region, owner string, ttl t
if absent {
return fmt.Errorf("reassert tombstone: %w", ErrTombstoneLost)
}
- if leaseFree(cur.ExpiresAt, time.Now()) {
+ if leaseFree(cur.ExpiresAt, leaseNow()) {
return fmt.Errorf("reassert tombstone: %w (%s, expired %s)", ErrTombstoneLost, cur.Owner, cur.ExpiresAt.UTC().Format(time.RFC3339))
}
if cur.Owner != owner {
@@ -276,7 +303,7 @@ func LiftTombstone(ctx context.Context, bucket, location, region string, mine fu
if err != nil {
return false, err
}
- cur, absent, err := readLock(ctx, leaser, key)
+ cur, etag, absent, err := readLockBody(ctx, leaser, key)
if err != nil {
return false, err
}
@@ -284,32 +311,26 @@ func LiftTombstone(ctx context.Context, bucket, location, region string, mine fu
return false, nil
}
if !tombstoneOwnerOK(cur.Owner) || mine == nil || !mine(cur.Owner) {
- return false, heldErr(cur)
+ return false, fmt.Errorf("acquire lease: %w", &HeldError{Owner: cur.Owner, ExpiresAt: cur.ExpiresAt})
}
if IsLiftedTombstone(cur.Owner, cur.ExpiresAt) {
return false, nil
}
- body, err := json.Marshal(&litestream.Lease{Generation: cur.Generation, ExpiresAt: liftedExpiry, Owner: cur.Owner})
+ // The lifted body is litestream's three fields as read, the cleanup
+ // count as read, and a write_id of its own per attempt (cleaning.go), so
+ // a lift never rebuilds an earlier lift byte for byte, even of a
+ // tombstone recreated at the same generation.
+ status, err := putLockUntilAnswered(ctx, leaser, key, lockBody{Generation: cur.Generation, ExpiresAt: liftedExpiry, Owner: cur.Owner, Cleanups: cur.Cleanups}, etag)
if err != nil {
return false, fmt.Errorf("lift tombstone: %w", err)
}
- _, err = leaser.Client().PutObject(ctx, &awss3.PutObjectInput{
- Bucket: aws.String(leaser.Bucket),
- Key: aws.String(key),
- Body: bytes.NewReader(body),
- ContentType: aws.String("application/json"),
- IfMatch: aws.String(cur.ETag),
- })
- if err == nil {
+ if status == http.StatusOK {
return true, nil
}
- if code, status := apiCode(err); code == "PreconditionFailed" || status == http.StatusPreconditionFailed || code == "NoSuchKey" || status == http.StatusNotFound {
- // Rewritten, or gone (AWS answers a conditional PUT on a missing key
- // 404, some stores 412). Read it back to tell which.
- if _, gone, rerr := readLock(ctx, leaser, key); rerr == nil && gone {
- return false, nil
- }
- return false, fmt.Errorf("lift tombstone: %w", ErrTombstoneChanged)
+ // Rewritten, or gone (AWS answers a conditional PUT on a missing key
+ // 404, some stores 412). Read it back to tell which.
+ if _, gone, rerr := readLock(ctx, leaser, key); rerr == nil && gone {
+ return false, nil
}
- return false, fmt.Errorf("lift tombstone: %w", err)
+ return false, fmt.Errorf("lift tombstone: %w", ErrTombstoneChanged)
}
diff --git a/internal/lease/tombstone_test.go b/internal/lease/tombstone_test.go
index abdc5e70b..76f0ae29f 100644
--- a/internal/lease/tombstone_test.go
+++ b/internal/lease/tombstone_test.go
@@ -42,10 +42,20 @@ func (s *tombStore) beforeWrite(hook func(method string)) {
}
func tombFixture(t *testing.T) *tombStore {
+ t.Helper()
+ return tombFixtureWrap(t, nil)
+}
+
+// tombFixtureWrap is tombFixture with wrap (when set) between the hook and
+// leasetest's server: a leasetest.Holder, say.
+func tombFixtureWrap(t *testing.T, wrap func(http.Handler) http.Handler) *tombStore {
t.Helper()
bucket := leasetest.NewVersionedBucket()
bucket.AddBucket("data")
inner, _ := leasetest.NewS3Server(bucket)
+ if wrap != nil {
+ inner = wrap(inner)
+ }
s := &tombStore{VersionedBucket: bucket}
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
key := strings.TrimPrefix(r.URL.Path, "/data/")
diff --git a/internal/offlinecleanup/seams_test.go b/internal/offlinecleanup/seams_test.go
index 2f8c9fcb8..4f455680d 100644
--- a/internal/offlinecleanup/seams_test.go
+++ b/internal/offlinecleanup/seams_test.go
@@ -28,12 +28,12 @@ import (
// composite literal key or an assignment, only in the _test.go files of
// the directories named in TestDirs;
// - var: a package variable never assigned, nor its address taken,
-// outside its declaration.
+// outside its declaration, except inside the functions Setters names.
type seam struct {
Kind string // call, field, var
Owner string // the declaring package's directory, relative to the module root
Name string
- Setters []string // call: the functions allowed to reference it
+ Setters []string // call: the functions allowed to reference it; var: to assign it
TestDirs []string // field: the directories whose _test.go files may set it
// OwnTests lets the declaring package's own _test.go files use it.
OwnTests bool
@@ -54,6 +54,10 @@ var e2eSeams = []seam{
Why: "moveCfg's injection of offlinecleanup.RootOnlyChain for the gone evidence's local paths (Task 1); host tests only"},
{Kind: "call", Owner: "internal/moves", Name: "SetTableRetryScheduleForTest", OwnTests: true, MinRefs: 1,
Why: "the reservation table's resend schedule in a unit test (Task 2a, ruling D11); its one setter is moves' own tests (fastResend), and a later task that needs it elsewhere names its setter here"},
+ {Kind: "var", Owner: "internal/lease", Name: "leaseNow", Setters: []string{"internal/lease.SetClockForTest"}, MinRefs: 1,
+ Why: "the lease's clock hook (Task 3): what the lease reads as now when it decides a lock has expired; set only by its export_test.go setter"},
+ {Kind: "var", Owner: "internal/lease", Name: "lockAttemptTimeout", Setters: []string{"internal/lease.SetLockAttemptTimeoutForTest"}, MinRefs: 1,
+ Why: "one lock write attempt's deadline (Task 3); shortened only by its export_test.go setter"},
}
const modulePath = "github.com/carlosframework/platform/"
@@ -131,7 +135,7 @@ func seamPermits(s seam, dir string, r seamRef) bool {
return true
}
switch s.Kind {
- case "call":
+ case "call", "var":
for _, set := range s.Setters {
if dir+"."+r.fn == set {
return true