amadan

carlos / platform Public

lease: UnmarkCleaning answers Unmarked, Settled or Abstained beside its error and MarkCleaning's refusal names the lock it found; the single-attempt test proves the retryer is off and the audit requires it; one 10 s budget per lock operation, reads included

Fix round 1 of plan 4 Task 3. The "every send reaches the store once"
subtest now runs with a 10 s attempt deadline against an instant 500, so a
retryer left on would resend a body inside the attempt (mutation-checked),
and TestEveryLockWriteOfOursCarriesItsOwnWriteID requires the PutObject's
o.Retryer = aws.NopRetryer{} option. An abstaining unmark is Abstained with
the lock's owner and expiry, never ErrNotOurLock; MarkCleaning refuses with
a *NotOurLockError (errors.Is ErrNotOurLock) carrying the same. The lift,
the mark and the unmark each run within lockOperationBudget, a seam with
its one setter in the audit. The inventory classes legacyMoveRound#1 as
build-state only and writerLease#1 as a lease-gated writer.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 74526ffc0240f863709e23d59496649629a1f721 parent f3265f5
6 files changed, +414 −45
  • internal/host/hibernateprefix_inventory_test.go +2 −2
  • internal/lease/cleaning.go +110 −29
  • internal/lease/cleaning_test.go +287 −14
  • internal/lease/export_test.go +9 −0
  • internal/lease/tombstone.go +4 −0
  • internal/offlinecleanup/seams_test.go +2 −0
diff --git a/internal/host/hibernateprefix_inventory_test.go b/internal/host/hibernateprefix_inventory_test.go
index 6f3b76dca..22e0303ce 100644
--- a/internal/host/hibernateprefix_inventory_test.go
+++ b/internal/host/hibernateprefix_inventory_test.go
@@ -62,14 +62,14 @@ var hibernatePrefixSites = map[string]string{
"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/movescut.go writerLease#1": "lease-gated writer",
"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/movespass.go legacyMoveRound#1": "build-state only",
"host/seedwiring.go deleteReplica#1": "delete only",
"ops/hibernate_verify.go VerifyHibernating#1": "reader",
}
diff --git a/internal/lease/cleaning.go b/internal/lease/cleaning.go
index 2744ff072..975e34a51 100644
--- a/internal/lease/cleaning.go
+++ b/internal/lease/cleaning.go
@@ -92,9 +92,12 @@ var leaseNow = time.Now
// 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
+// lockOperationBudget bounds one lock operation, the lift, the mark or the
+// unmark, its reads and every resend of its writes together: the
+// operation's 10 s, applied even when the caller's context has no
+// deadline. A caller's earlier deadline ends it sooner.
+// SetLockOperationBudgetForTest is its only setter.
+var lockOperationBudget = 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,
@@ -124,11 +127,9 @@ func newWriteID() string {
// 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.
+// offlinecleanup.RetrySchedule, until ctx ends: ErrLockWriteUncertain. Its
+// callers bound ctx by lockOperationBudget.
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
@@ -183,12 +184,17 @@ func putLockUntilAnswered(ctx context.Context, leaser *lss3.Leaser, key string,
// 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.
+// - absent, another owner's (another cleanup's mark included), or this
+// move's tombstone still live: a *NotOurLockError naming the lock as
+// found (errors.Is ErrNotOurLock), nothing written.
+//
+// The whole call, its reads and its writes, runs within lockOperationBudget.
func MarkCleaning(ctx context.Context, bucket, location, region string, o CleaningOwners) error {
if !o.valid() {
return ErrCleaningOwners
}
+ ctx, cancel := context.WithTimeout(ctx, lockOperationBudget)
+ defer cancel()
leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, "", 0)
if err != nil {
return err
@@ -202,7 +208,7 @@ func MarkCleaning(ctx context.Context, bucket, location, region string, o Cleani
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))
+ return fmt.Errorf("mark cleaning: %w", notOurLock(cur, absent))
}
if cur.WriteID != "" {
break
@@ -242,48 +248,123 @@ func MarkCleaning(ctx context.Context, bucket, location, region string, o Cleani
// 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);
+// It answers beside a nil error (an error comes with no answer):
+//
+// - written: Unmarked;
+// - the key absent at the write (a 404, or a lost condition read back as
+// absent): Settled, since no conditional write of ours can land on a
+// missing key;
+// - a lost condition read back as this move's lifted tombstone with a
+// higher count than read (an earlier attempt landed): Settled; anything
+// else is 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
+// - absent, another owner's (another cleanup's mark included), or this
+// move's tombstone still live: Abstained with the lock as found, 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 {
+//
+// The whole call, its reads and its write, runs within lockOperationBudget.
+func UnmarkCleaning(ctx context.Context, bucket, location, region string, o CleaningOwners) (UnmarkAnswer, error) {
if !o.valid() {
- return ErrCleaningOwners
+ return nil, ErrCleaningOwners
}
+ ctx, cancel := context.WithTimeout(ctx, lockOperationBudget)
+ defer cancel()
leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, "", 0)
if err != nil {
- return err
+ return nil, err
}
cur, etag, absent, err := readLockBody(ctx, leaser, key)
if err != nil {
- return err
+ return nil, err
}
if absent || !(cur.Owner == o.Cleaning || (cur.Owner == o.Move && IsLiftedTombstone(cur.Owner, cur.ExpiresAt))) {
- return nil
+ return abstainedAt(cur, absent), 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)
+ return nil, fmt.Errorf("unmark cleaning: %w", err)
}
switch status {
- case http.StatusOK, http.StatusNotFound:
- return nil
+ case http.StatusOK:
+ return Unmarked{}, nil
+ case http.StatusNotFound:
+ return Settled{Reason: "the lock was gone at the write"}, nil
}
now, _, gone, err := readLockBody(ctx, leaser, key)
if err != nil {
- return err
+ return nil, err
}
- if !gone && now.Owner == o.Move && IsLiftedTombstone(now.Owner, now.ExpiresAt) && now.Cleanups > cur.Cleanups {
- return nil // an earlier attempt of ours landed
+ if gone {
+ return Settled{Reason: "the lock was gone at the write"}, nil
}
- return fmt.Errorf("unmark cleaning: %w", ErrTombstoneChanged)
+ if now.Owner == o.Move && IsLiftedTombstone(now.Owner, now.ExpiresAt) && now.Cleanups > cur.Cleanups {
+ return Settled{Reason: "an earlier attempt of the unmark landed"}, nil
+ }
+ return nil, fmt.Errorf("unmark cleaning: %w", ErrTombstoneChanged)
+}
+
+// UnmarkAnswer is what UnmarkCleaning did, beside a nil error: Unmarked,
+// Settled or Abstained. An error comes with no answer.
+type UnmarkAnswer interface{ unmarkAnswer() }
+
+// Unmarked: the unmark's own write landed. The lock is this move's lifted
+// tombstone with the count raised and a write_id no other body has.
+type Unmarked struct{}
+
+// Settled: the lock is past our mark without this call's write landing: an
+// earlier attempt of the unmark landed (a lost condition read back as this
+// move's lifted tombstone with a higher count), or the key was gone at the
+// write (a 404, or a lost condition read back as absent), where no
+// conditional write of ours can ever land.
+type Settled struct{ Reason string }
+
+// Abstained: nothing of ours was there to unmark, and nothing was written.
+// The lock is absent, another owner's (another cleanup's mark included:
+// IsCleaningOwner(Owner)), or this move's tombstone still live. No mark
+// attempt of ours can land there. Retrying will not change it: it is not
+// ErrNotOurLock.
+type Abstained struct {
+ Lock string // the lock as found, for the log
+ Owner string
+ ExpiresAt time.Time
+ Absent bool
+}
+
+func (Unmarked) unmarkAnswer() {}
+func (Settled) unmarkAnswer() {}
+func (Abstained) unmarkAnswer() {}
+
+func abstainedAt(l *lockBody, absent bool) Abstained {
+ a := Abstained{Lock: describeLock(l, absent), Absent: absent || l == nil}
+ if !a.Absent {
+ a.Owner, a.ExpiresAt = l.Owner, l.ExpiresAt
+ }
+ return a
+}
+
+// NotOurLockError is MarkCleaning's refusal with the lock it found: absent,
+// another owner's (another cleanup's mark when IsCleaningOwner(Owner)), or
+// this move's tombstone still live. errors.Is(err, ErrNotOurLock) holds.
+// Nothing was written.
+type NotOurLockError struct {
+ Lock string // the lock as found
+ Owner string
+ ExpiresAt time.Time
+ Absent bool
+}
+
+func (e *NotOurLockError) Error() string {
+ return fmt.Sprintf("lease: the lock is not this move's lifted tombstone (%s); nothing was written", e.Lock)
+}
+
+func (e *NotOurLockError) Is(target error) bool { return target == ErrNotOurLock }
+
+func notOurLock(l *lockBody, absent bool) error {
+ a := abstainedAt(l, absent)
+ return &NotOurLockError{Lock: a.Lock, Owner: a.Owner, ExpiresAt: a.ExpiresAt, Absent: a.Absent}
}
// refuseCleaningMark is AcquireTTL's pre-read of the lock litestream's
diff --git a/internal/lease/cleaning_test.go b/internal/lease/cleaning_test.go
index 7a62d261b..a9c12190e 100644
--- a/internal/lease/cleaning_test.go
+++ b/internal/lease/cleaning_test.go
@@ -7,6 +7,7 @@ import (
"encoding/hex"
"encoding/json"
"errors"
+ "fmt"
"go/ast"
"go/token"
"go/types"
@@ -295,7 +296,7 @@ func TestUnmarkWritesALiftedTombstoneThatNeverRecurs(t *testing.T) {
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 {
+ if _, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
t.Fatal(err)
}
l, raw, etag := s.current(t)
@@ -306,7 +307,7 @@ func TestUnmarkWritesALiftedTombstoneThatNeverRecurs(t *testing.T) {
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 {
+ 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) {
@@ -316,7 +317,7 @@ func TestUnmarkWritesALiftedTombstoneThatNeverRecurs(t *testing.T) {
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 {
+ if _, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, second); err != nil {
t.Fatal(err)
}
vs, err := s.ListVersions(ctx, "data", tombLockKy)
@@ -362,8 +363,12 @@ func TestUnmarkWritesALiftedTombstoneThatNeverRecurs(t *testing.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)
+ ans, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ if err != nil {
+ t.Fatalf("UnmarkCleaning = %v, want an abstention", err)
+ }
+ if _, ok := ans.(lease.Abstained); !ok {
+ t.Fatalf("UnmarkCleaning answered %#v, want Abstained", ans)
}
requireUntouched(t, s, writes, raw)
})
@@ -402,7 +407,7 @@ func TestDelayedMarkCannotLandAfterTheUnmark(t *testing.T) {
t.Fatalf("the resend's fields differ: %+v and %+v", a1, a2)
}
- if err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ if _, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
t.Fatal(err)
}
l, unmarked, etag := s.current(t)
@@ -443,7 +448,7 @@ func TestUnmarkSettlesAnUncertainMark(t *testing.T) {
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 {
+ if _, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
t.Fatal(err)
}
l, after, _ := s.current(t)
@@ -463,7 +468,7 @@ func TestUnmarkSettlesAnUncertainMark(t *testing.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) {
+ 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 {
@@ -472,7 +477,7 @@ func TestUnmarkSettlesAnUncertainMark(t *testing.T) {
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 {
+ 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 {
@@ -486,6 +491,10 @@ func TestLockWritesAreSingleAttemptsEachWithItsOwnWriteID(t *testing.T) {
t.Run("every send reaches the store once", func(t *testing.T) {
s, h := holdFixture(t)
liftedAt(t, s, 4)
+ // A long attempt deadline against an instant 500: the SDK's retry
+ // backoff (up to 1 s, then 2 s) fits inside it, so a retryer left
+ // on would send the first body again within the attempt.
+ t.Cleanup(lease.SetLockAttemptTimeoutForTest(10 * time.Second))
s.FailOn = func(op, key string) error {
if op == "put" && key == tombLockKy {
return errors.New("injected 500")
@@ -499,9 +508,11 @@ func TestLockWritesAreSingleAttemptsEachWithItsOwnWriteID(t *testing.T) {
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.
+ // Our sends go at 0 s and after 1 s, and the context ends during the
+ // 2 s wait. Each body (its own write_id) reaches the store exactly
+ // once: with the SDK's default retryer the first body would arrive
+ // again inside its 10 s attempt, and this count and the distinct
+ // bodies below would both fail.
if len(reqs) != 2 || s.writeCount()-writes != 2 {
t.Fatalf("%d sends (%d writes), want 2", len(reqs), s.writeCount()-writes)
}
@@ -668,7 +679,7 @@ func TestDelayedTombstoneCreationCannotRestoreAMarkCondition(t *testing.T) {
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 {
+ if _, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, o); err != nil {
t.Fatal(err)
}
if l, _, _ := s.current(t); l.Cleanups != 1 {
@@ -796,6 +807,9 @@ func TestEveryLockWriteOfOursCarriesItsOwnWriteID(t *testing.T) {
if !a.writeIDDrawnPerSend() {
t.Error("putLockUntilAnswered does not draw WriteID from newWriteID() inside its loop before it marshals")
}
+ if !a.putHasNopRetryer() {
+ t.Error("putLockUntilAnswered's PutObject does not pass the option o.Retryer = aws.NopRetryer{}: the SDK would resend a body itself")
+ }
// In behaviour: a lift, a mark with one held attempt and an unmark send
// pairwise distinct bodies, each with its own write_id.
@@ -806,7 +820,7 @@ func TestEveryLockWriteOfOursCarriesItsOwnWriteID(t *testing.T) {
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 {
+ if _, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
t.Fatal(err)
}
release()
@@ -1053,3 +1067,262 @@ func (a *leaseAudit) writeIDDrawnPerSend() bool {
})
return ok
}
+
+// putHasNopRetryer checks that putLockUntilAnswered's PutObject call passes
+// an option function that sets the client's Retryer to aws.NopRetryer{}, so
+// one send is one HTTP request.
+func (a *leaseAudit) putHasNopRetryer() bool {
+ fd := a.decls["putLockUntilAnswered"]
+ if fd == nil {
+ return false
+ }
+ ok, puts := false, 0
+ ast.Inspect(fd.Body, func(n ast.Node) bool {
+ call, isCall := n.(*ast.CallExpr)
+ if !isCall {
+ return true
+ }
+ if f := a.callee(call); f == nil || f.Name() != "PutObject" {
+ return true
+ }
+ puts++
+ for _, arg := range call.Args[1:] {
+ lit, isLit := arg.(*ast.FuncLit)
+ if !isLit {
+ continue
+ }
+ ast.Inspect(lit.Body, func(n ast.Node) bool {
+ as, isAssign := n.(*ast.AssignStmt)
+ if !isAssign || len(as.Lhs) != 1 || len(as.Rhs) != 1 {
+ return true
+ }
+ sel, isSel := as.Lhs[0].(*ast.SelectorExpr)
+ if isSel && sel.Sel.Name == "Retryer" && types.TypeString(a.info.TypeOf(as.Rhs[0]), nil) == "github.com/aws/aws-sdk-go-v2/aws.NopRetryer" {
+ ok = true
+ }
+ return true
+ })
+ }
+ return true
+ })
+ return ok && puts == 1
+}
+
+// UnmarkCleaning says what it did beside its error (fix round 1, Important
+// 2): Unmarked when its own write landed, Settled when the lock is already
+// past our mark without that write (a 412 read back as an earlier attempt of
+// ours, or a key gone at the write), and Abstained, with what it found, when
+// there is nothing of ours to unmark. An abstention is never ErrNotOurLock:
+// a release loop that retried it would retry a lock that never changes.
+func TestUnmarkAnswersEachOutcome(t *testing.T) {
+ ctx := context.Background()
+ t.Run("Unmarked: the mark lifted by our write", 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)
+ }
+ ans, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ if _, ok := ans.(lease.Unmarked); err != nil || !ok {
+ t.Fatalf("UnmarkCleaning = %#v, %v; want Unmarked", ans, err)
+ }
+ // The settling write over this move's lifted tombstone is a write
+ // of ours too.
+ ans, err = lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ if _, ok := ans.(lease.Unmarked); err != nil || !ok {
+ t.Fatalf("a resumed UnmarkCleaning = %#v, %v; want Unmarked", ans, err)
+ }
+ })
+ t.Run("Settled: a 412 read back as an earlier attempt of ours", func(t *testing.T) {
+ s, h := holdFixture(t)
+ liftedAt(t, s, 4)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ var release func()
+ s.beforeWrite(func(string) { s.beforeWrite(func(string) { release() }) })
+ release = hold(t, h, 1)
+ ans, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ if _, ok := ans.(lease.Settled); err != nil || !ok {
+ t.Fatalf("UnmarkCleaning = %#v, %v; want Settled", ans, err)
+ }
+ if l, _, _ := s.current(t); l.Owner != tombMove || l.Cleanups != 1 {
+ t.Fatalf("lock = %+v", l)
+ }
+ })
+ t.Run("Settled: the key gone at the write, 404 as AWS answers it", func(t *testing.T) {
+ var gone atomic.Bool
+ s := tombFixtureWrap(t, func(next http.Handler) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method == http.MethodPut && strings.HasSuffix(r.URL.Path, "/lock.json") && gone.Load() {
+ w.Header().Set("Content-Type", "application/xml")
+ w.WriteHeader(http.StatusNotFound)
+ fmt.Fprint(w, "<Error><Code>NoSuchKey</Code><Message>NoSuchKey</Message></Error>")
+ return
+ }
+ next.ServeHTTP(w, r)
+ })
+ })
+ liftedAt(t, s, 4)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); err != nil {
+ t.Fatal(err)
+ }
+ s.beforeWrite(func(string) { s.BareDelete("data", tombLockKy); gone.Store(true) })
+ ans, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ if _, ok := ans.(lease.Settled); err != nil || !ok {
+ t.Fatalf("UnmarkCleaning = %#v, %v; want Settled", ans, err)
+ }
+ })
+ t.Run("Settled: the key gone at the write, 412 as other stores answer it", 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)
+ }
+ s.beforeWrite(func(string) { s.BareDelete("data", tombLockKy) })
+ ans, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ if _, ok := ans.(lease.Settled); err != nil || !ok {
+ t.Fatalf("UnmarkCleaning = %#v, %v; want Settled", ans, err)
+ }
+ })
+ t.Run("Abstained names the lock it found", func(t *testing.T) {
+ other := lease.CleaningOwner(cleanID2)
+ for _, c := range []struct {
+ name, owner string
+ absent bool
+ cleaning bool
+ }{
+ {"absent", "", true, false},
+ {"another cleanup's mark", other, false, true},
+ {"another move's tombstone", tombOther, false, false},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ s, _ := holdFixture(t)
+ if !c.absent {
+ s.putLock(t, tombLoc, c.owner, 3, lease.CleaningExpiry)
+ }
+ ans, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ if err != nil {
+ t.Fatalf("UnmarkCleaning = %v; an abstention is not an error", err)
+ }
+ a, ok := ans.(lease.Abstained)
+ if !ok || a.Absent != c.absent || a.Owner != c.owner || lease.IsCleaningOwner(a.Owner) != c.cleaning || a.Lock == "" {
+ t.Fatalf("UnmarkCleaning answered %#v", ans)
+ }
+ if !c.absent && !a.ExpiresAt.Equal(lease.CleaningExpiry) {
+ t.Fatalf("expiry %s", a.ExpiresAt)
+ }
+ })
+ }
+ })
+ t.Run("an uncertain write is an error and no answer", func(t *testing.T) {
+ s, h := holdFixture(t)
+ liftedAt(t, s, 4)
+ hold(t, h, 1)
+ uctx, cancel := context.WithTimeout(ctx, attempt+500*time.Millisecond)
+ defer cancel()
+ ans, err := lease.UnmarkCleaning(uctx, "data", tombLoc, tombRegion, owners)
+ if !errors.Is(err, lease.ErrLockWriteUncertain) || ans != nil {
+ t.Fatalf("UnmarkCleaning = %#v, %v", ans, err)
+ }
+ })
+}
+
+// MarkCleaning's refusal names what it found as a typed error, so Task 4a
+// can tell another cleanup's mark (IsCleaningOwner) from a lock that is not
+// a cleanup's at all without reading the error text.
+func TestMarkRefusalIsTyped(t *testing.T) {
+ ctx := context.Background()
+ other := lease.CleaningOwner(cleanID2)
+ for _, c := range []struct {
+ name, owner string
+ absent bool
+ }{
+ {"another cleanup's mark", other, false},
+ {"another move's tombstone", tombOther, false},
+ {"absent", "", true},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ s, _ := holdFixture(t)
+ if !c.absent {
+ s.putLock(t, tombLoc, c.owner, 3, lease.CleaningExpiry)
+ }
+ err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners)
+ var nl *lease.NotOurLockError
+ if !errors.Is(err, lease.ErrNotOurLock) || !errors.As(err, &nl) {
+ t.Fatalf("MarkCleaning = %v, want a *NotOurLockError", err)
+ }
+ if nl.Absent != c.absent || nl.Owner != c.owner || nl.Lock == "" || lease.IsCleaningOwner(nl.Owner) != (c.owner == other) {
+ t.Fatalf("refusal %+v", nl)
+ }
+ })
+ }
+}
+
+// slowReads delays every GET of the lock by d while armed, in front of the
+// store: a caller whose context has no deadline must still end.
+func slowReadsFixture(t *testing.T, d time.Duration) (*tombStore, *atomic.Bool) {
+ t.Helper()
+ var armed atomic.Bool
+ s := tombFixtureWrap(t, func(next http.Handler) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/lock.json") && armed.Load() {
+ time.Sleep(d)
+ }
+ next.ServeHTTP(w, r)
+ })
+ })
+ return s, &armed
+}
+
+// The 10 s budget is per operation (fix round 1, minor 3): the lift, the
+// mark and the unmark each bound their reads and their writes together,
+// even under a context with no deadline.
+func TestLockOperationsHaveOneBudget(t *testing.T) {
+ ctx := context.Background()
+ t.Run("the reads are bounded", func(t *testing.T) {
+ t.Cleanup(lease.SetLockOperationBudgetForTest(300 * time.Millisecond))
+ s, armed := slowReadsFixture(t, 2*time.Second)
+ liftedAt(t, s, 4)
+ armed.Store(true)
+ for name, op := range map[string]func() error{
+ "mark": func() error { return lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners) },
+ "unmark": func() error { _, err := lease.UnmarkCleaning(ctx, "data", tombLoc, tombRegion, owners); return err },
+ "lift": func() error {
+ _, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove))
+ return err
+ },
+ } {
+ start := time.Now()
+ err := op()
+ if err == nil || time.Since(start) > 1500*time.Millisecond {
+ t.Errorf("%s under a slow read: %v after %s, want an error within the budget", name, err, time.Since(start))
+ }
+ }
+ })
+ t.Run("one budget for the read and the write", func(t *testing.T) {
+ // The read takes 1 s of a 1.5 s budget; the mark's first attempt is
+ // held and times out at 1.2 s, and its resend would go at 2.2 s,
+ // after the operation's budget. A budget per write (from 1 s to
+ // 2.5 s) would let that resend land.
+ t.Cleanup(lease.SetLockOperationBudgetForTest(1500 * time.Millisecond))
+ t.Cleanup(lease.SetLockAttemptTimeoutForTest(attempt))
+ var h *leasetest.Holder
+ var armed atomic.Bool
+ s := tombFixtureWrap(t, func(next http.Handler) http.Handler {
+ h = leasetest.NewHolder(next)
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/lock.json") && armed.CompareAndSwap(true, false) {
+ time.Sleep(time.Second)
+ }
+ h.ServeHTTP(w, r)
+ })
+ })
+ liftedAt(t, s, 4)
+ armed.Store(true)
+ hold(t, h, 1)
+ if err := lease.MarkCleaning(ctx, "data", tombLoc, tombRegion, owners); !errors.Is(err, lease.ErrLockWriteUncertain) {
+ t.Fatalf("MarkCleaning = %v, want ErrLockWriteUncertain at the operation's budget", err)
+ }
+ })
+}
diff --git a/internal/lease/export_test.go b/internal/lease/export_test.go
index 554219449..7ab8fddee 100644
--- a/internal/lease/export_test.go
+++ b/internal/lease/export_test.go
@@ -27,3 +27,12 @@ func SetClockForTest(now func() time.Time) (restore func()) {
leaseNow = now
return func() { leaseNow = old }
}
+
+// SetLockOperationBudgetForTest shortens the budget of one lock operation
+// (the lift, the mark or the unmark: its reads and writes together) and
+// returns the restore.
+func SetLockOperationBudgetForTest(d time.Duration) (restore func()) {
+ old := lockOperationBudget
+ lockOperationBudget = d
+ return func() { lockOperationBudget = old }
+}
diff --git a/internal/lease/tombstone.go b/internal/lease/tombstone.go
index ef6e0c626..3b130c902 100644
--- a/internal/lease/tombstone.go
+++ b/internal/lease/tombstone.go
@@ -296,9 +296,13 @@ func Reassert(ctx context.Context, bucket, location, region, owner string, ttl t
// - the lock was rewritten since the read: (false, ErrTombstoneChanged),
// and the rewrite stands. One that was removed since is (false, nil).
//
+// The whole call runs within lockOperationBudget (cleaning.go).
+//
// After a lift, LockState reports held=false with the move's owner and a
// past expiry; IsLiftedTombstone recognises that. Never Reassert after it.
func LiftTombstone(ctx context.Context, bucket, location, region string, mine func(owner string) bool) (lifted bool, err error) {
+ ctx, cancel := context.WithTimeout(ctx, lockOperationBudget)
+ defer cancel()
leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, "", 0)
if err != nil {
return false, err
diff --git a/internal/offlinecleanup/seams_test.go b/internal/offlinecleanup/seams_test.go
index 4f455680d..8a6095daf 100644
--- a/internal/offlinecleanup/seams_test.go
+++ b/internal/offlinecleanup/seams_test.go
@@ -58,6 +58,8 @@ var e2eSeams = []seam{
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"},
+ {Kind: "var", Owner: "internal/lease", Name: "lockOperationBudget", Setters: []string{"internal/lease.SetLockOperationBudgetForTest"}, MinRefs: 1,
+ Why: "one lock operation's budget, its reads and writes together (Task 3 fix round 1); shortened only by its export_test.go setter"},
}
const modulePath = "github.com/carlosframework/platform/"