package console

import (
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/sns"
	"github.com/carlosframework/platform/internal/alerting"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/release"
)

func lossPageRecord(id string, pending bool, knowledge hostclaim.TimerKnowledge) hostclaim.LossRecord {
	at := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
	r := hostclaim.LossRecord{V: 1, ID: id, State: "prepared", Actor: "operator", AcceptedAt: at,
		Host: "host.example", HolderIncarnation: strings.Repeat("a", 64), HolderStamp: 1,
		ExpectedClaimGen: 1, ExpectedClaimETag: "claim", OpID: "operation", OpKind: hostclaim.KindMoving,
		OpPhase: "precommit", Box: "box-a", Enrollment: "enrollment", QuarantineRevision: 1, FencedAt: at,
		CloudObservation:      hostclaim.LossCloudObservation{ObservedAt: at, Account: "123", Region: "eu-west-1", InstanceID: "i-1", State: "stopped", RequestID: "request"},
		OperatorAttestation:   hostclaim.LossOperatorAttestation{MaintainComputeFence: true, ColdQuarantineStartOnly: true},
		ConfirmedConsequences: hostclaim.LossConsequences{RecoveryIntent: json.RawMessage(`{"kind":"lost-target"}`)},
		AlertsPending:         pending, Timers: hostclaim.TimerLoss{Knowledge: knowledge}}
	if knowledge != hostclaim.TimersUnknown {
		r.Timers.Source = &hostclaim.TimerSource{Bucket: "bucket", Key: "timers", VersionID: "v1", SHA256: strings.Repeat("b", 64), CapturedAt: at}
	}
	return r
}

func lossPageWorld(t *testing.T) (*Server, *release.DirStore, release.Signer, *[]map[string]string) {
	t.Helper()
	raw := release.NewDirStore(t.TempDir())
	_, key, err := ed25519.GenerateKey(nil)
	if err != nil {
		t.Fatal(err)
	}
	publishRestoreRoster(t, raw, key)
	s := newEnabledRestoreConsole(t, raw, key, crConsole)
	var pages []map[string]string
	alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(_ string, body []byte) error {
		var page map[string]string
		if err := json.Unmarshal(body, &page); err != nil {
			return err
		}
		pages = append(pages, page)
		return nil
	}})
	t.Cleanup(func() { alerting.Arm(nil) })
	return s, raw, release.RootSigner(key), &pages
}

func putLossPageRecord(t *testing.T, raw release.Store, signer release.Signer, r hostclaim.LossRecord) {
	t.Helper()
	if _, err := hostclaim.CreateLossRecord(t.Context(), raw, signer, r); err != nil {
		t.Fatal(err)
	}
}

func TestClaimLossPagesIntervalAndRestart(t *testing.T) {
	s, raw, signer, pages := lossPageWorld(t)
	for _, id := range []string{"loss-one", "loss-two"} {
		putLossPageRecord(t, raw, signer, lossPageRecord(id, true, hostclaim.TimersComplete))
	}
	now := time.Date(2026, 10, 12, 13, 0, 0, 123, time.UTC)
	pass := func() {
		t.Helper()
		if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now }); err != nil {
			t.Fatal(err)
		}
	}
	pass()
	pass()
	if len(*pages) != 2 {
		t.Fatalf("first passes: %v", *pages)
	}
	// Reopen the actual SQLite file; no in-memory limiter survives this.
	path := s.store.gate.path
	if err := s.store.Close(); err != nil {
		t.Fatal(err)
	}
	st, err := OpenGated(path)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { st.Close() })
	s = &Server{cfg: s.cfg, store: st}
	now = now.Add(time.Hour - time.Nanosecond)
	pass()
	if len(*pages) != 2 {
		t.Fatalf("restart lost rate limit: %v", *pages)
	}
	now = now.Add(time.Nanosecond)
	pass()
	if len(*pages) != 4 {
		t.Fatalf("interval did not repeat each record: %v", *pages)
	}
	key, _ := hostclaim.LossRecordKey("loss-one")
	if err := raw.Delete(t.Context(), key); err != nil {
		t.Fatal(err)
	}
	now = now.Add(time.Hour)
	pass()
	if len(*pages) != 5 || !strings.Contains((*pages)[4]["body"], "loss-two") {
		t.Fatalf("absent pending record paged: %v", *pages)
	}
}

func TestClaimLossPagesClearedAndTimers(t *testing.T) {
	for _, knowledge := range []hostclaim.TimerKnowledge{hostclaim.TimersComplete, hostclaim.TimersPartial, hostclaim.TimersUnknown} {
		t.Run(string(knowledge), func(t *testing.T) {
			s, raw, signer, pages := lossPageWorld(t)
			r := lossPageRecord("loss-timer", false, knowledge)
			putLossPageRecord(t, raw, signer, r)
			now := r.AcceptedAt
			if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now }); err != nil {
				t.Fatal(err)
			}
			want := 1
			if knowledge == hostclaim.TimersComplete {
				want = 0
			}
			if len(*pages) != want {
				t.Fatalf("pages: %v", *pages)
			}
			if want != 0 && (!strings.Contains((*pages)[0]["body"], r.Host) || !strings.Contains((*pages)[0]["body"], "re-arm")) {
				t.Fatalf("timer instructions: %v", *pages)
			}
			key, _ := hostclaim.LossRecordKey(r.ID)
			if err := raw.Delete(t.Context(), key); err != nil {
				t.Fatal(err)
			}
			if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now.Add(2 * time.Hour) }); err != nil {
				t.Fatal(err)
			}
			if len(*pages) != want {
				t.Fatal("cleared record paged")
			}
		})
	}
}

func TestClaimLossPagesRecoveryWiringWithoutMovesOrClaims(t *testing.T) {
	s, raw, signer, pages := lossPageWorld(t)
	putLossPageRecord(t, raw, signer, lossPageRecord("orphan-loss", true, hostclaim.TimersComplete))
	if s.cfg.MovesEnabled {
		t.Fatal("fixture enabled moves")
	}
	if err := s.RunClaimRecoveryPass(t.Context()); err != nil {
		t.Fatal(err)
	}
	waitClaimLossPages(t, s)
	if len(*pages) != 1 {
		t.Fatalf("record without a claim was not paged: %v", *pages)
	}
}

func TestClaimLossPagesFlagOff(t *testing.T) {
	calls := 0
	alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(string, []byte) error { calls++; return nil }})
	t.Cleanup(func() { alerting.Arm(nil) })
	s := &Server{}
	if err := s.pageClaimLossesAt(t.Context(), time.Now); err != nil {
		t.Fatal(err)
	}
	if err := s.RunClaimRecoveryPass(t.Context()); err != nil {
		t.Fatal(err)
	}
	s.scheduleClaimLossPages(t.Context())
	if s.claimLossPageDone != nil {
		t.Fatal("disabled runtime started a worker")
	}
	if calls != 0 {
		t.Fatal("disabled runtime paged")
	}
	st, err := Open(filepath.Join(t.TempDir(), "off.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer st.Close()
	var count int
	if err := st.db.QueryRow(`SELECT count(*) FROM sqlite_master WHERE name = 'claim_loss_pages'`).Scan(&count); err != nil || count != 0 {
		t.Fatalf("flag-off schema changed: %d %v", count, err)
	}
}

func TestClaimLossPageFailedDeliveryIsRateLimited(t *testing.T) {
	s, raw, signer, _ := lossPageWorld(t)
	putLossPageRecord(t, raw, signer, lossPageRecord("failed-page", true, hostclaim.TimersComplete))
	calls := 0
	failure := errors.New("delivery failed")
	alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(string, []byte) error { calls++; return failure }})
	now := time.Now()
	if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now }); !errors.Is(err, failure) {
		t.Fatalf("delivery: %v", err)
	}
	if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now.Add(time.Minute) }); err != nil {
		t.Fatal(err)
	}
	if calls != 1 {
		t.Fatalf("attempts: %d", calls)
	}
}

func TestClaimLossPagesAutomaticReportingDoesNotAcknowledge(t *testing.T) {
	s, raw, _, pages := lossPageWorld(t)
	ref := hostclaim.FenceRef{Enrollment: strings.Repeat("a", 32), SHA256: strings.Repeat("b", 64), Key: "control/box-fences/" + strings.Repeat("a", 32) + ".json"}
	sum := sha256.Sum256([]byte("operation\x00box-a\x00" + ref.Enrollment + "\x00" + ref.SHA256))
	id := "auto-" + hex.EncodeToString(sum[:16])
	loss := hostclaim.AutomaticLoss{ID: id, Box: "box-a", Enrollment: ref.Enrollment, Fence: ref,
		ObservedAt: time.Now().UTC().Format(time.RFC3339Nano), Op: "operation", Kind: hostclaim.KindLanding,
		Phase: hostclaim.PhaseProbing, Role: "target", Incarnation: strings.Repeat("c", 64), Stamp: 1, Timers: hostclaim.TimersUnknown}
	_, err := hostclaim.Update(t.Context(), raw, hostclaim.Observed{Host: "host.example", Absent: true}, func(c *hostclaim.Claim) error {
		c.LossRefs = []string{id}
		c.EvidenceRefs = []hostclaim.AutomaticLoss{loss}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := s.RunClaimRecoveryPass(t.Context()); err != nil {
		t.Fatal(err)
	}
	waitClaimLossPages(t, s)
	if len(*pages) != 2 {
		t.Fatalf("materialized record must page both conditions in the same pass: %v", *pages)
	}
	c, err := hostclaim.Read(t.Context(), raw, "host.example")
	if err != nil || !c.Claim.EvidenceRefs[0].Reported {
		t.Fatalf("reporting did not finish: %+v %v", c, err)
	}
	if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return time.Now().Add(2 * time.Hour) }); err != nil {
		t.Fatal(err)
	}
	if len(*pages) != 4 {
		t.Fatal("claim reporting silenced pending operator alerts")
	}
}

func TestClaimLossPageReservationSerializesPasses(t *testing.T) {
	st := openGatedStore(t)
	now := time.Now()
	var wg sync.WaitGroup
	var attempts atomic.Int64
	for i := 0; i < 12; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			ok, err := st.ClaimLossPageAttempt(t.Context(), "pending loss", now)
			if err != nil {
				t.Error(err)
			}
			if ok {
				attempts.Add(1)
			}
		}()
	}
	wg.Wait()
	if attempts.Load() != 1 {
		t.Fatalf("reserved %d attempts", attempts.Load())
	}
}

func TestClaimLossPagesRejectBadEvidenceAndContinue(t *testing.T) {
	s, raw, signer, pages := lossPageWorld(t)
	if _, err := raw.Put(t.Context(), "control/host-loss/bad.json", []byte(`{}`), ""); err != nil {
		t.Fatal(err)
	}
	putLossPageRecord(t, raw, signer, lossPageRecord("valid", true, hostclaim.TimersComplete))
	if err := s.pageClaimLossesAt(t.Context(), time.Now); !errors.Is(err, hostclaim.ErrLossRecord) {
		t.Fatalf("invalid signature accepted: %v", err)
	}
	if len(*pages) != 1 {
		t.Fatalf("bad record blocked good record: %v", *pages)
	}
}

func TestClaimLossPagesJournalFailureDoesNotSend(t *testing.T) {
	s, raw, signer, pages := lossPageWorld(t)
	putLossPageRecord(t, raw, signer, lossPageRecord("loss", true, hostclaim.TimersComplete))
	if err := s.store.Close(); err != nil {
		t.Fatal(err)
	}
	if err := s.pageClaimLossesAt(t.Context(), time.Now); err == nil {
		t.Fatal("journal error hidden")
	}
	if len(*pages) != 0 {
		t.Fatal("sent without durable reservation")
	}
}

func waitClaimLossPages(t *testing.T, s *Server) {
	t.Helper()
	s.claimLossPageMu.Lock()
	done := s.claimLossPageDone
	s.claimLossPageMu.Unlock()
	if done == nil {
		t.Fatal("no loss paging worker scheduled")
	}
	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("loss paging worker did not finish")
	}
}

func TestClaimLossPagesHourlyToDailyAcrossReopen(t *testing.T) {
	s, raw, signer, pages := lossPageWorld(t)
	r := lossPageRecord("old-loss", true, hostclaim.TimersUnknown)
	putLossPageRecord(t, raw, signer, r)
	first := r.AcceptedAt.Add(30 * 24 * time.Hour)
	pass := func(at time.Time, want int) {
		t.Helper()
		if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return at }); err != nil {
			t.Fatal(err)
		}
		if len(*pages) != want {
			t.Fatalf("at %s: got %d pages, want %d", at.Sub(first), len(*pages), want)
		}
	}
	for hour := 0; hour < 24; hour++ {
		pass(first.Add(time.Duration(hour)*time.Hour), 2*(hour+1))
	}
	path := s.store.gate.path
	if err := s.store.Close(); err != nil {
		t.Fatal(err)
	}
	st, err := OpenGated(path)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { st.Close() })
	s = &Server{cfg: s.cfg, store: st}
	pass(first.Add(24*time.Hour-time.Nanosecond), 48)
	// Once the first 24 hours end, the last attempt must be a day old.
	pass(first.Add(24*time.Hour), 48)
	pass(first.Add(47*time.Hour-time.Nanosecond), 48)
	pass(first.Add(47*time.Hour), 50)
	pass(first.Add(71*time.Hour-time.Nanosecond), 50)
	pass(first.Add(71*time.Hour), 52)
	for _, condition := range []string{"pending", "timers"} {
		var saved int64
		if err := st.db.QueryRow(`SELECT first_attempted_at FROM claim_loss_pages WHERE rule = ?`, condition+" "+r.ID).Scan(&saved); err != nil || saved != first.UnixNano() {
			t.Fatalf("first %s attempt changed: %d %v", condition, saved, err)
		}
	}
}

func TestClaimLossPagesClockAtEachReservation(t *testing.T) {
	s, raw, signer, _ := lossPageWorld(t)
	for _, id := range []string{"a-early", "b-late"} {
		putLossPageRecord(t, raw, signer, lossPageRecord(id, true, hostclaim.TimersComplete))
	}
	first := time.Now()
	now := first
	late := 0
	calls := 0
	alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(_ string, b []byte) error {
		calls++
		if strings.Contains(string(b), "b-late") {
			late++
		}
		if calls == 1 {
			now = now.Add(time.Hour)
		}
		return nil
	}})
	clock := func() time.Time { return now }
	if err := s.pageClaimLossesAt(t.Context(), clock); err != nil {
		t.Fatal(err)
	}
	now = now.Add(10 * time.Second)
	if err := s.pageClaimLossesAt(t.Context(), clock); err != nil {
		t.Fatal(err)
	}
	if late != 1 || calls != 3 {
		t.Fatalf("late record retried ten seconds later: late=%d, total=%d", late, calls)
	}
	var reserved int64
	if err := s.store.db.QueryRow(`SELECT attempted_at FROM claim_loss_pages WHERE rule = 'pending b-late'`).Scan(&reserved); err != nil || reserved != first.Add(time.Hour).UnixNano() {
		t.Fatalf("late reservation: %d %v", reserved, err)
	}
}

func TestClaimLossPagesBlockedSinkDoesNotBlockRecovery(t *testing.T) {
	s, raw, signer, _ := lossPageWorld(t)
	putLossPageRecord(t, raw, signer, lossPageRecord("blocking", true, hostclaim.TimersComplete))
	entered, releaseSink := make(chan struct{}), make(chan struct{})
	var unblock sync.Once
	var calls atomic.Int64
	alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(string, []byte) error {
		if calls.Add(1) == 1 {
			close(entered)
			<-releaseSink
		}
		return nil
	}})
	t.Cleanup(func() {
		unblock.Do(func() { close(releaseSink) })
		waitClaimLossPages(t, s)
	})
	var records, purges, admissions atomic.Int64
	wrap := func(fn func(context.Context) error, count *atomic.Int64) func(context.Context) error {
		return func(ctx context.Context) error { count.Add(1); return fn(ctx) }
	}
	r := s.cfg.HostClaimRecovery
	r.SweepRecords = wrap(r.SweepRecords, &records)
	r.SweepPurges = wrap(r.SweepPurges, &purges)
	r.SweepAdmissions = wrap(r.SweepAdmissions, &admissions)
	pass := func() {
		t.Helper()
		done := make(chan error, 1)
		go func() { done <- s.RunClaimRecoveryPass(t.Context()) }()
		select {
		case err := <-done:
			if err != nil {
				t.Fatal(err)
			}
		case <-time.After(2 * time.Second):
			unblock.Do(func() { close(releaseSink) })
			<-done
			t.Fatal("paging blocked recovery")
		}
	}
	pass()
	select {
	case <-entered:
	case <-time.After(2 * time.Second):
		t.Fatal("sink was not called")
	}
	s.claimLossPageMu.Lock()
	worker := s.claimLossPageDone
	s.claimLossPageMu.Unlock()
	for range 3 {
		pass()
	}
	if records.Load() != 4 || purges.Load() != 4 || admissions.Load() != 4 {
		t.Fatalf("sweeps delayed: records=%d purges=%d admissions=%d", records.Load(), purges.Load(), admissions.Load())
	}
	s.claimLossPageMu.Lock()
	overlapped := s.claimLossPageDone != worker
	s.claimLossPageMu.Unlock()
	if overlapped || calls.Load() != 1 {
		t.Fatal("paging workers overlapped")
	}
	unblock.Do(func() { close(releaseSink) })
	waitClaimLossPages(t, s)
	putLossPageRecord(t, raw, signer, lossPageRecord("next-worker", true, hostclaim.TimersComplete))
	pass()
	waitClaimLossPages(t, s)
	if calls.Load() != 2 {
		t.Fatal("next recovery did not schedule a new paging worker")
	}
}

func TestClaimLossPageJournalUpgrade(t *testing.T) {
	path := filepath.Join(t.TempDir(), "old.db")
	st, err := Open(path)
	if err != nil {
		t.Fatal(err)
	}
	at := time.Now().Add(-23 * time.Hour)
	_, err = st.db.Exec(`CREATE TABLE claim_loss_pages (rule TEXT PRIMARY KEY, attempted_at INTEGER NOT NULL);
INSERT INTO claim_loss_pages VALUES ('pending old', ?)`, at.UnixNano())
	if err != nil {
		t.Fatal(err)
	}
	if err := st.Close(); err != nil {
		t.Fatal(err)
	}
	for range 2 {
		st, err := OpenGated(path)
		if err != nil {
			t.Fatal(err)
		}
		var first, last int64
		err = st.db.QueryRow(`SELECT first_attempted_at, attempted_at FROM claim_loss_pages WHERE rule = 'pending old'`).Scan(&first, &last)
		var cursor string
		cursorErr := st.db.QueryRow(`SELECT last_key FROM claim_loss_pages WHERE rule = 'pending old'`).Scan(&cursor)
		st.Close()
		if err != nil || first != at.UnixNano() || last != at.UnixNano() {
			t.Fatalf("upgrade lost durable attempt: %d %d %v", first, last, err)
		}
		if cursorErr != nil || cursor != "" {
			t.Fatalf("upgrade scan progress: %q %v", cursor, cursorErr)
		}
	}
}

type lossPageDeadlineStore struct {
	release.Store
	deadline chan time.Time
}

func (s lossPageDeadlineStore) List(ctx context.Context, prefix string) ([]string, error) {
	deadline, _ := ctx.Deadline()
	s.deadline <- deadline
	<-ctx.Done()
	return nil, ctx.Err()
}

func TestClaimLossPagesWorkerBudget(t *testing.T) {
	s, _, _, pages := lossPageWorld(t)
	deadline := make(chan time.Time, 1)
	s.cfg.Releases = lossPageDeadlineStore{Store: s.cfg.Releases, deadline: deadline}
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	before := time.Now()
	s.scheduleClaimLossPages(ctx)
	select {
	case at := <-deadline:
		if at.Before(before.Add(claimLossPageBudget)) || at.After(time.Now().Add(claimLossPageBudget)) {
			t.Errorf("scan has no per-pass budget: deadline=%v", at)
		}
	case <-time.After(time.Second):
		t.Error("worker never started discovery")
	}
	cancel()
	waitClaimLossPages(t, s)
	if len(*pages) != 0 {
		t.Fatal("canceled scan delivered a page")
	}
	s.claimLossPageMu.Lock()
	done := s.claimLossPageDone
	s.claimLossPageMu.Unlock()
	s.scheduleClaimLossPages(ctx)
	if s.claimLossPageDone != done {
		t.Fatal("canceled recovery scheduled another worker")
	}
}

type lossPageBlockingSNS struct {
	entered chan time.Time
}

func (p lossPageBlockingSNS) Publish(ctx context.Context, _ *sns.PublishInput, _ ...func(*sns.Options)) (*sns.PublishOutput, error) {
	deadline, _ := ctx.Deadline()
	p.entered <- deadline
	<-ctx.Done()
	return nil, ctx.Err()
}

func TestClaimLossPagesSinkTimeoutAndCancellation(t *testing.T) {
	s, raw, signer, _ := lossPageWorld(t)
	putLossPageRecord(t, raw, signer, lossPageRecord("blocked-sns", true, hostclaim.TimersUnknown))
	entered := make(chan time.Time, 1)
	alerting.Arm(&alerting.Sinks{TopicARN: "test", Pub: lossPageBlockingSNS{entered: entered}})
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	s.scheduleClaimLossPages(ctx)
	select {
	case deadline := <-entered:
		if deadline.IsZero() || time.Until(deadline) > alerting.PublishTimeout {
			t.Errorf("SNS sink has no timeout: %v", deadline)
		}
	case <-time.After(time.Second):
		t.Error("worker never called SNS")
	}
	cancel()
	waitClaimLossPages(t, s)
	var attempts int
	if err := s.store.db.QueryRow(`SELECT count(*) FROM claim_loss_pages`).Scan(&attempts); err != nil || attempts != 1 {
		t.Fatalf("canceled pass reserved another condition: %d %v", attempts, err)
	}
}

type lossPageSlowStore struct {
	release.Store
	reads   []string
	reverse bool
}

func (s *lossPageSlowStore) List(ctx context.Context, prefix string) ([]string, error) {
	keys, err := s.Store.List(ctx, prefix)
	if s.reverse {
		slices.Reverse(keys)
	}
	return keys, err
}

func (s *lossPageSlowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.reads = append(s.reads, key)
	if len(s.reads) > 2 {
		<-ctx.Done()
		return nil, "", ctx.Err()
	}
	timer := time.NewTimer(10 * time.Millisecond)
	defer timer.Stop()
	select {
	case <-timer.C:
		return s.Store.Get(ctx, key)
	case <-ctx.Done():
		return nil, "", ctx.Err()
	}
}

func TestClaimLossPagesScanProgressSurvivesRestartAndWraps(t *testing.T) {
	s, raw, signer, pages := lossPageWorld(t)
	for _, id := range []string{"b-retained", "c-retained", "d-retained", "e-retained", "f-retained", "z-pending"} {
		putLossPageRecord(t, raw, signer, lossPageRecord(id, id == "z-pending", hostclaim.TimersComplete))
	}
	slow := &lossPageSlowStore{Store: raw}
	s.cfg.Releases = slow
	pass := func(first string) {
		t.Helper()
		slow.reads = nil
		ctx, cancel := context.WithTimeout(t.Context(), 250*time.Millisecond)
		defer cancel()
		if err := s.pageClaimLossesAt(ctx, time.Now); !errors.Is(err, context.DeadlineExceeded) {
			t.Fatalf("slow scan: %v", err)
		}
		key, _ := hostclaim.LossRecordKey(first)
		if len(slow.reads) != 3 || slow.reads[0] != key {
			t.Fatalf("scan reads: %v; want first %s and one unfinished read", slow.reads, key)
		}
	}
	pass("b-retained")
	if len(*pages) != 0 {
		t.Fatalf("first scan reached beyond its budget: %v", *pages)
	}
	path := s.store.gate.path
	if err := s.store.Close(); err != nil {
		t.Fatal(err)
	}
	st, err := OpenGated(path)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { st.Close() })
	s = &Server{cfg: s.cfg, store: st}
	pass("d-retained")
	pass("f-retained")
	if len(*pages) != 1 || !strings.Contains((*pages)[0]["body"], "z-pending") {
		t.Fatalf("pending record not delivered within three scans: %v", *pages)
	}
	putLossPageRecord(t, raw, signer, lossPageRecord("a-new", true, hostclaim.TimersComplete))
	key, _ := hostclaim.LossRecordKey("z-pending")
	if err := raw.Delete(t.Context(), key); err != nil {
		t.Fatal(err)
	}
	slow.reverse = true
	pass("a-new")
	if len(*pages) != 2 || !strings.Contains((*pages)[1]["body"], "a-new") {
		t.Fatalf("scan did not wrap past the removed cursor to the new early record: %v", *pages)
	}
}

func TestClaimLossPagesScanRetriesUnfinishedPendingRead(t *testing.T) {
	s, raw, signer, pages := lossPageWorld(t)
	for _, id := range []string{"a-retained", "b-retained", "c-pending"} {
		putLossPageRecord(t, raw, signer, lossPageRecord(id, id == "c-pending", hostclaim.TimersComplete))
	}
	s.cfg.Releases = &lossPageSlowStore{Store: raw}
	ctx, cancel := context.WithTimeout(t.Context(), 250*time.Millisecond)
	defer cancel()
	if err := s.pageClaimLossesAt(ctx, time.Now); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("slow scan: %v", err)
	}
	if len(*pages) != 0 {
		t.Fatalf("unfinished read paged: %v", *pages)
	}
	slow := &lossPageSlowStore{Store: raw}
	s.cfg.Releases = slow
	ctx, cancel = context.WithTimeout(t.Context(), 250*time.Millisecond)
	defer cancel()
	if err := s.pageClaimLossesAt(ctx, time.Now); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("resumed scan: %v", err)
	}
	key, _ := hostclaim.LossRecordKey("c-pending")
	if !slices.Equal(slow.reads, []string{key, "control/host-loss/a-retained.json", "control/host-loss/b-retained.json"}) || len(*pages) != 1 {
		t.Fatalf("unfinished pending read skipped: reads=%v pages=%v", slow.reads, *pages)
	}
}

// lossPageStalledStore never answers a read of one key.
type lossPageStalledStore struct {
	release.Store
	stalled string
	reads   []string
}

func (s *lossPageStalledStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.reads = append(s.reads, key)
	if key == s.stalled {
		<-ctx.Done()
		return nil, "", ctx.Err()
	}
	return s.Store.Get(ctx, key)
}

func TestClaimLossPagesPersistentlyStalledReadDoesNotStarve(t *testing.T) {
	prior := claimLossReadBudget
	claimLossReadBudget = 50 * time.Millisecond
	t.Cleanup(func() { claimLossReadBudget = prior })
	s, raw, signer, pages := lossPageWorld(t)
	for _, id := range []string{"a-stalled", "b-pending"} {
		putLossPageRecord(t, raw, signer, lossPageRecord(id, true, hostclaim.TimersComplete))
	}
	stalledKey, _ := hostclaim.LossRecordKey("a-stalled")
	pendingKey, _ := hostclaim.LossRecordKey("b-pending")
	store := &lossPageStalledStore{Store: raw, stalled: stalledKey}
	s.cfg.Releases = store
	for lap := 0; lap < 2; lap++ {
		store.reads = nil
		ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
		err := s.pageClaimLossesAt(ctx, time.Now)
		cancel()
		if err == nil || !errors.Is(err, context.DeadlineExceeded) {
			t.Fatalf("lap %d: the stalled read should surface as a record error, got %v", lap, err)
		}
		if !slices.Contains(store.reads, pendingKey) {
			t.Fatalf("lap %d: the record behind the stalled one was never read: %v", lap, store.reads)
		}
		if !slices.Contains(store.reads, stalledKey) {
			t.Fatalf("lap %d: the stalled record was not retried: %v", lap, store.reads)
		}
	}
	if len(*pages) != 1 || !strings.Contains((*pages)[0]["body"], "b-pending") {
		t.Fatalf("pending record behind a stalled read was not paged exactly once: %v", *pages)
	}
}