package host

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// storesOf is a storeFor over fixed stores, one per bucket name.
func storesOf(m map[string]release.Store) func(context.Context, string) (release.Store, error) {
	return func(_ context.Context, bucket string) (release.Store, error) {
		if s, ok := m[bucket]; ok {
			return s, nil
		}
		return nil, fmt.Errorf("no store for bucket %q", bucket)
	}
}

func testBuildStates(t *testing.T) (*buildStates, release.Store, *registry.Registry) {
	t.Helper()
	reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
	if err != nil {
		t.Fatal(err)
	}
	if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/carlos/a.sock"}); err != nil {
		t.Fatal(err)
	}
	s := release.NewDirStore(t.TempDir())
	return &buildStates{hibBucket: "hib", prefix: func(h string) string { return "inst/" + h },
		storeFor: storesOf(map[string]release.Store{"hib": s}), reg: reg, optDir: t.TempDir(), bakeDir: t.TempDir(), now: time.Now}, s, reg
}

func copyOf(t *testing.T, reg *registry.Registry) release.BuildStateCopy {
	t.Helper()
	rt, _ := reg.Get("a.example.com")
	c, err := release.ParseBuildStateCopy(rt.BuildState)
	if err != nil {
		t.Fatal(err)
	}
	return c
}

func TestRefreshRecordsNoStateAndNoReplica(t *testing.T) {
	b, _, reg := testBuildStates(t)
	if err := b.refresh(context.Background(), "a.example.com"); err != nil {
		t.Fatal(err)
	}
	if c := copyOf(t, reg); !c.Known || c.Replica || c.State != nil || c.ReadErr != "" {
		t.Fatalf("copy = %+v", c)
	}
}

func TestRefreshSeesAReplicaAndAState(t *testing.T) {
	b, s, reg := testBuildStates(t)
	ctx := context.Background()
	if _, err := s.Put(ctx, "inst/a.example.com/generations/0000/x.ltx", []byte("x"), ""); err != nil {
		t.Fatal(err)
	}
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1", Completed: 2}); err != nil {
		t.Fatal(err)
	}
	if err := b.refresh(ctx, "a.example.com"); err != nil {
		t.Fatal(err)
	}
	c := copyOf(t, reg)
	if !c.Known || !c.Replica || c.State == nil || c.State.LastExecuted != "v1" || c.State.Completed != 2 {
		t.Fatalf("copy = %+v", c)
	}
}

// Astra I1 #6: a failed read keeps the last state and is recorded, and the
// next good read clears the record.
func TestAFailedRefreshIsRecorded(t *testing.T) {
	b, s, reg := testBuildStates(t)
	ctx := context.Background()
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
		t.Fatal(err)
	}
	if err := b.refresh(ctx, "a.example.com"); err != nil {
		t.Fatal(err)
	}
	key := "inst/a.example.com/build-state.json"
	if _, err := s.Put(ctx, key, []byte("{"), mustEtag(t, s, key)); err != nil {
		t.Fatal(err)
	}
	if err := b.refresh(ctx, "a.example.com"); err == nil {
		t.Fatal("an unparseable build state refreshed cleanly")
	}
	c := copyOf(t, reg)
	if c.ReadErr == "" || !c.Known || c.State == nil || c.State.LastExecuted != "v1" {
		t.Fatalf("copy = %+v; want the last state kept and the failure recorded", c)
	}
	// update never writes over an object it cannot read: the merge would
	// drop what that object recorded.
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err == nil {
		t.Fatal("an update wrote over an unparseable build state")
	}
	if c := copyOf(t, reg); c.State == nil || c.State.LastExecuted != "v1" {
		t.Fatalf("copy = %+v; a refused update changed it", c)
	}
	// Repaired by hand, the next good read clears the failure.
	if _, err := s.Put(ctx, key, []byte(`{"schema":1,"host":"a.example.com","last_executed":"v2"}`), mustEtag(t, s, key)); err != nil {
		t.Fatal(err)
	}
	if err := b.refresh(ctx, "a.example.com"); err != nil {
		t.Fatal(err)
	}
	if c := copyOf(t, reg); c.ReadErr != "" || c.State == nil || c.State.LastExecuted != "v2" {
		t.Fatalf("copy = %+v; want a good read to clear the failure", c)
	}
}

// A host whose copy was never read stays unknown when its first read fails.
func TestAFailedFirstRefreshStaysUnknown(t *testing.T) {
	b, s, reg := testBuildStates(t)
	ctx := context.Background()
	if _, err := s.Put(ctx, "inst/a.example.com/build-state.json", []byte("{"), ""); err != nil {
		t.Fatal(err)
	}
	if err := b.refresh(ctx, "a.example.com"); err == nil {
		t.Fatal("an unparseable build state refreshed cleanly")
	}
	if c := copyOf(t, reg); c.Known || c.ReadErr == "" || c.State != nil {
		t.Fatalf("copy = %+v; want unknown with the failure recorded", c)
	}
}

func mustEtag(t *testing.T, s release.Store, key string) string {
	t.Helper()
	_, etag, err := s.Get(context.Background(), key)
	if err != nil {
		t.Fatal(err)
	}
	return etag
}

func TestALockAloneIsNotAReplica(t *testing.T) {
	t.Parallel()
	b, s, reg := testBuildStates(t)
	ctx := context.Background()
	if _, err := s.Put(ctx, "inst/a.example.com/lock.json", []byte("{}"), ""); err != nil {
		t.Fatal(err)
	}
	if err := b.refresh(ctx, "a.example.com"); err != nil {
		t.Fatal(err)
	}
	if copyOf(t, reg).Replica {
		t.Fatal("a lease's lock was taken for a replica")
	}
}

// A state naming another host is refused, not adopted.
func TestAStateForAnotherHostIsAReadError(t *testing.T) {
	b, s, _ := testBuildStates(t)
	ctx := context.Background()
	if _, err := s.Put(ctx, "inst/a.example.com/build-state.json", []byte(`{"schema":1,"host":"b.example.com","last_executed":"v1"}`), ""); err != nil {
		t.Fatal(err)
	}
	if _, _, err := b.read(ctx, "a.example.com"); err == nil {
		t.Fatal("a state naming another host was read as this one's")
	}
}

// refreshAll covers the activator-managed exec instances and the standalone
// workers (ruling R-WORKER-ROWS): every row whose spawns run under a lease
// in the replica bucket.
func TestRefreshAllCoversExecInstancesAndWorkers(t *testing.T) {
	b, _, reg := testBuildStates(t)
	for _, rt := range []registry.Route{
		{Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true},
		{Host: "always.example.com", Kind: "sidecar", App: "km"},
		{Host: "unit.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/u.sock", Unit: "km-unit"},
		{Host: "backed.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/b.sock", Backing: "unit"},
		{Host: "backedw.example.com", Kind: "sidecar", App: "km", Hibernate: true, Backing: "unit"},
		{Host: "awake.example.com", Kind: "instance", App: "km", Socket: "/run/w.sock"},
		{Host: "plat.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/p.sock", Platform: true},
		{Host: "svc.example.com", Kind: "service", App: "km", Hibernate: true, Socket: "/run/s.sock"},
	} {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	if skipped := b.refreshAll(context.Background()); skipped != 0 {
		t.Fatalf("skipped %d", skipped)
	}
	for _, h := range []string{"a.example.com", "w.example.com", "always.example.com"} {
		rt, err := reg.Get(h)
		if err != nil || rt == nil {
			t.Fatalf("%s: %v", h, err)
		}
		if c, _ := release.ParseBuildStateCopy(rt.BuildState); !c.Known {
			t.Fatalf("%s was not refreshed: %+v", h, c)
		}
	}
	for _, h := range []string{"unit.example.com", "backed.example.com", "backedw.example.com", "awake.example.com", "plat.example.com", "svc.example.com"} {
		rt, err := reg.Get(h)
		if err != nil || rt == nil {
			t.Fatalf("%s: %v", h, err)
		}
		if rt.BuildState != "" {
			t.Fatalf("%s got a build state copy: %s", h, rt.BuildState)
		}
	}
}

// With host claims on (settings plan 4 Task 5d, carried from Task 4) a
// worker row never leases: the claim runtime refuses sidecar rows before
// any lease (claimwakeassembly.go Prerequisites). refreshAll skips them, so
// no claim is read for them and no copy is written; the claimed instance is
// still refreshed through its claim.
func TestRefreshAllSkipsWorkerRowsWithClaimsOn(t *testing.T) {
	b, hib, reg := testBuildStates(t)
	data := release.NewDirStore(t.TempDir())
	b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": data})
	b.claimBucket = "data"
	var read []string
	b.claims = func(_ context.Context, host string) (hostclaim.Observed, error) {
		read = append(read, host)
		return hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Holder: &hostclaim.Holder{Location: "instances/" + host, Generation: "7", Session: 3}}}, nil
	}
	for _, rt := range []registry.Route{
		{Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true},
		{Host: "always.example.com", Kind: "sidecar", App: "km"},
	} {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
	}
	if skipped := b.refreshAll(context.Background()); skipped != 0 {
		t.Fatalf("skipped %d", skipped)
	}
	if c := copyOf(t, reg); !c.Known || c.ReadErr != "" {
		t.Fatalf("the claimed instance was not refreshed: %+v", c)
	}
	if len(read) != 1 || read[0] != "a.example.com" {
		t.Fatalf("claims read for %v, want the instance alone", read)
	}
	for _, h := range []string{"w.example.com", "always.example.com"} {
		if rt, err := reg.Get(h); err != nil || rt == nil || rt.BuildState != "" {
			t.Fatalf("%s: a worker row got a build state copy with claims on: %+v %v", h, rt, err)
		}
	}
	// With claims off the same rows are refreshed (ruling R-WORKER-ROWS).
	b.claims = nil
	b.refreshAll(context.Background())
	if rt, _ := reg.Get("w.example.com"); rt.BuildState == "" {
		t.Fatal("claims off: the worker row was not refreshed")
	}
}

// Ruling R-BS-REFRESH: refreshAll stops at its budget; the hosts it did not
// reach keep their last copy and are counted.
func TestRefreshAllStopsAtItsBudget(t *testing.T) {
	b, _, reg := testBuildStates(t)
	ctx := context.Background()
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
		t.Fatal(err)
	}
	before := copyOf(t, reg)
	if err := reg.Upsert(registry.Route{Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true}); err != nil {
		t.Fatal(err)
	}
	b.budget = time.Nanosecond
	if skipped := b.refreshAll(ctx); skipped != 2 {
		t.Fatalf("skipped %d, want 2", skipped)
	}
	if after := copyOf(t, reg); after.Column() != before.Column() {
		t.Fatalf("a host not reached changed its copy: %+v -> %+v", before, after)
	}
	if rt, _ := reg.Get("w.example.com"); rt.BuildState != "" {
		t.Fatalf("a host not reached got a copy: %s", rt.BuildState)
	}
}

// listCounter records how the replica was listed; hidePages hides the
// bounded listing so the fallback is exercised.
type listCounter struct {
	release.Store
	pages []int
	full  int
}

func (l *listCounter) List(ctx context.Context, prefix string) ([]string, error) {
	l.full++
	return l.Store.List(ctx, prefix)
}

type pagedCounter struct{ *listCounter }

func (p pagedCounter) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
	p.pages = append(p.pages, limit)
	return p.Store.(release.PageLister).ListPage(ctx, prefix, startAfter, limit)
}

// Ruling R-BS-REFRESH: three keys decide the replica, so a store with a
// bounded listing is asked for three, never for the whole prefix; one
// without it still answers through List.
func TestTheReplicaIsDecidedByABoundedListing(t *testing.T) {
	ctx := context.Background()
	dir := release.NewDirStore(t.TempDir())
	for _, k := range []string{"inst/a/build-state.json", "inst/a/lock.json", "inst/a/generations/0000/a.ltx", "inst/a/generations/0000/b.ltx", "inst/a/generations/0001/c.ltx"} {
		if _, err := dir.Put(ctx, k, []byte("x"), ""); err != nil {
			t.Fatal(err)
		}
	}
	lc := &listCounter{Store: dir}
	paged := pagedCounter{lc}
	if ok, err := hasReplica(ctx, paged, "inst/a"); err != nil || !ok {
		t.Fatalf("paged: %v, %v", ok, err)
	}
	if lc.full != 0 || len(lc.pages) != 1 || lc.pages[0] != 3 {
		t.Fatalf("the whole prefix was listed %d times; pages asked %v, want [3]", lc.full, lc.pages)
	}
	if ok, err := hasReplica(ctx, lc, "inst/a"); err != nil || !ok || lc.full != 1 {
		t.Fatalf("fallback: %v, %v, %d lists", ok, err, lc.full)
	}
	lockOnly := release.NewDirStore(t.TempDir())
	for _, k := range []string{"inst/a/build-state.json", "inst/a/lock.json"} {
		if _, err := lockOnly.Put(ctx, k, []byte("x"), ""); err != nil {
			t.Fatal(err)
		}
	}
	if ok, err := hasReplica(ctx, lockOnly, "inst/a"); err != nil || ok {
		t.Fatalf("lock and state alone: %v, %v", ok, err)
	}
}

// --- where the object lives (ruling R-CLAIMPREFIX) ---

// Host claims off: the hibernate bucket, at the hibernate prefix.
func TestClaimsOffUsesTheHibernatePrefix(t *testing.T) {
	b, s, _ := testBuildStates(t)
	ctx := context.Background()
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.Get(ctx, "inst/a.example.com/build-state.json"); err != nil {
		t.Fatalf("the build state is not beside the hibernate lease: %v", err)
	}
}

// A location on the context (the wake's own lease) wins over both modes.
func TestAContextLocationWins(t *testing.T) {
	b, hib, reg := testBuildStates(t)
	other := release.NewDirStore(t.TempDir())
	b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": other})
	b.claimBucket = "data"
	b.claims = func(context.Context, string) (hostclaim.Observed, error) {
		return hostclaim.Observed{}, errors.New("the claim must not be read when the wake names its lease")
	}
	ctx := activator.WithBuildStateLocation(context.Background(), activator.BuildStateLocation{Bucket: "data", Prefix: "instances/a.example.com/inc1"})
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := other.Get(ctx, "instances/a.example.com/inc1/build-state.json"); err != nil {
		t.Fatalf("the build state is not at the context's location: %v", err)
	}
	if _, _, err := hib.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the hibernate prefix was written too: %v", err)
	}
	if err := b.refresh(ctx, "a.example.com"); err != nil {
		t.Fatal(err)
	}
	if c := copyOf(t, reg); c.State == nil || c.State.LastExecuted != "v1" {
		t.Fatalf("copy = %+v", c)
	}
}

func claimedAt(location string) func(context.Context, string) (hostclaim.Observed, error) {
	return func(_ context.Context, host string) (hostclaim.Observed, error) {
		return hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Holder: &hostclaim.Holder{Location: location, Generation: "7", Session: 3}}}, nil
	}
}

// Host claims on: the claim actor's bucket, at the holder's Location, never
// under the session's g-<gen>/ prefix; the generation's objects are the
// replica.
func TestClaimsOnUsesTheHolderLocationInTheActorBucket(t *testing.T) {
	b, hib, reg := testBuildStates(t)
	data := release.NewDirStore(t.TempDir())
	b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": data})
	b.claimBucket = "data"
	b.claims = claimedAt("instances/a.example.com/inc1")
	ctx := context.Background()
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := data.Get(ctx, "instances/a.example.com/inc1/build-state.json"); err != nil {
		t.Fatalf("the build state is not beside the claimed lease: %v", err)
	}
	if _, _, err := hib.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a claimed host's build state went to the hibernate prefix: %v", err)
	}
	if err := b.refresh(ctx, "a.example.com"); err != nil {
		t.Fatal(err)
	}
	if c := copyOf(t, reg); !c.Known || c.Replica || c.State == nil || c.State.LastExecuted != "v1" {
		t.Fatalf("copy = %+v", c)
	}
	if _, err := data.Put(ctx, "instances/a.example.com/inc1/g-7/0000.ltx", []byte("x"), ""); err != nil {
		t.Fatal(err)
	}
	if err := b.refresh(ctx, "a.example.com"); err != nil {
		t.Fatal(err)
	}
	if c := copyOf(t, reg); !c.Replica {
		t.Fatalf("copy = %+v; want the generation's objects seen as the replica", c)
	}
}

// A claimed host whose claim cannot be read (or has no holder) records the
// failure and keeps its last state; it never falls back to the hibernate
// prefix, even when an object sits there.
func TestAnUnreadableClaimIsAReadError(t *testing.T) {
	t.Parallel()
	for name, read := range map[string]func(context.Context, string) (hostclaim.Observed, error){
		"unreadable": func(context.Context, string) (hostclaim.Observed, error) {
			return hostclaim.Observed{}, errors.New("bucket blip")
		},
		"absent": func(_ context.Context, h string) (hostclaim.Observed, error) {
			return hostclaim.Observed{Host: h, Absent: true}, nil
		},
		"no holder": func(_ context.Context, h string) (hostclaim.Observed, error) {
			return hostclaim.Observed{Host: h, Claim: &hostclaim.Claim{Host: h}}, nil
		},
	} {
		t.Run(name, func(t *testing.T) {
			b, hib, reg := testBuildStates(t)
			data := release.NewDirStore(t.TempDir())
			b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": data})
			b.claimBucket = "data"
			b.claims = claimedAt("instances/a.example.com")
			ctx := context.Background()
			if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
				t.Fatal(err)
			}
			if err := b.refresh(ctx, "a.example.com"); err != nil {
				t.Fatal(err)
			}
			if _, err := hib.Put(ctx, "inst/a.example.com/build-state.json", []byte(`{"schema":1,"host":"a.example.com","last_executed":"hib"}`), ""); err != nil {
				t.Fatal(err)
			}
			b.claims = read
			if err := b.refresh(ctx, "a.example.com"); err == nil {
				t.Fatal("a claimed host refreshed without its claim")
			}
			c := copyOf(t, reg)
			if c.ReadErr == "" || !c.Known || c.State == nil || c.State.LastExecuted != "v1" {
				t.Fatalf("copy = %+v; want the last state kept and the failure recorded", c)
			}
			if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err == nil {
				t.Fatal("a claimed host's build state was written without its claim")
			}
		})
	}
}

// The production store opener: one store per known bucket, opened once,
// each with its lease's region; an unknown bucket is refused.
func TestStoresAreOpenedPerBucketAndCached(t *testing.T) {
	opened := map[string]int{}
	open := func(_ context.Context, bucket, region string) (release.Store, error) {
		opened[bucket+"@"+region]++
		return release.NewDirStore(t.TempDir()), nil
	}
	storeFor := cachedStores(map[string]string{"hib": "eu-west-1", "data": "us-east-1"}, open)
	ctx := context.Background()
	for _, bucket := range []string{"hib", "data", "hib", "data"} {
		if _, err := storeFor(ctx, bucket); err != nil {
			t.Fatal(err)
		}
	}
	if len(opened) != 2 || opened["hib@eu-west-1"] != 1 || opened["data@us-east-1"] != 1 {
		t.Fatalf("opened = %v", opened)
	}
	if _, err := storeFor(ctx, "elsewhere"); err == nil {
		t.Fatal("a bucket the box does not lease in was opened")
	}
}

func put(t *testing.T, b *buildStates, ctx context.Context, host string, s release.BuildState) error {
	t.Helper()
	_, err := b.update(ctx, host, func(*release.BuildState) release.BuildState { return s })
	return err
}

// --- update (ruling R-BS-UPDATE) ---

// bump is a merge that keeps what it read and raises Completed by one.
func bump(cur *release.BuildState) release.BuildState {
	next := release.BuildState{}
	if cur != nil {
		next = *cur
	}
	next.Completed++
	return next
}

// Concurrent updates for one host (an instance spawn and its worker or
// sidecar spawns, under the same lease) never lose a merge.
func TestConcurrentUpdatesKeepEveryMerge(t *testing.T) {
	b, _, reg := testBuildStates(t)
	ctx := context.Background()
	const n = 8
	errs := make(chan error, n)
	for i := 0; i < n; i++ {
		go func() {
			_, err := b.update(ctx, "a.example.com", bump)
			errs <- err
		}()
	}
	for i := 0; i < n; i++ {
		if err := <-errs; err != nil {
			t.Fatal(err)
		}
	}
	st, _, err := b.read(ctx, "a.example.com")
	if err != nil || st == nil || st.Completed != n {
		t.Fatalf("state = %+v, %v; want Completed %d", st, err, n)
	}
	if c := copyOf(t, reg); !c.Known || !c.Replica || c.State == nil || c.State.Completed != n {
		t.Fatalf("copy = %+v", c)
	}
}

// A write that lands between update's read and its Put (another process)
// is not overwritten: update re-reads and merges again.
func TestAnUpdateRacedBetweenReadAndPutMergesAgain(t *testing.T) {
	b, _, reg := testBuildStates(t)
	ctx := context.Background()
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1", Completed: 1}); err != nil {
		t.Fatal(err)
	}
	raced := false
	b.beforePut = func(ctx context.Context, s release.Store, key string) {
		if raced {
			return
		}
		raced = true
		other := []byte(`{"schema":1,"host":"a.example.com","last_executed":"v1","resolved":{"id":7,"target":"v3"},"completed":6}`)
		if _, err := s.Put(ctx, key, other, mustEtag(t, s, key)); err != nil {
			t.Fatal(err)
		}
	}
	merges := 0
	got, err := b.update(ctx, "a.example.com", func(cur *release.BuildState) release.BuildState {
		merges++
		next := *cur
		next.LastExecuted = "v2"
		return next
	})
	if err != nil {
		t.Fatal(err)
	}
	if merges != 2 || got.LastExecuted != "v2" || got.Completed != 6 || got.Resolved == nil || got.Resolved.ID != 7 {
		t.Fatalf("merges %d, got %+v; want the raced write's Completed and Resolved kept", merges, got)
	}
	st, _, err := b.read(ctx, "a.example.com")
	if err != nil || st == nil || st.Completed != 6 || st.LastExecuted != "v2" {
		t.Fatalf("stored %+v, %v", st, err)
	}
	if c := copyOf(t, reg); c.State == nil || c.State.Completed != 6 || c.State.LastExecuted != "v2" {
		t.Fatalf("copy = %+v", c)
	}
}

// A conflict on every try is an error, and the copy keeps what it had.
func TestAPersistentConflictIsAnError(t *testing.T) {
	b, _, reg := testBuildStates(t)
	ctx := context.Background()
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
		t.Fatal(err)
	}
	before := copyOf(t, reg)
	n := 0
	b.beforePut = func(ctx context.Context, s release.Store, key string) {
		n++
		other := []byte(fmt.Sprintf(`{"schema":1,"host":"a.example.com","last_executed":"v1","completed":%d}`, n))
		if _, err := s.Put(ctx, key, other, mustEtag(t, s, key)); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := b.update(ctx, "a.example.com", bump); !errors.Is(err, release.ErrCASConflict) {
		t.Fatalf("err = %v; want a CAS conflict", err)
	}
	// The put above left this agent a copy of the object, so the first try
	// goes on that copy (buildStates.write) and every fresh read still gets
	// its own: updateTries of them after it.
	if n != updateTries+1 {
		t.Fatalf("tried %d times, want %d (the copy's try and %d fresh ones)", n, updateTries+1, updateTries)
	}
	if after := copyOf(t, reg); after.Column() != before.Column() {
		t.Fatalf("a lost update changed the copy: %+v -> %+v", before, after)
	}
}

// The first update on a host with no object is a create-only write.
func TestTheFirstUpdateCreates(t *testing.T) {
	b, s, _ := testBuildStates(t)
	ctx := context.Background()
	b.beforePut = func(ctx context.Context, st release.Store, key string) {
		if _, _, err := st.Get(ctx, key); err == nil {
			return
		}
		if _, err := st.Put(ctx, key, []byte(`{"schema":1,"host":"a.example.com","completed":4}`), ""); err != nil {
			t.Fatal(err)
		}
	}
	got, err := b.update(ctx, "a.example.com", bump)
	if err != nil || got.Completed != 5 {
		t.Fatalf("got %+v, %v; want the created object merged (Completed 5)", got, err)
	}
	if _, _, err := s.Get(ctx, "inst/a.example.com/build-state.json"); err != nil {
		t.Fatal(err)
	}
}

// Minor (fix round 1): with host claims on only the claim bucket is
// registered, so a hibernate bucket that is the same bucket with no region
// set cannot fail the box.
func TestClaimsOnRegistersOnlyTheClaimBucket(t *testing.T) {
	read := func(context.Context, string) (hostclaim.Observed, error) { return hostclaim.Observed{}, nil }
	cfg := &activator.Config{Bucket: "data", Region: "", Prefix: "hib",
		HostWrites: hostwrite.Config{Enabled: true},
		Claims:     &activator.ClaimRuntime{Bucket: "data", LeaseRegion: "eu-west-1", Read: read}}
	regions, err := buildStateRegions(cfg)
	if err != nil {
		t.Fatal(err)
	}
	if len(regions) != 1 || regions["data"] != "eu-west-1" {
		t.Fatalf("regions = %v", regions)
	}
	b, err := newBuildStates(cfg, nil, "")
	if err != nil || b == nil || b.claimBucket != "data" || b.claims == nil {
		t.Fatalf("newBuildStates = %+v, %v", b, err)
	}
	cfg.Claims.Bucket = "other"
	if regions, err := buildStateRegions(cfg); err != nil || len(regions) != 1 || regions["other"] != "eu-west-1" {
		t.Fatalf("regions = %v, %v; want only the claim bucket", regions, err)
	}
	cfg.HostWrites.Enabled = false
	if regions, err := buildStateRegions(cfg); err != nil || len(regions) != 1 || regions["data"] != "" {
		t.Fatalf("claims off: regions = %v, %v", regions, err)
	}
	cfg.HostWrites.Enabled, cfg.Claims = true, nil
	if _, err := buildStateRegions(cfg); err == nil {
		t.Fatal("claims on with no claim runtime was accepted")
	}
}

// --- the pre-exec write (Service.BeforeExec, settings spec §4.2) ---

// writeConsumed writes root's consumed file for the test host. An untagged
// c is tagged with the route's tenant, as root writes every file since
// ruling R22. Tests of an untagged or foreign file use writeConsumedAsIs.
func writeConsumed(t *testing.T, b *buildStates, c release.Consumed) {
	t.Helper()
	c.Host = "a.example.com"
	if rt, err := b.reg.Get(c.Host); err == nil && rt != nil && c.Account == "" && c.App == "" {
		c.Account, c.App = release.RouteAccount(*rt), rt.App
	}
	if err := release.WriteConsumed(b.optDir, c); err != nil {
		t.Fatal(err)
	}
}

// The write before the exec names the build, carries root's resolution
// from the consumed file, and completes the request when the build is its
// target. The registry copy follows.
func TestBeforeExecCompletesAResolvedRequestWhenItsTargetRuns(t *testing.T) {
	b, _, reg := testBuildStates(t)
	ctx := context.Background()
	writeConsumed(t, b, release.Consumed{Resolved: &release.ResolvedRequest{ID: 7, Target: "v2"}, Completed: 3})
	if err := b.beforeExec(ctx, "a.example.com", "v1"); err != nil {
		t.Fatal(err)
	}
	st, _, err := b.read(ctx, "a.example.com")
	if err != nil {
		t.Fatal(err)
	}
	if st.LastExecuted != "v1" || st.Resolved == nil || st.Resolved.ID != 7 || st.Resolved.Target != "v2" || st.Completed != 3 {
		t.Fatalf("state after v1 = %+v, want the resolution carried and R pending", st)
	}
	if err := b.beforeExec(ctx, "a.example.com", "v2"); err != nil {
		t.Fatal(err)
	}
	st, _, _ = b.read(ctx, "a.example.com")
	if st.LastExecuted != "v2" || st.Completed != 7 || st.CompletedFailed {
		t.Fatalf("state after T = %+v, want R completed", st)
	}
	if c := copyOf(t, reg); c.State == nil || c.State.LastExecuted != "v2" || c.State.Completed != 7 || !c.Known {
		t.Fatalf("registry copy = %+v", c)
	}
}

// Fail closed: a consumed file the agent cannot read refuses the exec, and
// nothing is written. So does a build with no name.
func TestBeforeExecRefusesAnUnreadableConsumedFile(t *testing.T) {
	b, s, _ := testBuildStates(t)
	ctx := context.Background()
	if err := os.MkdirAll(filepath.Join(b.optDir, "consumed"), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(release.ConsumedPath(b.optDir, "a.example.com"), []byte("{not json"), 0o644); err != nil {
		t.Fatal(err)
	}
	if err := b.beforeExec(ctx, "a.example.com", "v1"); err == nil {
		t.Fatal("an unreadable consumed file let the exec through")
	}
	if _, _, err := s.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a build state was written without root's resolution: %v", err)
	}
	if err := os.Remove(release.ConsumedPath(b.optDir, "a.example.com")); err != nil {
		t.Fatal(err)
	}
	if err := b.beforeExec(ctx, "a.example.com", ""); err == nil {
		t.Fatal("a build with no name was recorded")
	}
}

// The write lands at the location the wake's context names (its lease).
func TestBeforeExecWritesAtTheContextLocation(t *testing.T) {
	b, hib, _ := testBuildStates(t)
	other := release.NewDirStore(t.TempDir())
	b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": other})
	ctx := activator.WithBuildStateLocation(context.Background(), activator.BuildStateLocation{Bucket: "data", Prefix: "instances/a.example.com/inc1"})
	if err := b.beforeExec(ctx, "a.example.com", "v1"); err != nil {
		t.Fatal(err)
	}
	if _, _, err := other.Get(ctx, "instances/a.example.com/inc1/build-state.json"); err != nil {
		t.Fatalf("the build state is not at the context's location: %v", err)
	}
	if _, _, err := hib.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the hibernate prefix was written too: %v", err)
	}
}

// Ruling R-BS-CORRUPT: an object that does not parse is never overwritten;
// the refusal names the key and says what to do.
func TestBeforeExecRefusesACorruptStateAndNamesIt(t *testing.T) {
	b, s, _ := testBuildStates(t)
	ctx := context.Background()
	key := "inst/a.example.com/build-state.json"
	if _, err := s.Put(ctx, key, []byte("{torn"), ""); err != nil {
		t.Fatal(err)
	}
	err := b.beforeExec(ctx, "a.example.com", "v1")
	if err == nil || !strings.Contains(err.Error(), key) || !strings.Contains(err.Error(), "repair or delete") {
		t.Fatalf("err = %v, want a refusal naming %s and saying to repair or delete it", err, key)
	}
	if raw, _, _ := s.Get(ctx, key); string(raw) != "{torn" {
		t.Fatalf("the corrupt object was overwritten: %q", raw)
	}
}

// Astra I3 #1: an initialization failure never leaves the nil hook.
func TestAFailedBuildStateStartRefusesEveryExec(t *testing.T) {
	hook := buildStateHook(nil, errors.New("s3: no credentials"))
	if hook == nil {
		t.Fatal("an initialization failure installed the nil hook: execs would run unrecorded")
	}
	if err := hook(context.Background(), "a.example.com", "v1"); err == nil {
		t.Fatal("the refusing hook let an exec through")
	}
	if buildStateHook(nil, nil) != nil {
		t.Fatal("a box with no replica bucket must keep today's nil hook")
	}
	b, _, _ := testBuildStates(t)
	if hook := buildStateHook(b, nil); hook == nil || hook(context.Background(), "a.example.com", "v1") != nil {
		t.Fatal("a started store's hook does not record")
	}
}

// --- the held wake check's answer (Service.HeldTarget, settings spec §4.3
// rule 3) ---

func setSettings(t *testing.T, reg *registry.Registry, u release.UpdateSettings) {
	t.Helper()
	if err := reg.SetUpdateSettings("a.example.com", u.Column()); err != nil {
		t.Fatal(err)
	}
}

func setCopy(t *testing.T, reg *registry.Registry, c release.BuildStateCopy) {
	t.Helper()
	if err := reg.SetBuildState("a.example.com", c.Column()); err != nil {
		t.Fatal(err)
	}
}

// stageWithManifest stages body as acct/app at ver, with its manifest
// sibling naming body's sha256 for this platform, as root's staging leaves
// a verified copy.
func stageWithManifest(t *testing.T, optDir, acct, app, ver string, body []byte) {
	t.Helper()
	if err := release.StageArtifact(optDir, acct, ver, app, body); err != nil {
		t.Fatal(err)
	}
	sum := sha256.Sum256(body)
	m := release.Manifest{Schema: 1, App: app, Version: ver, Kind: "binary",
		Artifacts: map[string]release.Artifact{runtime.GOOS + "-" + runtime.GOARCH: {File: app, SHA256: hex.EncodeToString(sum[:]), Size: int64(len(body))}}}
	raw, err := json.Marshal(m)
	if err != nil {
		t.Fatal(err)
	}
	if err := release.StageManifest(optDir, acct, ver, app, raw); err != nil {
		t.Fatal(err)
	}
}

func fileSHA(t *testing.T, path string) string {
	t.Helper()
	sum, err := release.FileSHA256(path)
	if err != nil {
		t.Fatal(err)
	}
	return sum
}

// Astra I1 #8, the agent's half: root reports a major hold with a request
// baking beside it (Consumed.Baking); the wake check still treats the
// instance as held, on last_executed.
func TestAMajorHoldStaysHeldWhileARequestBakes(t *testing.T) {
	b, _, reg := testBuildStates(t)
	two := 2
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", Major: &two, InstallRequest: 1})
	setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
	decided(t, b, release.Consumed{Reason: string(release.ReasonMajor), Desired: "v2", Baking: "v3"})
	if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v2" {
		t.Fatalf("heldTarget = %+v; want v2, held", h)
	}
}

// Astra I2 #7: a pin set after a request resolved outranks it, as in root's
// Target. The wake targets the pin P, never the request's T.
func TestAPinOutranksAPendingRequestAtTheWake(t *testing.T) {
	b, _, reg := testBuildStates(t)
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable", Pin: "v5", InstallRequest: 2})
	setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2", Completed: 1}})
	writeConsumed(t, b, release.Consumed{Completed: 1, Resolved: &release.ResolvedRequest{ID: 2, Target: "v4"}})
	if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v5" {
		t.Fatalf("heldTarget = %+v; want the pin v5", h)
	}
	// Without the pin, the pending request's T outranks last_executed.
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable", InstallRequest: 2})
	if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v4" {
		t.Fatalf("heldTarget = %+v; want the request's v4", h)
	}
}

// Ruling R-RESOLUTION-STATE-WINS: when root's file and the build state
// resolve the same request to different builds (a box that resolved R, lost
// the instance to a move before any wake, and got it back after another box
// resolved R again), the wake targets the build state's T, as root does.
func TestAHeldWakeFollowsTheBuildStatesResolutionOverTheFile(t *testing.T) {
	b, _, reg := testBuildStates(t)
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable", InstallRequest: 1})
	setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true,
		State: &release.BuildState{LastExecuted: "v1", Resolved: &release.ResolvedRequest{ID: 1, Target: "v4"}}})
	writeConsumed(t, b, release.Consumed{Reason: string(release.ReasonRequest), Desired: "v1", Resolved: &release.ResolvedRequest{ID: 1, Target: "v1"}})
	if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v4" {
		t.Fatalf("heldTarget = %+v; want the build state's v4, held", h)
	}
}

// Task 2 review: with the build state unread (Known false) root does no
// request bookkeeping and will not install T, so neither does the wake
// follow it. Held settings then read root's unknown; auto ones are not held
// (root's decisions here are for the current column, ruling R-HELD-PURE). A
// copy whose last refresh failed is refused as unread, whatever the settings
// (Astra round 1 finding 1), and never follows T either.
func TestAnUnreadBuildStateDoesNotFollowAPendingRequest(t *testing.T) {
	b, _, reg := testBuildStates(t)
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable", InstallRequest: 2})
	setCopy(t, reg, release.BuildStateCopy{Known: false})
	decided(t, b, release.Consumed{Reason: string(release.ReasonUnknown), Resolved: &release.ResolvedRequest{ID: 2, Target: "v4"}})
	if h := b.heldTarget("a.example.com"); !h.Held || !h.Unknown || h.Target != "" {
		t.Fatalf("heldTarget = %+v; want held, unknown, never the request's v4", h)
	}
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", InstallRequest: 2})
	decided(t, b, release.Consumed{Reason: string(release.ReasonAdopt), Desired: "v5", Resolved: &release.ResolvedRequest{ID: 2, Target: "v4"}})
	if h := b.heldTarget("a.example.com"); h.Held {
		t.Fatalf("heldTarget = %+v; an auto instance with an unread state follows root's adoption", h)
	}
	setCopy(t, reg, release.BuildStateCopy{Known: false, ReadErr: "s3: timeout"})
	if h := b.heldTarget("a.example.com"); !h.Held || !h.Unread || h.Target != "" {
		t.Fatalf("heldTarget = %+v; a failed refresh is refused as unread, never the request's v4", h)
	}
}

// Ruling R-SETTINGS-UNREADABLE (Task 6): a column that does not parse even
// leniently holds the wake as unknown when root says unknown for it. With no
// word from root for that column the wake holds and asks root (ruling
// R-HELD-PURE: never "not held"). A Control name this binary does not know
// is dropped, not read as unreadable, so the hold stays.
func TestAnUnreadableColumnHoldsOnlyAsRootDoes(t *testing.T) {
	b, _, reg := testBuildStates(t)
	if err := reg.SetUpdateSettings("a.example.com", `{"updates":"sometimes"}`); err != nil {
		t.Fatal(err)
	}
	if h := b.heldTarget("a.example.com"); !h.Held || !h.AskRoot || h.Unknown {
		t.Fatalf("heldTarget = %+v with no word from root; want held, asking root", h)
	}
	decided(t, b, release.Consumed{Reason: string(release.ReasonUnknown), Detail: "update settings: updates \"sometimes\""})
	if h := b.heldTarget("a.example.com"); !h.Held || !h.Unknown {
		t.Fatalf("heldTarget = %+v; want held as unknown, as root holds it", h)
	}
	if err := reg.SetUpdateSettings("a.example.com", `{"updates":"manual","channel":"stable","control":["updates","colour"]}`); err != nil {
		t.Fatal(err)
	}
	setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
	decided(t, b, release.Consumed{Reason: string(release.ReasonManual), Desired: "v2"})
	if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v2" {
		t.Fatalf("heldTarget = %+v; want the manual hold on v2 through an unknown Control name", h)
	}
}

// Astra I3 #2: root's failed decision, or its refusal record, marks the
// target failed, so the wake's own rule refuses it with no planner. Only a
// not-up refusal is a build failure (ruling R-REFUSALKIND).
func TestARootFailedTargetIsReportedFailed(t *testing.T) {
	b, _, reg := testBuildStates(t)
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", Pin: "v5"})
	writeConsumed(t, b, release.Consumed{Reason: string(release.ReasonFailed), Detail: "v5"})
	stageWithManifest(t, b.optDir, "bab", "km", "v5", []byte("five"))
	if h := b.heldTarget("a.example.com"); !h.Held || !h.Failed || h.Target != "v5" || h.Staged != nil {
		t.Fatalf("heldTarget = %+v; want v5 held and failed, no retained copy offered", h)
	}
	writeConsumed(t, b, release.Consumed{Reason: string(release.ReasonPin), Detail: "v5"})
	for _, kind := range []string{release.RefusalWrongProgram, release.RefusalNotProven} {
		if err := release.WriteRefusal(b.bakeDir, release.Refusal{Kind: kind, Host: "a.example.com", App: "km", Version: "v5", Reason: "refused"}); err != nil {
			t.Fatal(err)
		}
		if h := b.heldTarget("a.example.com"); h.Failed {
			t.Fatalf("a %s refusal of the pin = %+v; it is not a build failure", kind, h)
		}
	}
	if err := release.WriteRefusal(b.bakeDir, release.Refusal{Host: "a.example.com", App: "km", Version: "v5", Reason: "not up"}); err != nil {
		t.Fatal(err)
	}
	if h := b.heldTarget("a.example.com"); !h.Failed || h.Staged != nil {
		t.Fatalf("a not-up refusal record for the pin = %+v; want failed", h)
	}
}

// Astra I3 #5: a retained copy whose bytes match its staged manifest is
// offered; one whose bytes differ is not.
func TestARetainedCopyIsOfferedOnlyWhenVerified(t *testing.T) {
	b, _, reg := testBuildStates(t)
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"})
	setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
	decided(t, b, release.Consumed{Reason: string(release.ReasonManual), Desired: "v2"})
	stageWithManifest(t, b.optDir, "bab", "km", "v2", []byte("two"))
	h := b.heldTarget("a.example.com")
	if !h.Held || h.Target != "v2" || h.Staged == nil || h.Staged.Version != "v2" || h.Staged.StagedPath != release.VersionPath(b.optDir, "bab", "v2", "km") || h.Staged.Digest != fileSHA(t, h.Staged.StagedPath) {
		t.Fatalf("heldTarget = %+v; want the verified retained v2", h)
	}
	if err := os.WriteFile(release.VersionPath(b.optDir, "bab", "v2", "km"), []byte("tampered"), 0o755); err != nil {
		t.Fatal(err)
	}
	if h := b.heldTarget("a.example.com"); h.Staged != nil {
		t.Fatal("a staged copy whose bytes differ from its manifest was offered")
	}
}

// Default settings are today's wake: not held, and nothing is read for it.
// Held settings refresh the copy first (§4.2: "refreshed ... before every
// wake"), so a last_executed the reconcile has not copied yet still counts.
func TestTheWakeTargetRefreshesOnlyAHeldRoute(t *testing.T) {
	b, s, reg := testBuildStates(t)
	ctx := context.Background()
	if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err != nil {
		t.Fatal(err)
	}
	setCopy(t, reg, release.BuildStateCopy{}) // what the last reconcile saw: nothing yet
	gets := &getCounter{Store: s}
	b.storeFor = storesOf(map[string]release.Store{"hib": gets})
	if h := b.wakeTarget(ctx, "a.example.com"); h.Held || gets.n() != 0 {
		t.Fatalf("default settings: %+v after %d reads; want today's wake, nothing read", h, gets.n())
	}
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"})
	writeConsumed(t, b, release.Consumed{Reason: string(release.ReasonManual), Desired: "v2"})
	if h := b.wakeTarget(ctx, "a.example.com"); !h.Held || h.Target != "v2" || gets.n() == 0 {
		t.Fatalf("held: %+v after %d reads; want the refreshed last_executed v2", h, gets.n())
	}
}

// Ruling R-HELD-SCOPE (Task 12a): settings hold only the rows root applies
// them to (Kind instance, no Unit; adopt.go's settings condition). A named
// unit row and an exec worker row under an app-policy manual wake as today:
// not held, and nothing is read for them. Before, the unit row read
// "build unknown" on every wake with nothing to clear it, and the worker was
// held on a build root never targets.
func TestOnlyRowsRootHoldsAreHeldAtTheWake(t *testing.T) {
	b, s, reg := testBuildStates(t)
	ctx := context.Background()
	manual := release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"}
	for _, rt := range []registry.Route{
		{Host: "unit.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/u.sock", Unit: "km-unit"},
		{Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true},
	} {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
		if err := reg.SetUpdateSettings(rt.Host, manual.Column()); err != nil {
			t.Fatal(err)
		}
	}
	// The worker keeps a build state (its lease's), read and copied.
	if err := put(t, b, ctx, "w.example.com", release.BuildState{LastExecuted: "v2"}); err != nil {
		t.Fatal(err)
	}
	if err := reg.SetBuildState("w.example.com", release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}}.Column()); err != nil {
		t.Fatal(err)
	}
	gets := &getCounter{Store: s}
	b.storeFor = storesOf(map[string]release.Store{"hib": gets})
	for _, host := range []string{"unit.example.com", "w.example.com"} {
		if h := b.heldTarget(host); h.Held || h.Unknown || h.Target != "" {
			t.Errorf("%s: heldTarget = %+v; want today's wake", host, h)
		}
		if h := b.wakeTarget(ctx, host); h.Held || h.Unknown || h.Target != "" {
			t.Errorf("%s: wakeTarget = %+v; want today's wake", host, h)
		}
	}
	if n := gets.n(); n != 0 {
		t.Errorf("%d bucket reads for rows root never holds; want none", n)
	}
	// The instance row under the same settings is still held.
	setSettings(t, reg, manual)
	setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v3"}})
	if h := b.heldTarget("a.example.com"); !h.Held || h.Target != "v3" {
		t.Fatalf("the instance: heldTarget = %+v; want held on v3", h)
	}
}

type getCounter struct {
	release.Store
	mu   sync.Mutex
	gets int
}

func (g *getCounter) Get(ctx context.Context, key string) ([]byte, string, error) {
	g.mu.Lock()
	g.gets++
	g.mu.Unlock()
	return g.Store.Get(ctx, key)
}

func (g *getCounter) n() int {
	g.mu.Lock()
	defer g.mu.Unlock()
	return g.gets
}

// Ruling R-HELD-FAILED: root's rule 3 rewrites a major hold whose
// last_executed this box holds as failed to Reason failed, Detail that
// build. The wake still reads the major hold: held on it, and failed, so
// with no planner the wake is refused rather than run the live link.
func TestAMajorHoldOnAFailedBuildStaysHeldAndFailed(t *testing.T) {
	b, _, reg := testBuildStates(t)
	two := 2
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesAuto, Channel: "stable", Major: &two})
	setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
	decided(t, b, release.Consumed{Reason: string(release.ReasonFailed), Detail: "v2"})
	if h := b.heldTarget("a.example.com"); !h.Held || !h.Failed || h.Target != "v2" {
		t.Fatalf("heldTarget = %+v; want v2 held and failed", h)
	}
	// Root's failed head under rule 4 (Detail is the head, not
	// last_executed) is not a hold.
	decided(t, b, release.Consumed{Reason: string(release.ReasonFailed), Detail: "v3"})
	if h := b.heldTarget("a.example.com"); h.Held {
		t.Fatalf("heldTarget = %+v; a failed head under adoption is not a hold", h)
	}
}

// Ruling R-HELD-UNKNOWN: rule 3 as root's Target decides it. An unread copy
// is unknown whatever root last wrote; a read copy with last_executed
// targets it over a stale unknown; root's unknown stands for a read copy
// with no last_executed.
func TestAHeldWakeReadsTheCopyAsRootsTargetDoes(t *testing.T) {
	b, _, reg := testBuildStates(t)
	setSettings(t, reg, release.UpdateSettings{Updates: release.UpdatesManual, Channel: "stable"})
	setCopy(t, reg, release.BuildStateCopy{}) // never read
	decided(t, b, release.Consumed{Reason: string(release.ReasonManual), Desired: "v2"})
	if h := b.heldTarget("a.example.com"); !h.Held || !h.Unknown {
		t.Fatalf("an unread copy under manual = %+v; want held, unknown", h)
	}
	setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true, State: &release.BuildState{LastExecuted: "v2"}})
	decided(t, b, release.Consumed{Reason: string(release.ReasonUnknown)}) // written while the copy was unread
	if h := b.heldTarget("a.example.com"); !h.Held || h.Unknown || h.Target != "v2" {
		t.Fatalf("a read copy with last_executed under root's stale unknown = %+v; want held on v2", h)
	}
	setCopy(t, reg, release.BuildStateCopy{Known: true, Replica: true})
	if h := b.heldTarget("a.example.com"); !h.Held || !h.Unknown {
		t.Fatalf("a read copy with no last_executed and root's unknown = %+v; want unknown", h)
	}
}

// budgetEater records which host's build state each read asks for and
// holds the read until the caller's context ends: the host it is reading
// spends the whole refresh budget.
type budgetEater struct {
	release.Store
	mu    sync.Mutex
	hosts []string
}

func (e *budgetEater) Get(ctx context.Context, key string) ([]byte, string, error) {
	if host, ok := strings.CutSuffix(strings.TrimPrefix(key, "inst/"), "/build-state.json"); ok {
		e.mu.Lock()
		e.hosts = append(e.hosts, host)
		e.mu.Unlock()
		<-ctx.Done()
		return nil, "", ctx.Err()
	}
	return e.Store.Get(ctx, key)
}

// Ruling R-REFRESH-ROTATE: a box whose refresh budget runs out does not
// starve the same tail every pass. Each refreshAll starts at the first host
// the last one did not reach, so every host is reached in turn.
func TestRefreshAllRotatesPastWhereTheLastOneStopped(t *testing.T) {
	t.Parallel()
	b, s, reg := testBuildStates(t)
	for _, h := range []string{"b.example.com", "c.example.com"} {
		if err := reg.Upsert(registry.Route{Host: h, Kind: "instance", App: "km", Hibernate: true, Socket: "/run/carlos/" + h + ".sock"}); err != nil {
			t.Fatal(err)
		}
	}
	eater := &budgetEater{Store: s}
	b.storeFor = storesOf(map[string]release.Store{"hib": eater})
	b.budget = 200 * time.Millisecond
	for i := 0; i < 4; i++ {
		if skipped := b.refreshAll(context.Background()); skipped != 2 {
			t.Fatalf("pass %d skipped %d, want 2 (one host eats the budget)", i, skipped)
		}
	}
	want := []string{"a.example.com", "b.example.com", "c.example.com", "a.example.com"}
	eater.mu.Lock()
	defer eater.mu.Unlock()
	if strings.Join(eater.hosts, " ") != strings.Join(want, " ") {
		t.Fatalf("passes read %v, want %v: each pass starts where the last stopped", eater.hosts, want)
	}
}