Skip to content
quarry Artifact [924c664348]
Login

Artifact [924c664348]

Artifact 924c664348c67ea4fea799faa3d25fcb611226af8b381da4e282ba10aab27d66:


package lease

import (
	"errors"
	"testing"
	"time"

	"craftdesign.group/quarry/internal/events"
)

func manager(t *testing.T, ev *events.Server, node string, ttl time.Duration) *Manager {
	t.Helper()
	m, err := New(ev.JS(), node, ttl)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(m.Close)
	return m
}

// Two nodes contend for one store: exactly one may write, and its
// commit-point re-check succeeds while the loser is refused.
func TestMutualExclusion(t *testing.T) {
	ev, err := events.Start(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(ev.Close)
	a := manager(t, ev, "node-a", time.Second)
	b := manager(t, ev, "node-b", time.Second)

	wrote := 0
	if err := a.Guard("repo1", func(Token) error { wrote++; return nil }); err != nil {
		t.Fatalf("first acquirer refused: %v", err)
	}
	if !a.Held("repo1") {
		t.Fatal("acquirer does not report holding")
	}
	if err := b.Guard("repo1", func(Token) error { wrote++; return nil }); err != ErrNotHolder {
		t.Fatalf("second node write: got %v, want ErrNotHolder", err)
	}
	if b.Held("repo1") {
		t.Fatal("non-holder reports holding")
	}
	if wrote != 1 {
		t.Fatalf("%d writes ran, want 1", wrote)
	}
	// A different store is independently acquirable.
	if err := b.Guard("repo2", func(Token) error { return nil }); err != nil {
		t.Fatalf("second store on other node refused: %v", err)
	}
}

// When the holder stops renewing, its lease ages out and the other node
// takes over — but only after the failover grace, never during it.
func TestFailover(t *testing.T) {
	ev, err := events.Start(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(ev.Close)
	ttl := time.Second
	a := manager(t, ev, "node-a", ttl)
	b := manager(t, ev, "node-b", ttl)

	if err := a.Guard("repo", func(Token) error { return nil }); err != nil {
		t.Fatal(err)
	}
	// b has seen a's foreign hold (it will try during exclusion below).
	if err := b.Guard("repo", func(Token) error { return nil }); err != ErrNotHolder {
		t.Fatalf("b write while a holds: %v", err)
	}
	a.Close() // a stops renewing; its entry ages out after ttl

	// Immediately after expiry b can acquire, but the failover grace
	// refuses its first write for one more TTL.
	deadline := time.Now().Add(4 * ttl)
	grace, wrote := false, false
	for time.Now().Before(deadline) {
		err := b.Guard("repo", func(Token) error { return nil })
		if err == nil {
			wrote = true
			break
		}
		if errors.Is(err, ErrNotHolder) && b.Held("repo") {
			grace = true // acquired, but grace refuses the write
		}
		time.Sleep(100 * time.Millisecond)
	}
	if !wrote {
		t.Fatal("failover never completed")
	}
	if !grace {
		t.Fatal("failover grace window was never observed")
	}
}

// With the coordinator gone, writes refuse (ErrUnavailable), never
// proceed uncoordinated.
func TestCoordinatorGoneRefusesWrites(t *testing.T) {
	ev, err := events.Start(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	a := manager(t, ev, "node-a", time.Second)
	if err := a.Guard("repo", func(Token) error { return nil }); err != nil {
		t.Fatal(err)
	}
	ev.Close() // coordinator down
	err = a.Guard("repo", func(Token) error {
		t.Fatal("write ran with no coordinator")
		return nil
	})
	if err == nil {
		t.Fatal("write not refused with coordinator down")
	}
}

// The write closure receives the token that proves it runs under the
// lease; a token made any other way than Guard or Unguarded is refused.
func TestGuardHandsTheWriteAToken(t *testing.T) {
	ev, err := events.Start(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(ev.Close)
	a := manager(t, ev, "node-a", time.Second)
	if err := a.Guard("repo", func(tok Token) error { return tok.Err() }); err != nil {
		t.Fatalf("FAIL: the guard's token is not usable: %v", err)
	}
	if err := (Token{}).Err(); !errors.Is(err, ErrNoToken) {
		t.Fatalf("FAIL: a zero token answered %v, want ErrNoToken", err)
	}
	if err := Unguarded("test: no coordinator in this path").Err(); err != nil {
		t.Fatalf("FAIL: an Unguarded token is not usable: %v", err)
	}
}

// Release gives the key up at once: this node stops holding it and
// another takes it without waiting out the TTL. Releasing a key held by
// another node takes nothing from it.
func TestReleaseFreesTheKeyForAnotherNode(t *testing.T) {
	ev, err := events.Start(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(ev.Close)
	a := manager(t, ev, "node-a", time.Minute)
	b := manager(t, ev, "node-b", time.Minute)
	if err := a.Guard("old", func(Token) error { return nil }); err != nil {
		t.Fatal(err)
	}
	if !a.Holds("old") || b.Holds("old") {
		t.Fatalf("FAIL: holds a=%v b=%v before release", a.Holds("old"), b.Holds("old"))
	}
	if err := b.Release("old"); err != nil {
		t.Fatal(err)
	}
	if !a.Holds("old") {
		t.Errorf("FAIL: another node's release took the lease from its holder")
	}
	if err := a.Release("old"); err != nil {
		t.Fatal(err)
	}
	if a.Holds("old") {
		t.Errorf("FAIL: a released key is still held")
	}
	if err := b.Guard("old", func(Token) error { return nil }); err != nil {
		t.Errorf("FAIL: the next node was refused a released key: %v", err)
	}
	if !t.Failed() {
		t.Log("ok")
	}
}

// Release never cuts a running write short, and the key is given up only
// once nothing runs under it.
func TestReleaseWaitsForARunningWrite(t *testing.T) {
	ev, err := events.Start(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(ev.Close)
	a := manager(t, ev, "node-a", time.Minute)
	b := manager(t, ev, "node-b", time.Minute)
	started, finish := make(chan struct{}), make(chan struct{})
	var ctxErr error
	done := make(chan struct{})
	go func() {
		defer close(done)
		a.Guard("old", func(tok Token) error {
			close(started)
			<-finish
			ctxErr = tok.Err()
			return nil
		})
	}()
	<-started
	if err := a.Release("old"); err != nil {
		t.Fatal(err)
	}
	if a.Holds("old") {
		t.Errorf("FAIL: a released key is still renewed")
	}
	if err := b.Guard("old", func(Token) error { return nil }); err == nil {
		t.Errorf("FAIL: another node took the key under a running write")
	}
	close(finish)
	<-done
	if ctxErr != nil {
		t.Errorf("FAIL: the running write was cut short: %v", ctxErr)
	}
	// A node that never saw the old holder owes no grace, so each try is
	// a new node: the first to find the key gone takes it.
	deadline := time.Now().Add(5 * time.Second)
	for i := 0; ; i++ {
		c := manager(t, ev, "node-c"+string(rune('0'+i%10))+string(rune('a'+i/10)), time.Minute)
		if c.Guard("old", func(Token) error { return nil }) == nil {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("FAIL: the key was never given up")
		}
		time.Sleep(50 * time.Millisecond)
	}
	if !t.Failed() {
		t.Log("ok")
	}
}