package lease_test

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"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 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. The conditions
// themselves (If-Match, If-None-Match) are leasetest's own.
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)
			}
		}
		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 (s *tombStore) methods() []string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]string(nil), s.writes...)
}

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")
	}
}

// Free means what Acquire means by it: the expiry has passed. There is no
// RenewMargin grace, so a lease that lapsed a moment ago is taken by Tombstone
// exactly as lease.Acquire would take it. The two are run side by side on
// identical locks so they cannot drift.
func TestTombstoneAndAcquireAgreeOnFree(t *testing.T) {
	ctx := context.Background()
	for name, tc := range map[string]struct {
		expires time.Time
		free    bool
	}{
		"expired a second ago":      {time.Now().Add(-time.Second), true},
		"expired within the margin": {time.Now().Add(-lease.RenewMargin / 2), true},
		"expires in a minute":       {time.Now().Add(time.Minute), false},
	} {
		t.Run(name, func(t *testing.T) {
			s := tombFixture(t)
			s.putLock(t, tombLoc, "box-a:1", 3, tc.expires)
			_, before, _ := s.lock(t, tombLoc)
			acq, acqErr := lease.Acquire(ctx, "data", tombLoc, tombRegion)
			if acqErr == nil {
				_ = acq.Release(ctx)
			}
			// Put the same lock back for Tombstone.
			s.putLock(t, tombLoc, "box-a:1", 3, tc.expires)
			tombErr := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour)
			if (acqErr == nil) != tc.free || (tombErr == nil) != tc.free {
				t.Fatalf("free %v: Acquire err %v, Tombstone err %v", tc.free, acqErr, tombErr)
			}
			if !tc.free {
				wantHeld(t, tombErr, "box-a:1")
				if _, after, _ := s.lock(t, tombLoc); string(after) != string(before) {
					t.Fatal("a live lease was replaced")
				}
			}
		})
	}
}

// An own tombstone that has expired is free like any other: Tombstone writes
// a fresh one rather than reporting the dead one held.
func TestTombstoneRewritesItsOwnExpiredTombstone(t *testing.T) {
	s := tombFixture(t)
	s.putLock(t, tombLoc, tombMove, 4, time.Now().Add(-time.Second))
	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 != 5 {
		t.Fatalf("lock = %+v, want owner %s at generation 5", l, tombMove)
	}
	near(t, l.ExpiresAt, time.Hour)
}

// The write loses its condition and AcquireLease cannot read a winner back:
// there is no holder to name, so the answer is ErrTombstoneChanged, never a
// HeldError with an empty owner.
func TestTombstoneWriteLostThenReadLostIsChanged(t *testing.T) {
	s := tombFixture(t)
	s.putLock(t, tombLoc, "box-a:1", 3, time.Now().Add(-time.Hour))
	s.beforeWrite(func(string) { s.BareDelete("data", tombLoc+"/lock.json") })
	err := lease.Tombstone(context.Background(), "data", tombLoc, tombRegion, tombMove, time.Hour)
	var he *lease.HeldError
	if !errors.Is(err, lease.ErrTombstoneChanged) || errors.As(err, &he) {
		t.Fatalf("err = %v, want ErrTombstoneChanged and no HeldError", err)
	}
	if _, _, ok := s.lock(t, tombLoc); ok {
		t.Fatal("Tombstone wrote over the vanished lock")
	}
}

// The write loses its condition and the read-back shows this very owner
// (another pass of the same move got there first): nothing to do.
func TestTombstoneWriteLostToItsOwnOwnerIsNil(t *testing.T) {
	s := tombFixture(t)
	s.putLock(t, tombLoc, "box-a:1", 3, time.Now().Add(-time.Hour))
	var theirs []byte
	s.beforeWrite(func(string) {
		s.putLock(t, tombLoc, tombMove, 4, time.Now().Add(2*time.Hour))
		_, theirs, _ = s.lock(t, tombLoc)
	})
	if err := lease.Tombstone(context.Background(), "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
		t.Fatalf("err = %v, want nil", err)
	}
	if _, now, _ := s.lock(t, tombLoc); string(now) != string(theirs) {
		t.Fatalf("the winner's lock was overwritten: %s -> %s", theirs, now)
	}
}

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)
}

// With nothing live to keep, Reassert reports the stop lost and writes
// nothing: absent, its own tombstone expired or lifted, and someone else's
// expired lease. It never takes a tombstone again (a wake may have used the
// lease in between).
func TestReassertWithNothingToReassertReportsTheStopLost(t *testing.T) {
	s := tombFixture(t)
	ctx := context.Background()
	if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); !errors.Is(err, lease.ErrTombstoneLost) {
		t.Fatalf("absent: %v, want ErrTombstoneLost", err)
	}
	if _, _, ok := s.lock(t, tombLoc); ok || s.writeCount() != 0 {
		t.Fatalf("absent: a lock was written (%d writes)", s.writeCount())
	}
	for _, c := range []struct {
		name    string
		owner   string
		expires time.Time
	}{
		{"its own, expired", tombMove, time.Now().Add(-time.Hour)},
		{"its own, lifted", tombMove, time.Unix(0, 0)},
		{"someone else's, expired", tombOther, time.Now().Add(-time.Hour)},
		{"an ordinary lease, expired", "box-a:1", time.Now().Add(-time.Second)},
	} {
		s.putLock(t, tombLoc, c.owner, 5, c.expires)
		_, before, _ := s.lock(t, tombLoc)
		writes := s.writeCount()
		if err := lease.Reassert(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); !errors.Is(err, lease.ErrTombstoneLost) {
			t.Fatalf("%s: %v, want ErrTombstoneLost", c.name, err)
		}
		if _, after, _ := s.lock(t, tombLoc); !bytes.Equal(before, after) || s.writeCount() != writes {
			t.Fatalf("%s: the lock was rewritten", c.name)
		}
	}
}

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)
	}
	live, 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: by a conditional PUT, leaving the same owner and generation
	// with a long-past expiry, never by a DELETE.
	writes := s.writeCount()
	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 got := s.methods()[writes:]; len(got) != 1 || got[0] != http.MethodPut {
		t.Fatalf("lift wrote %v, want one PUT", got)
	}
	l, _, ok := s.lock(t, tombLoc)
	if !ok || l.Owner != tombMove || l.Generation != live.Generation || !l.ExpiresAt.Before(time.Now().Add(-lease.RenewMargin)) {
		t.Fatalf("after the lift: %+v present %v, want %s at generation %d, long expired", l, ok, tombMove, live.Generation)
	}
	if held, owner, expires, err := lease.LockState(ctx, "data", tombLoc, tombRegion); err != nil || held || !lease.IsLiftedTombstone(owner, expires) {
		t.Fatalf("LockState after the lift: held %v owner %q expires %v err %v", held, owner, expires, err)
	}
	// The host can be acquired again, and so can a new tombstone.
	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)
	}
}

func TestLiftedTombstoneIsFreeToTheNextMove(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)
	}
	if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove)); err != nil || !lifted {
		t.Fatalf("lifted %v err %v", lifted, err)
	}
	if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombOther, time.Hour); err != nil {
		t.Fatalf("Tombstone over a lifted one: %v", err)
	}
	if l, _, _ := s.lock(t, tombLoc); l.Owner != tombOther {
		t.Fatalf("lock = %+v", l)
	}
}

// Already lifted, however it came about, is (false, nil) with nothing written.
func TestLiftTombstoneAlreadyLifted(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)
	}
	if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove)); err != nil || !lifted {
		t.Fatalf("first lift: %v %v", lifted, err)
	}
	writes := s.writeCount()
	for i := 0; i < 2; i++ {
		if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove)); lifted || err != nil {
			t.Fatalf("already lifted: lifted %v err %v", lifted, err)
		}
	}
	// An own tombstone that simply ran out is lifted too.
	s.putLock(t, tombLoc, tombMove, 4, time.Now().Add(-48*time.Hour))
	writes = s.writeCount()
	if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove)); lifted || err != nil {
		t.Fatalf("expired own: lifted %v err %v", lifted, err)
	}
	if s.writeCount() != writes {
		t.Fatal("lifting an already-lifted tombstone wrote")
	}
	// And so is an absent one.
	s.BareDelete("data", tombLoc+"/lock.json")
	if lifted, err := lease.LiftTombstone(ctx, "data", tombLoc, tombRegion, isMine(tombMove)); lifted || err != nil {
		t.Fatalf("absent: lifted %v err %v", lifted, err)
	}
}

func TestIsLiftedTombstone(t *testing.T) {
	past, future := time.Now().Add(-time.Second), time.Now().Add(time.Hour)
	for name, tc := range map[string]struct {
		owner   string
		expires time.Time
		want    bool
	}{
		"expired move":            {tombMove, past, true},
		"long expired move":       {tombMove, time.Unix(0, 0), true},
		"live move":               {tombMove, future, false},
		"expired ordinary lease":  {"box-a:1", past, false},
		"live ordinary lease":     {"box-a:1", future, false},
		"absent (zero owner/exp)": {"", time.Time{}, false},
		"bare prefix":             {"moved:", past, false},
		"zero expiry":             {tombMove, time.Time{}, false},
	} {
		if got := lease.IsLiftedTombstone(tc.owner, tc.expires); got != tc.want {
			t.Errorf("%s: IsLiftedTombstone(%q, %v) = %v, want %v", name, tc.owner, tc.expires, got, tc.want)
		}
	}
}

// Whatever mine says, only a moved:<id> lease is lifted: a live process's
// lease is never this function's to touch.
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 { // putLock writes the bucket directly; only requests count
		t.Fatal("a refused lift wrote")
	}
}

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.MethodPut {
				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 overwrote the rewrite: %s", now)
		}
	})
	t.Run("taken over by a holder", func(t *testing.T) {
		s := tombFixture(t)
		// Live tombstone read; before the write a holder cannot legitimately
		// appear, but a rewrite by anything must still stand.
		if err := lease.Tombstone(ctx, "data", tombLoc, tombRegion, tombMove, time.Hour); err != nil {
			t.Fatal(err)
		}
		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 overwrote 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)
		}
		if _, _, ok := s.lock(t, tombLoc); ok {
			t.Fatal("the lift re-created a removed lock")
		}
	})
}

// 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
}