amadan

carlos / platform Public

offline-move: tombstone lease

Tombstone, Reassert and LiftTombstone over the lock.json leaser: create
with IfNoneMatch, replace and delete with IfMatch on the ETag read. A live
holder is never replaced; Reassert and Lift act only on moved:<id>.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude a1ec0532af13f0d615dafe6a3c5762fc617d6de6 parent 8cbf503
2 files changed, +899 −0
  • internal/lease/tombstone.go +241 −0
  • internal/lease/tombstone_test.go +658 −0
diff --git a/internal/lease/tombstone.go b/internal/lease/tombstone.go
new file mode 100644
index 000000000..293ec9ec3
--- /dev/null
+++ b/internal/lease/tombstone.go
@@ -0,0 +1,241 @@
+package lease
+
+// The tombstone lease: the offline operator move's linearization point.
+//
+// A move that has stopped a host on its source box writes a lease at the
+// host's own lock.json under the owner "moved:<id>" and a long expiry. From
+// then on lease.Acquire on that location fails with a *HeldError naming the
+// move, so no box can wake the host and write to the replica behind the
+// move's back. Three operations manage it, each a conditional write over the
+// client AcquireTTL builds (IfNoneMatch "*" on create, IfMatch on replace
+// or delete) and none of them starts a renewer: the move's sealer re-asserts
+// the tombstone itself on every pass.
+//
+// - Tombstone writes it. It never replaces a live holder's lease.
+// - Reassert pushes the expiry out again, only on a lease the same move
+// owns, with the ETag it just read.
+// - LiftTombstone deletes it, only when the owner is a moved:<id> lease the
+// caller claims as its own, with the ETag it just read.
+//
+// "Live" is the reading LockState gives: a lease counts as held until its
+// expiry has passed by more than RenewMargin, so a holder whose clock runs a
+// little behind is never displaced.
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "strings"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ "github.com/benbjohnson/litestream"
+ lss3 "github.com/benbjohnson/litestream/s3"
+)
+
+// TombstonePrefix starts the owner of every tombstone lease. The three
+// functions here refuse any other owner.
+const TombstonePrefix = "moved:"
+
+// MaxTombstoneTTL is the longest a tombstone is ever written for. The moves
+// package's TombstoneTTL equals it; this package cannot import that one.
+const MaxTombstoneTTL = 7 * 24 * time.Hour
+
+var (
+ // ErrTombstoneOwner refuses an owner that is not moved:<id>.
+ ErrTombstoneOwner = errors.New("lease: a tombstone owner is moved:<id>")
+ // ErrTombstoneTTL refuses a TTL that is not positive or exceeds
+ // MaxTombstoneTTL.
+ ErrTombstoneTTL = errors.New("lease: a tombstone lives for at most 7 days")
+ // ErrTombstoneChanged reports that the lease was rewritten or removed
+ // between the read and the conditional write. Nothing was overwritten;
+ // the caller's next pass reads it afresh.
+ ErrTombstoneChanged = errors.New("lease: the lock changed under the tombstone write")
+)
+
+func tombstoneOwnerOK(owner string) bool {
+ return strings.HasPrefix(owner, TombstonePrefix) && len(owner) > len(TombstonePrefix)
+}
+
+func tombstoneTTLOK(ttl time.Duration) bool { return ttl > 0 && ttl <= MaxTombstoneTTL }
+
+// tombstoneLeaser is the leaser a tombstone call writes with, built over the
+// client AcquireTTL builds, and the lock's key. ttl is zero for calls that
+// write no expiry (LiftTombstone).
+func tombstoneLeaser(ctx context.Context, bucket, location, region, owner string, ttl time.Duration) (*lss3.Leaser, string, error) {
+ key, err := lockKey(bucket, location)
+ if err != nil {
+ return nil, "", err
+ }
+ client, _, err := newLeaserClient(ctx, region)
+ if err != nil {
+ return nil, "", err
+ }
+ leaser := lss3.NewLeaser()
+ leaser.SetClient(client)
+ leaser.Bucket = bucket
+ leaser.Path = strings.TrimSuffix(location, "/")
+ leaser.TTL = ttl
+ leaser.Owner = owner
+ 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".
+func readLock(ctx context.Context, leaser *lss3.Leaser, key string) (lock *litestream.Lease, 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, false, fmt.Errorf("lease: read %s: %w", key, err)
+ }
+ defer out.Body.Close()
+ body, 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)
+ }
+ var l litestream.Lease
+ if err := json.Unmarshal(body, &l); err != nil {
+ return nil, false, fmt.Errorf("lease: decode %s: %w", key, err)
+ }
+ if l.ExpiresAt.IsZero() {
+ return nil, false, fmt.Errorf("lease: %s has no expiry", key)
+ }
+ if out.ETag == nil || *out.ETag == "" {
+ return nil, false, fmt.Errorf("lease: %s came back without an ETag", key)
+ }
+ l.ETag = *out.ETag
+ return &l, false, nil
+}
+
+func heldErr(l *litestream.Lease) error {
+ return fmt.Errorf("acquire lease: %w", &HeldError{Owner: l.Owner, ExpiresAt: l.ExpiresAt})
+}
+
+// Tombstone writes the lease at s3://bucket/<location>/lock.json owned by
+// owner (moved:<id>), expiring ttl from now (at most MaxTombstoneTTL).
+//
+// - absent, or expired past RenewMargin: created (IfNoneMatch "*") or
+// replaced (IfMatch on the expired lease's ETag);
+// - held by owner: nil and no write; the caller re-asserts with Reassert;
+// - held by anyone else, live: a *HeldError, nothing written.
+//
+// A live holder is never replaced: the replace is conditional on the ETag of
+// the lease that was read as expired, so a holder that renews in between
+// wins and Tombstone reports it as held.
+func Tombstone(ctx context.Context, bucket, location, region, owner string, ttl time.Duration) error {
+ if !tombstoneOwnerOK(owner) {
+ return ErrTombstoneOwner
+ }
+ if !tombstoneTTLOK(ttl) {
+ return ErrTombstoneTTL
+ }
+ leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, owner, ttl)
+ if err != nil {
+ return err
+ }
+ cur, absent, err := readLock(ctx, leaser, key)
+ if err != nil {
+ return err
+ }
+ if !absent && lockHeld(cur.ExpiresAt, time.Now()) {
+ if cur.Owner == owner {
+ return nil
+ }
+ return heldErr(cur)
+ }
+ // Absent or expired. AcquireLease reads again and writes conditionally:
+ // IfNoneMatch "*" when absent, IfMatch on the ETag it reads otherwise.
+ // If a holder appeared since our read it answers LeaseExistsError.
+ if _, err := leaser.AcquireLease(ctx); err != nil {
+ var lexists *litestream.LeaseExistsError
+ if errors.As(err, &lexists) {
+ if lexists.Owner == owner {
+ return nil
+ }
+ return fmt.Errorf("acquire lease: %w", &HeldError{Owner: lexists.Owner, ExpiresAt: lexists.ExpiresAt})
+ }
+ return fmt.Errorf("tombstone lease: %w", err)
+ }
+ return nil
+}
+
+// Reassert moves the expiry of a tombstone owner already holds to now+ttl:
+//
+// - held by owner: rewritten with IfMatch on the ETag just read. If the
+// lock changed since, ErrTombstoneChanged and nothing is written;
+// - absent, or expired past RenewMargin: Tombstone;
+// - held by anyone else, live: a *HeldError, nothing written.
+func Reassert(ctx context.Context, bucket, location, region, owner string, ttl time.Duration) error {
+ if !tombstoneOwnerOK(owner) {
+ return ErrTombstoneOwner
+ }
+ if !tombstoneTTLOK(ttl) {
+ return ErrTombstoneTTL
+ }
+ leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, owner, ttl)
+ if err != nil {
+ return err
+ }
+ cur, absent, err := readLock(ctx, leaser, key)
+ if err != nil {
+ return err
+ }
+ if absent || !lockHeld(cur.ExpiresAt, time.Now()) {
+ return Tombstone(ctx, bucket, location, region, owner, ttl)
+ }
+ if cur.Owner != owner {
+ return heldErr(cur)
+ }
+ if _, err := leaser.RenewLease(ctx, cur); err != nil {
+ if errors.Is(err, litestream.ErrLeaseNotHeld) {
+ return fmt.Errorf("reassert tombstone: %w", ErrTombstoneChanged)
+ }
+ return fmt.Errorf("reassert tombstone: %w", err)
+ }
+ return nil
+}
+
+// LiftTombstone deletes the tombstone at s3://bucket/<location>/lock.json.
+//
+// - absent: (false, nil);
+// - owner is not a moved:<id> that mine accepts: (false, *HeldError),
+// nothing deleted. This holds for an expired lease too: only the move
+// that wrote a tombstone lifts it;
+// - mine: deleted with IfMatch on the ETag just read, (true, nil). A lock
+// that was rewritten since is (false, ErrTombstoneChanged); one that
+// vanished since is (false, nil).
+func LiftTombstone(ctx context.Context, bucket, location, region string, mine func(owner string) bool) (lifted bool, err error) {
+ leaser, key, err := tombstoneLeaser(ctx, bucket, location, region, "", 0)
+ if err != nil {
+ return false, err
+ }
+ cur, absent, err := readLock(ctx, leaser, key)
+ if err != nil {
+ return false, err
+ }
+ if absent {
+ return false, nil
+ }
+ if !tombstoneOwnerOK(cur.Owner) || mine == nil || !mine(cur.Owner) {
+ return false, heldErr(cur)
+ }
+ switch err := leaser.ReleaseLease(ctx, cur); {
+ case err == nil:
+ return true, nil
+ case errors.Is(err, lss3.ErrLeaseAlreadyReleased):
+ return false, nil
+ case errors.Is(err, litestream.ErrLeaseNotHeld):
+ return false, fmt.Errorf("lift tombstone: %w", ErrTombstoneChanged)
+ default:
+ return false, fmt.Errorf("lift tombstone: %w", err)
+ }
+}
diff --git a/internal/lease/tombstone_test.go b/internal/lease/tombstone_test.go
new file mode 100644
index 000000000..363acaa68
--- /dev/null
+++ b/internal/lease/tombstone_test.go
@@ -0,0 +1,658 @@
+package lease_test
+
+import (
+ "context"
+ "crypto/md5"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+)
+
+const (
+ tombLoc = "instances/h.example.com"
+ tombMove = "moved:mv_01AAAAAAAAAAAAAAAAAAAAAAAA"
+ tombOther = "moved:mv_01BBBBBBBBBBBBBBBBBBBBBBBB"
+ tombRegion = "test-region"
+)
+
+// tombStore is leasetest's S3 server behind a wrapper that does two things the
+// server does not: it enforces If-Match on DELETE (as S3 does; the leaser's
+// release depends on it), and it runs a hook in the window between the
+// code under test's read of lock.json and its conditional write, so a race is
+// a request ordering, not a sleep.
+type tombStore struct {
+ *leasetest.VersionedBucket
+ mu sync.Mutex
+ before func(method string)
+ writes []string // "PUT" and "DELETE" requests that reached lock.json, in order
+}
+
+// beforeWrite arms hook for the next write to lock.json, once.
+func (s *tombStore) beforeWrite(hook func(method string)) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.before = hook
+}
+
+func tombFixture(t *testing.T) *tombStore {
+ t.Helper()
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ inner, _ := leasetest.NewS3Server(bucket)
+ s := &tombStore{VersionedBucket: bucket}
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ key := strings.TrimPrefix(r.URL.Path, "/data/")
+ if strings.HasSuffix(key, "lock.json") && (r.Method == http.MethodPut || r.Method == http.MethodDelete) {
+ s.mu.Lock()
+ hook := s.before
+ s.before = nil
+ s.writes = append(s.writes, r.Method)
+ s.mu.Unlock()
+ if hook != nil {
+ hook(r.Method)
+ }
+ if r.Method == http.MethodDelete {
+ match := strings.Trim(r.Header.Get("If-Match"), `"`)
+ cur, ok := bucket.Current("data", key)
+ switch {
+ case match == "":
+ case !ok:
+ tombXMLErr(w, http.StatusNotFound, "NoSuchKey")
+ return
+ case fmt.Sprintf("%x", md5.Sum(cur)) != match:
+ tombXMLErr(w, http.StatusPreconditionFailed, "PreconditionFailed")
+ return
+ }
+ }
+ }
+ inner.ServeHTTP(w, r)
+ }))
+ t.Cleanup(srv.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", srv.URL)
+ return s
+}
+
+func tombXMLErr(w http.ResponseWriter, status int, code string) {
+ w.Header().Set("Content-Type", "application/xml")
+ w.WriteHeader(status)
+ fmt.Fprintf(w, `<?xml version="1.0" encoding="UTF-8"?><Error><Code>%s</Code><Message>%s</Message></Error>`, code, code)
+}
+
+func (s *tombStore) writeCount() int {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return len(s.writes)
+}
+
+type tombLock struct {
+ Generation int64 `json:"generation"`
+ ExpiresAt time.Time `json:"expires_at"`
+ Owner string `json:"owner"`
+}
+
+func (s *tombStore) putLock(t *testing.T, loc, owner string, gen int64, expires time.Time) {
+ t.Helper()
+ body, err := json.Marshal(tombLock{Generation: gen, ExpiresAt: expires.UTC(), Owner: owner})
+ if err != nil {
+ t.Fatal(err)
+ }
+ s.Put("data", loc+"/lock.json", body)
+}
+
+func (s *tombStore) lock(t *testing.T, loc string) (tombLock, []byte, bool) {
+ t.Helper()
+ body, ok := s.Current("data", loc+"/lock.json")
+ if !ok {
+ return tombLock{}, nil, false
+ }
+ var l tombLock
+ if err := json.Unmarshal(body, &l); err != nil {
+ t.Fatalf("lock.json is not a lock: %v", err)
+ }
+ return l, body, true
+}
+
+func wantHeld(t *testing.T, err error, owner string) {
+ t.Helper()
+ var held *lease.HeldError
+ if !errors.As(err, &held) {
+ t.Fatalf("err = %v, want a *HeldError", err)
+ }
+ if held.Owner != owner {
+ t.Fatalf("held by %q, want %q", held.Owner, owner)
+ }
+}
+
+func near(t *testing.T, got time.Time, want time.Duration) {
+ t.Helper()
+ if d := time.Until(got) - want; d < -time.Minute || d > time.Minute {
+ t.Fatalf("expiry %s is %s from now, want about %s", got, time.Until(got), want)
+ }
+}
+
+func isMine(owner string) func(string) bool {
+ return func(o string) bool { return o == owner }
+}
+
+// A tombstone is a lease like any other: Acquire on the host's location
+// fails with a HeldError that names the move.
+func TestTombstoneBlocksAcquire(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, 2*time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ l, _, ok := s.lock(t, tombLoc)
+ if !ok || l.Owner != tombMove {
+ t.Fatalf("lock = %+v present %v, want owner %s", l, ok, tombMove)
+ }
+ near(t, l.ExpiresAt, 2*time.Hour)
+ _, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ wantHeld(t, err, tombMove)
+ var held *lease.HeldError
+ errors.As(err, &held)
+ near(t, held.ExpiresAt, 2*time.Hour)
+ // The trailing-slash form ReplicaSource hands over reads the same lock.
+ got, owner, _, err := lease.LockState(ctx, "data", tombLoc+"/", tombRegion)
+ if err != nil || !got || owner != tombMove {
+ t.Fatalf("LockState = held %v owner %q err %v", got, owner, err)
+ }
+}
+
+func TestTombstoneNeverReplacesALiveHolder(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ held, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = held.Release(context.Background()) })
+ _, before, _ := s.lock(t, tombLoc)
+ writes := s.writeCount()
+ err = lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour)
+ var he *lease.HeldError
+ if !errors.As(err, &he) || strings.HasPrefix(he.Owner, "moved:") {
+ t.Fatalf("Tombstone over a live holder = %v, want a *HeldError naming the holder", err)
+ }
+ if _, after, _ := s.lock(t, tombLoc); string(after) != string(before) {
+ t.Fatalf("lock.json changed: %s -> %s", before, after)
+ }
+ if s.writeCount() != writes {
+ t.Fatal("Tombstone wrote to a live holder's lock")
+ }
+ if err := held.Err(); err != nil {
+ t.Fatalf("the holder lost its lease: %v", err)
+ }
+ // Reassert is no way round it either.
+ err = lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour)
+ if !errors.As(err, &he) {
+ t.Fatalf("Reassert over a live holder = %v, want a *HeldError", err)
+ }
+ if s.writeCount() != writes {
+ t.Fatal("Reassert wrote to a live holder's lock")
+ }
+}
+
+// A holder whose expiry passed less than RenewMargin ago still counts as
+// holding, as LockState reads it.
+func TestTombstoneRespectsTheRenewMargin(t *testing.T) {
+ s := tombFixture(t)
+ s.putLock(t, tombLoc, "box-a:1", 3, time.Now().Add(-lease.RenewMargin/2))
+ _, before, _ := s.lock(t, tombLoc)
+ err := lease.Tombstone(context.Background(), "data", tombLoc, tombRegion, tombMove, time.Hour)
+ wantHeld(t, err, "box-a:1")
+ if _, after, _ := s.lock(t, tombLoc); string(after) != string(before) {
+ t.Fatal("a lease inside the margin was replaced")
+ }
+}
+
+func TestTombstoneReplacesAnExpiredLease(t *testing.T) {
+ s := tombFixture(t)
+ s.putLock(t, tombLoc, "box-a:1", 3, time.Now().Add(-time.Hour))
+ if err := lease.Tombstone(context.Background(), "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ l, _, _ := s.lock(t, tombLoc)
+ if l.Owner != tombMove || l.Generation != 4 {
+ t.Fatalf("lock = %+v, want owner %s at generation 4", l, tombMove)
+ }
+ near(t, l.ExpiresAt, time.Hour)
+}
+
+func TestTombstoneByTheSameOwnerIsANoOp(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ _, first, _ := s.lock(t, tombLoc)
+ writes := s.writeCount()
+ // A longer ttl on the repeat changes nothing: Reassert is the call that moves it.
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, 48*time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ if _, second, _ := s.lock(t, tombLoc); string(second) != string(first) || s.writeCount() != writes {
+ t.Fatalf("the repeat wrote: %s -> %s", first, second)
+ }
+}
+
+func TestTombstoneRefusesAnotherMove(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ _, before, _ := s.lock(t, tombLoc)
+ wantHeld(t, lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombOther, time.Hour), tombMove)
+ wantHeld(t, lease.Reassert(ctx, "data", tombLoc, tombRegion, tombOther, time.Hour), tombMove)
+ if _, after, _ := s.lock(t, tombLoc); string(after) != string(before) {
+ t.Fatalf("another move's tombstone was overwritten: %s -> %s", before, after)
+ }
+}
+
+func TestTombstoneRefusesWhatItShouldNotWrite(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ for _, owner := range []string{"", "box-a:1", "moved:", "move:mv_1", "Moved:mv_1"} {
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, owner, time.Hour); !errors.Is(err, lease.ErrTombstoneOwner) {
+ t.Fatalf("Tombstone owner %q = %v, want ErrTombstoneOwner", owner, err)
+ }
+ if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, owner, time.Hour); !errors.Is(err, lease.ErrTombstoneOwner) {
+ t.Fatalf("Reassert owner %q = %v, want ErrTombstoneOwner", owner, err)
+ }
+ }
+ for _, ttl := range []time.Duration{0, -time.Hour, lease.MaxTombstoneTTL + time.Second, 30 * 24 * time.Hour} {
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, ttl); !errors.Is(err, lease.ErrTombstoneTTL) {
+ t.Fatalf("Tombstone ttl %s = %v, want ErrTombstoneTTL", ttl, err)
+ }
+ if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, ttl); !errors.Is(err, lease.ErrTombstoneTTL) {
+ t.Fatalf("Reassert ttl %s = %v, want ErrTombstoneTTL", ttl, err)
+ }
+ }
+ for _, loc := range []string{"", "/abs", "../x", "a/../b", `a\b`} {
+ if err := lease.Tombstone(ctx, "data", loc, tombRegion, tombMove, time.Hour); !errors.Is(err, lease.ErrLockLocation) {
+ t.Fatalf("Tombstone location %q = %v, want ErrLockLocation", loc, err)
+ }
+ }
+ if s.writeCount() != 0 {
+ t.Fatal("a refused call wrote")
+ }
+ // Seven days exactly is the longest there is.
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, lease.MaxTombstoneTTL); err != nil {
+ t.Fatal(err)
+ }
+ l, _, _ := s.lock(t, tombLoc)
+ near(t, l.ExpiresAt, lease.MaxTombstoneTTL)
+ if time.Until(l.ExpiresAt) > lease.MaxTombstoneTTL {
+ t.Fatalf("expiry %s is beyond 7 days", l.ExpiresAt)
+ }
+}
+
+func TestReassertMovesTheExpiry(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ first, _, _ := s.lock(t, tombLoc)
+ if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, 48*time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ second, _, _ := s.lock(t, tombLoc)
+ if second.Owner != tombMove || second.Generation != first.Generation {
+ t.Fatalf("lock = %+v, want the same owner and generation as %+v", second, first)
+ }
+ near(t, second.ExpiresAt, 48*time.Hour)
+ // And it can be pulled in again: the expiry is what the caller says.
+ if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ third, _, _ := s.lock(t, tombLoc)
+ near(t, third.ExpiresAt, time.Hour)
+ _, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ wantHeld(t, err, tombMove)
+}
+
+func TestReassertWithNothingToReassertTombstones(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatalf("absent: %v", err)
+ }
+ if l, _, ok := s.lock(t, tombLoc); !ok || l.Owner != tombMove {
+ t.Fatalf("absent: lock = %+v present %v", l, ok)
+ }
+ // An expired tombstone of its own is replaced and live again.
+ s.putLock(t, tombLoc, tombMove, 5, time.Now().Add(-time.Hour))
+ if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatalf("expired: %v", err)
+ }
+ l, _, _ := s.lock(t, tombLoc)
+ if l.Owner != tombMove || l.Generation != 6 {
+ t.Fatalf("expired: lock = %+v, want owner %s at generation 6", l, tombMove)
+ }
+ near(t, l.ExpiresAt, time.Hour)
+ // An expired lease of someone else is replaced the same way.
+ s.putLock(t, tombLoc, tombOther, 2, time.Now().Add(-time.Hour))
+ if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatalf("expired other: %v", err)
+ }
+ if l, _, _ := s.lock(t, tombLoc); l.Owner != tombMove {
+ t.Fatalf("expired other: lock = %+v", l)
+ }
+}
+
+func TestReassertRefusesAHolderThatIsNotTheMove(t *testing.T) {
+ s := tombFixture(t)
+ s.putLock(t, tombLoc, "box-a:1", 3, time.Now().Add(time.Hour))
+ _, before, _ := s.lock(t, tombLoc)
+ wantHeld(t, lease.Reassert(context.Background(), "data", tombLoc, tombRegion, tombMove, time.Hour), "box-a:1")
+ if _, after, _ := s.lock(t, tombLoc); string(after) != string(before) {
+ t.Fatal("Reassert overwrote a live holder")
+ }
+}
+
+// The lock is rewritten between Reassert's read and its write: the write
+// carries the stale ETag, S3 refuses it, and the rewrite stands.
+func TestReassertFailsWhenTheLockChangesUnderIt(t *testing.T) {
+ for name, rewrite := range map[string]func(s *tombStore, t *testing.T){
+ "another move takes it": func(s *tombStore, t *testing.T) {
+ s.putLock(t, tombLoc, tombOther, 9, time.Now().Add(3*time.Hour))
+ },
+ "the same move rewrites it": func(s *tombStore, t *testing.T) {
+ s.putLock(t, tombLoc, tombMove, 9, time.Now().Add(3*time.Hour))
+ },
+ "it is removed": func(s *tombStore, t *testing.T) {
+ s.BareDelete("data", tombLoc+"/lock.json")
+ },
+ } {
+ t.Run(name, func(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ var after []byte
+ var present bool
+ s.beforeWrite(func(method string) {
+ if method != http.MethodPut {
+ t.Errorf("Reassert sent a %s", method)
+ }
+ rewrite(s, t)
+ _, after, present = s.lock(t, tombLoc)
+ })
+ err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, 48*time.Hour)
+ if !errors.Is(err, lease.ErrTombstoneChanged) {
+ t.Fatalf("err = %v, want ErrTombstoneChanged", err)
+ }
+ _, now, ok := s.lock(t, tombLoc)
+ if ok != present || string(now) != string(after) {
+ t.Fatalf("Reassert overwrote the rewrite: %s -> %s", after, now)
+ }
+ })
+ }
+}
+
+// Tombstone's own windows: the lock is absent or expired when it reads, and a
+// live holder is there when it writes.
+func TestTombstoneLosesToAHolderThatArrivesFirst(t *testing.T) {
+ for name, setup := range map[string]func(s *tombStore, t *testing.T){
+ "absent": func(s *tombStore, t *testing.T) {},
+ "expired": func(s *tombStore, t *testing.T) { s.putLock(t, tombLoc, "box-a:1", 3, time.Now().Add(-time.Hour)) },
+ } {
+ t.Run(name, func(t *testing.T) {
+ s := tombFixture(t)
+ setup(s, t)
+ var holder []byte
+ s.beforeWrite(func(string) {
+ s.putLock(t, tombLoc, "box-b:2", 7, time.Now().Add(15*time.Minute))
+ _, holder, _ = s.lock(t, tombLoc)
+ })
+ err := lease.Tombstone(context.Background(), "data", tombLoc, tombRegion, tombMove, time.Hour)
+ wantHeld(t, err, "box-b:2")
+ if _, now, _ := s.lock(t, tombLoc); string(now) != string(holder) {
+ t.Fatalf("Tombstone replaced the holder that arrived first: %s -> %s", holder, now)
+ }
+ })
+ }
+}
+
+func TestLiftTombstone(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ called := ""
+ mine := func(o string) bool { called = o; return o == tombMove }
+
+ // Absent: nothing to lift, and mine is not consulted.
+ if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, mine); lifted || err != nil || called != "" {
+ t.Fatalf("absent: lifted %v err %v consulted %q", lifted, err, called)
+ }
+
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ _, before, _ := s.lock(t, tombLoc)
+
+ // Another move's tombstone is not ours to lift.
+ lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombOther))
+ wantHeld(t, err, tombMove)
+ if lifted {
+ t.Fatal("lifted another move's tombstone")
+ }
+ if _, after, _ := s.lock(t, tombLoc); string(after) != string(before) {
+ t.Fatal("the tombstone changed")
+ }
+ // No ownership test at all lifts nothing.
+ if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, nil); lifted || err == nil {
+ t.Fatalf("nil mine: lifted %v err %v", lifted, err)
+ }
+
+ // Ours lifts, and the host can be acquired again.
+ lifted, err = lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, mine)
+ if err != nil || !lifted || called != tombMove {
+ t.Fatalf("lift: lifted %v err %v consulted %q", lifted, err, called)
+ }
+ if _, _, ok := s.lock(t, tombLoc); ok {
+ t.Fatal("lock.json is still there")
+ }
+ held, err := lease.Acquire(ctx, "data", tombLoc, tombRegion)
+ if err != nil {
+ t.Fatalf("Acquire after the lift: %v", err)
+ }
+ if err := held.Release(ctx); err != nil {
+ t.Fatal(err)
+ }
+
+ // Already lifted: (false, nil), however often it is asked.
+ for i := 0; i < 2; i++ {
+ if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, mine); lifted || err != nil {
+ t.Fatalf("already lifted: lifted %v err %v", lifted, err)
+ }
+ }
+}
+
+// Whatever mine says, only a moved:<id> lease is lifted: a live process's
+// lease is never this function's to delete.
+func TestLiftTombstoneLeavesOrdinaryLeasesAlone(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ for name, expires := range map[string]time.Time{"live": time.Now().Add(time.Hour), "expired": time.Now().Add(-time.Hour)} {
+ s.putLock(t, tombLoc, "box-a:1", 3, expires)
+ _, before, _ := s.lock(t, tombLoc)
+ lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, func(string) bool { return true })
+ wantHeld(t, err, "box-a:1")
+ if lifted {
+ t.Fatalf("%s: lifted an ordinary lease", name)
+ }
+ if _, after, _ := s.lock(t, tombLoc); string(after) != string(before) {
+ t.Fatalf("%s: the lease changed", name)
+ }
+ }
+ if s.writeCount() != 0 {
+ t.Fatal("a refused lift wrote")
+ }
+}
+
+// Our own tombstone is lifted whether or not it has run out: the move that
+// wrote it is the one that says so.
+func TestLiftTombstoneLiftsAnExpiredOwn(t *testing.T) {
+ s := tombFixture(t)
+ s.putLock(t, tombLoc, tombMove, 4, time.Now().Add(-48*time.Hour))
+ lifted, err := lease.LiftTombstone(context.Background(), "data", tombLoc, tombRegion, isMine(tombMove))
+ if err != nil || !lifted {
+ t.Fatalf("lifted %v err %v", lifted, err)
+ }
+ if _, _, ok := s.lock(t, tombLoc); ok {
+ t.Fatal("lock.json is still there")
+ }
+}
+
+func TestLiftTombstoneFailsWhenTheLockChangesUnderIt(t *testing.T) {
+ ctx := context.Background()
+ t.Run("rewritten", func(t *testing.T) {
+ s := tombFixture(t)
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ var after []byte
+ s.beforeWrite(func(method string) {
+ if method != http.MethodDelete {
+ t.Errorf("LiftTombstone sent a %s", method)
+ }
+ // Still ours by owner, but a different lock: Reassert beat the lift.
+ s.putLock(t, tombLoc, tombMove, 8, time.Now().Add(72*time.Hour))
+ _, after, _ = s.lock(t, tombLoc)
+ })
+ lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove))
+ if lifted || !errors.Is(err, lease.ErrTombstoneChanged) {
+ t.Fatalf("lifted %v err %v, want (false, ErrTombstoneChanged)", lifted, err)
+ }
+ if _, now, ok := s.lock(t, tombLoc); !ok || string(now) != string(after) {
+ t.Fatalf("the lift deleted the rewrite: %s", now)
+ }
+ })
+ t.Run("taken over by a holder", func(t *testing.T) {
+ s := tombFixture(t)
+ s.putLock(t, tombLoc, tombMove, 4, time.Now().Add(-time.Hour))
+ var after []byte
+ s.beforeWrite(func(string) {
+ s.putLock(t, tombLoc, "box-b:2", 5, time.Now().Add(15*time.Minute))
+ _, after, _ = s.lock(t, tombLoc)
+ })
+ lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove))
+ if lifted || !errors.Is(err, lease.ErrTombstoneChanged) {
+ t.Fatalf("lifted %v err %v, want (false, ErrTombstoneChanged)", lifted, err)
+ }
+ if _, now, _ := s.lock(t, tombLoc); string(now) != string(after) {
+ t.Fatalf("the lift deleted a live holder's lock: %s", now)
+ }
+ })
+ t.Run("removed", func(t *testing.T) {
+ s := tombFixture(t)
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ s.beforeWrite(func(string) { s.BareDelete("data", tombLoc+"/lock.json") })
+ if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove)); lifted || err != nil {
+ t.Fatalf("lifted %v err %v, want (false, nil): already gone", lifted, err)
+ }
+ })
+}
+
+// The tombstone is a lock.json like any other, and MirrorPrefix, which copies
+// a host's replica to a staging prefix, leaves lock.json alone both ways.
+func TestMirrorPrefixStillSkipsTheTombstone(t *testing.T) {
+ s := tombFixture(t)
+ ctx := context.Background()
+ s.Put("data", "instances/h.example.com/generations/a/snapshot", []byte("snap"))
+ if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ // A staging lease of the target's own, under the destination prefix.
+ s.putLock(t, "moves/h.example.com", "box-z:9", 1, time.Now().Add(time.Hour))
+ _, staged, _ := s.lock(t, "moves/h.example.com")
+
+ m := bucketMirror{s.VersionedBucket}
+ r, err := lease.MirrorPrefix(ctx, m, "data", tombLoc+"/", "data", "moves/h.example.com/", lease.Receipts{})
+ if err != nil || r.Changed != 1 {
+ t.Fatalf("round = %+v, %v; want only the snapshot to move", r, err)
+ }
+ if _, now, _ := s.lock(t, "moves/h.example.com"); string(now) != string(staged) {
+ t.Fatal("the staging lease was touched")
+ }
+ if l, _, ok := s.lock(t, tombLoc); !ok || l.Owner != tombMove {
+ t.Fatalf("the tombstone was touched: %+v", l)
+ }
+ // Reasserting the tombstone is churn the next round does not see.
+ if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, 24*time.Hour); err != nil {
+ t.Fatal(err)
+ }
+ if r, err := lease.MirrorPrefix(ctx, m, "data", tombLoc+"/", "data", "moves/h.example.com/", lease.Receipts{}); err != nil || r.Changed != 0 {
+ t.Fatalf("caught-up round = %+v, %v; want 0 despite the reassert", r, err)
+ }
+}
+
+// bucketMirror is a MirrorStore over the versioned bucket the S3 server also
+// serves, so MirrorPrefix runs over the very lock.json objects the tombstone
+// calls wrote (the S3 server does not speak ListObjectsV2).
+type bucketMirror struct{ b *leasetest.VersionedBucket }
+
+func (m bucketMirror) List(ctx context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) {
+ vs, err := m.b.ListVersions(ctx, bucket, prefix)
+ if err != nil {
+ return nil, err
+ }
+ out := map[string]lease.MirrorObject{}
+ for _, v := range vs {
+ if v.IsLatest && !v.DeleteMarker {
+ out[v.Key] = lease.MirrorObject{ETag: v.ETag, Size: v.Size}
+ }
+ }
+ return out, nil
+}
+
+func (m bucketMirror) Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) {
+ vs, err := m.b.ListVersions(ctx, srcBucket, srcKey)
+ if err != nil {
+ return "", err
+ }
+ for _, v := range vs {
+ if v.Key == srcKey && v.IsLatest && !v.DeleteMarker && v.ETag == srcETag {
+ _, etag, err := m.b.CopyVersion(ctx, srcBucket, srcKey, v.VersionID, dstBucket, dstKey, false)
+ return etag, err
+ }
+ }
+ return "", lease.ErrMirrorNotFound
+}
+
+func (m bucketMirror) Delete(_ context.Context, bucket, key string) error {
+ m.b.BareDelete(bucket, key)
+ return nil
+}
+
+func (m bucketMirror) Get(_ context.Context, bucket, key string) ([]byte, error) {
+ if body, ok := m.b.Current(bucket, key); ok {
+ return body, nil
+ }
+ return nil, lease.ErrMirrorNotFound
+}
+
+func (m bucketMirror) Put(_ context.Context, bucket, key string, body []byte) error {
+ m.b.Put(bucket, key, body)
+ return nil
+}