package main

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/rolloutwords"
	"github.com/carlosframework/platform/internal/status"
)

const B = "b2"

func TestClassifyRollout(t *testing.T) {
	for _, tc := range []struct {
		name string
		r    apiRollout
		want rolloutState
	}{
		{"box has not installed B", apiRollout{Installed: "b1", Serving: "b1", Awake: true}, rolloutPending},
		{"idle scheduler, B not installed", apiRollout{Installed: "b1", Good: "b1", SchedulerIdle: true}, rolloutPending},
		{"swapped", apiRollout{Installed: B, Serving: B, Awake: true}, rolloutSwapped},
		{"asleep with B installed", apiRollout{Installed: B, Good: "b1"}, rolloutAsleep},
		// A process starting or stopping is neither asleep nor swapped.
		{"transient, idle, no serving build", apiRollout{Installed: B, Good: "b1", Transient: true, SchedulerIdle: true}, rolloutPending},
		{"transient with B recorded as serving", apiRollout{Installed: B, Serving: B, Awake: true, Transient: true, SchedulerIdle: true}, rolloutPending},
		{"transient but failed on B", apiRollout{Installed: B, Attempt: status.AttemptDown, AttemptBuild: B, HeldBy: B, Transient: true}, rolloutFailed},
		{"queued", apiRollout{Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptQueued, AttemptBuild: B}, rolloutPending},
		{"canary settling", apiRollout{Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}, rolloutPending},
		{"failed here", apiRollout{Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptFailed, AttemptBuild: B, HeldBy: B, Reason: status.ReasonHealth}, rolloutFailed},
		{"hold mode, down", apiRollout{Installed: B, Attempt: status.AttemptDown, AttemptBuild: B, HeldBy: B}, rolloutFailed},
		{"held by a sibling's failure", apiRollout{Installed: B, Serving: "b1", Awake: true, HeldBy: B}, rolloutHeld},
		{"asleep and held", apiRollout{Installed: B, Good: "b1", HeldBy: B}, rolloutHeld},
		{"an older build's failure is not B's", apiRollout{Installed: B, Serving: B, Awake: true, Attempt: status.AttemptFailed, AttemptBuild: "b0"}, rolloutSwapped},
		// R8: an open attempt is pending whatever awake says, and B serving
		// is not swapped until the canary's settle closes.
		{"R8 canary serving B, settle open", apiRollout{Installed: B, Serving: B, Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}, rolloutPending},
		{"R8 first wake in flight", apiRollout{Installed: B, Good: "b1", Attempt: status.AttemptSettling, AttemptBuild: B}, rolloutPending},
		{"R8 asleep, queued", apiRollout{Installed: B, Good: "b1", Attempt: status.AttemptQueued, AttemptBuild: B}, rolloutPending},
		{"R8 asleep, running", apiRollout{Installed: B, Good: "b1", Attempt: status.AttemptRunning, AttemptBuild: B}, rolloutPending},
		// R6: a box that cannot arm reports no per-host state: its awake
		// false is meaningless, so B installed there is not asleep.
		{"R6 refusing box, B installed", apiRollout{Box: "x1", Installed: B, Unavailable: "no-release-key"}, rolloutPending},
		{"R6 declined box, B installed", apiRollout{Box: "x1", Installed: B, Unavailable: "generations-off"}, rolloutPending},
	} {
		r := tc.r
		if got := classifyRollout(&r, B); got != tc.want {
			t.Errorf("%s: %s, want %s", tc.name, got, tc.want)
		}
	}
}

// sweep is one scripted answer of the instance list.
type sweep struct {
	recs       []apiInstanceRecord
	truncated  bool
	incomplete bool
	err        error
}

// scripted answers each list call with the next sweep, repeating the last.
func scripted(calls *atomic.Int64, steps ...sweep) func(time.Time) (watchList, error) {
	return func(time.Time) (watchList, error) {
		i := int(calls.Add(1)) - 1
		s := steps[min(i, len(steps)-1)]
		return watchList{recs: s.recs, truncated: s.truncated, incomplete: s.incomplete}, s.err
	}
}

// sweeps answers each list call with the next set, repeating the last; no
// list is truncated.
func sweeps(calls *atomic.Int64, sets ...[]apiInstanceRecord) func(time.Time) (watchList, error) {
	steps := make([]sweep, 0, len(sets))
	for _, s := range sets {
		steps = append(steps, sweep{recs: s})
	}
	return scripted(calls, steps...)
}

func rec(host string, r *apiRollout) apiInstanceRecord {
	return apiInstanceRecord{Host: host, Rollout: r}
}

func targetsFor(hosts ...string) []watchTarget {
	var out []watchTarget
	for _, h := range hosts {
		out = append(out, watchTarget{Host: h, Origin: "http://127.0.0.1:1"})
	}
	return out
}

// noURL fails the test if a host is polled over HTTP: a reported host must
// never be requested, because a request wakes an asleep instance.
func noURL(t *testing.T) func(watchTarget, int64) (bool, string) {
	var told sync.Map
	return func(tg watchTarget, _ int64) (bool, string) {
		if _, dup := told.LoadOrStore(tg.Host, true); !dup {
			t.Errorf("%s was polled over HTTP", tg.Host)
		}
		return false, ""
	}
}

// pollLog records which hosts were polled over HTTP and answers live as told.
type pollLog struct {
	mu     sync.Mutex
	polled []string
}

func (p *pollLog) check(live map[string]bool) func(watchTarget, int64) (bool, string) {
	return func(tg watchTarget, _ int64) (bool, string) {
		p.mu.Lock()
		defer p.mu.Unlock()
		p.polled = append(p.polled, tg.Host)
		if live[tg.Host] {
			return true, ""
		}
		return false, "still serving b1"
	}
}

func (p *pollLog) hosts() map[string]int {
	p.mu.Lock()
	defer p.mu.Unlock()
	out := map[string]int{}
	for _, h := range p.polled {
		out[h]++
	}
	return out
}

func testWatch(w io.Writer, list func(time.Time) (watchList, error), check func(watchTarget, int64) (bool, string)) rolloutWatch {
	return rolloutWatch{w: w, app: "shop", ver: B, channel: "edge", timeout: 300 * time.Millisecond,
		interval: 5 * time.Millisecond, list: list, check: check}
}

func TestWatchRolloutFollowsTheBoxToDone(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	list := sweeps(&calls,
		[]apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true, SchedulerIdle: true}),
			rec("b.test", &apiRollout{Box: "x1", Installed: "b1", Good: "b1", SchedulerIdle: true})},
		[]apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}),
			rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"})},
		[]apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
			rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})},
	)
	if err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test")); err != nil {
		t.Fatalf("watch: %v\n%s", err, out.String())
	}
	for _, want := range []string{
		fmt.Sprintf(watchIntro, B),
		watchWordSwapped + " https://a.test",
		watchWordInstalled + " https://b.test (" + watchAsleepTail + ")",
		fmt.Sprintf(watchBox, "x1", fmt.Sprintf(watchCountSwapped, 1)+", "+fmt.Sprintf(watchCountAsleep, 1)),
	} {
		if !strings.Contains(out.String(), want) {
			t.Errorf("output lacks %q:\n%s", want, out.String())
		}
	}
	if calls.Load() < 3 {
		t.Errorf("done after %d sweeps: an open settle must hold the box", calls.Load())
	}
}

// A host whose process is starting or stopping stays pending through a sweep,
// idle scheduler or not, and completes once it reports asleep or swapped.
func TestWatchRolloutATransientHostStaysPendingUntilItSettles(t *testing.T) {
	for _, tc := range []struct {
		name  string
		final apiRollout
		want  string
	}{
		{"asleep", apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true}, watchWordInstalled + " https://a.test (" + watchAsleepTail + ")"},
		{"swapped", apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}, watchWordSwapped + " https://a.test"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			var out bytes.Buffer
			var calls atomic.Int64
			list := sweeps(&calls,
				[]apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", Transient: true, SchedulerIdle: true})},
				[]apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", Transient: true, SchedulerIdle: true})},
				[]apiInstanceRecord{rec("a.test", &tc.final)},
			)
			if err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test")); err != nil {
				t.Fatalf("watch: %v\n%s", err, out.String())
			}
			if calls.Load() < 3 {
				t.Fatalf("done after %d sweeps, while the host was still starting or stopping", calls.Load())
			}
			if !strings.Contains(out.String(), tc.want) {
				t.Errorf("output lacks %q:\n%s", tc.want, out.String())
			}
		})
	}
}

// R8: a canary already serving B is not swapped while its settle is open.
func TestWatchRolloutCanaryServingBWaitsForItsSettle(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	list := sweeps(&calls,
		[]apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B, SchedulerIdle: false})},
		[]apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, Attempt: status.AttemptFailed, AttemptBuild: B, HeldBy: B, Reason: status.ReasonSettle, SchedulerIdle: true})},
	)
	err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test"))
	if err == nil || !strings.HasPrefix(err.Error(), fmt.Sprintf(watchFailedSummary, B, 1, 0, "shop", B, "edge")) {
		t.Fatalf("err = %v\n%s", err, out.String())
	}
	if strings.Contains(out.String(), watchWordSwapped) {
		t.Fatalf("a canary in its settle was counted swapped:\n%s", out.String())
	}
}

// A host that is final keeps being refreshed: here the sleeper is final
// (installed, asleep) from the first sweep, while its box's scheduler is
// still busy, and the box goes idle a sweep later. A loop that skips final
// hosts keeps the first report's SchedulerIdle=false for ever, so boxDone
// never holds and a finished deploy times out. The host prints once.
func TestWatchRolloutRefreshesAFinalHostUntilTheBoxIsDone(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	list := sweeps(&calls,
		[]apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"})},
		[]apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"})},
		[]apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})},
	)
	if err := testWatch(&out, list, noURL(t)).run(targetsFor("b.test")); err != nil {
		t.Fatalf("watch: %v\n%s", err, out.String())
	}
	line := watchWordInstalled + " https://b.test (" + watchAsleepTail + ")"
	if got := strings.Count(out.String(), line); got != 1 {
		t.Fatalf("%q printed %d times, want once:\n%s", line, got, out.String())
	}
	if calls.Load() < 3 {
		t.Fatalf("done after %d sweeps, before the box's scheduler went idle", calls.Load())
	}
}

// Review Focus 2.
func TestWatchRolloutNoOpDeployEndsAtOnce(t *testing.T) {
	var calls atomic.Int64
	list := sweeps(&calls, []apiInstanceRecord{
		rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
		rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: B, SchedulerIdle: true}),
	})
	if err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test")); err != nil {
		t.Fatal(err)
	}
	if calls.Load() != 1 {
		t.Fatalf("an already-converged deploy took %d sweeps, want 1", calls.Load())
	}
}

func TestWatchRolloutIdleSchedulerBeforeInstallIsPending(t *testing.T) {
	var calls atomic.Int64
	list := sweeps(&calls, []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true, SchedulerIdle: true})})
	err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
	if err == nil || !strings.Contains(err.Error(), fmt.Sprintf(watchTimeout, B)) || !strings.Contains(err.Error(), fmt.Sprintf(watchPendingOther, "b1")) {
		t.Fatalf("err = %v", err)
	}
}

// Review Focus 4: a newer build replaced the watched one.
func TestWatchRolloutNewerBuildNamesWhatIsInstalled(t *testing.T) {
	var calls atomic.Int64
	list := sweeps(&calls, []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true})})
	err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
	if err == nil || !strings.Contains(err.Error(), "a.test: "+fmt.Sprintf(watchPendingOther, "b3")) {
		t.Fatalf("err = %v", err)
	}
}

func TestWatchRolloutOpenSettleHoldsTheWatch(t *testing.T) {
	var calls atomic.Int64
	list := sweeps(&calls, []apiInstanceRecord{
		rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true}),
		rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"}),
	})
	err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test"))
	if err == nil || !strings.Contains(err.Error(), fmt.Sprintf(watchBoxBusy, "x1")) {
		t.Fatalf("err = %v", err)
	}
}

// R7: the host the build failed on is failed; a sibling its record holds
// back is held.
func TestWatchRolloutReportsFailureAndNamesPromoteRetry(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	list := sweeps(&calls, []apiInstanceRecord{
		rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptFailed, AttemptBuild: B, HeldBy: B, Reason: status.ReasonHealth, SchedulerIdle: true}),
		rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, HeldBy: B, SchedulerIdle: true}),
		rec("c.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", HeldBy: B, Reason: status.ReasonHealth, SchedulerIdle: true}),
	})
	err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test", "c.test"))
	if err == nil || err.Error() != fmt.Sprintf(watchFailedSummary, B, 1, 2, "shop", B, "edge") {
		t.Fatalf("err = %v", err)
	}
	for _, want := range []string{
		watchWordFailed + " https://a.test: " + rolloutwords.Reason(status.ReasonHealth),
		watchWordHeld + " https://b.test: " + fmt.Sprintf(watchHeldTail, "b1", B),
		watchWordHeld + " https://c.test: " + fmt.Sprintf(watchHeldTail, "b1", B),
		fmt.Sprintf(watchBox, "x1", fmt.Sprintf(watchCountFailed, 1)+", "+fmt.Sprintf(watchCountHeld, 2)),
	} {
		if !strings.Contains(out.String(), want) {
			t.Errorf("output lacks %q:\n%s", want, out.String())
		}
	}
	if strings.Contains(err.Error(), watchHoldDownSummary[:20]) {
		t.Errorf("an overlap failure named the hold-down rollback: %v", err)
	}
}

// R7: a hold failure (attempt down for B) is failed, and the summary adds
// the hold-down sentence.
func TestWatchRolloutHoldDownNamesRollback(t *testing.T) {
	var calls atomic.Int64
	list := sweeps(&calls, []apiInstanceRecord{
		rec("a.test", &apiRollout{Box: "x1", Installed: B, Attempt: status.AttemptDown, AttemptBuild: B, HeldBy: B, Reason: status.ReasonExit, SchedulerIdle: true}),
		rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, HeldBy: B, SchedulerIdle: true}),
	})
	err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test"))
	want := fmt.Sprintf(watchFailedSummary, B, 1, 1, "shop", B, "edge") + "\n" + fmt.Sprintf(watchHoldDownSummary, "shop")
	if err == nil || err.Error() != want {
		t.Fatalf("err = %v\nwant %s", err, want)
	}
}

// Review Focus 3 and R1: a box that stops reporting never falls back to URL
// polling.
func TestWatchRolloutDarkBoxNeverPollsOverHTTP(t *testing.T) {
	var calls atomic.Int64
	list := sweeps(&calls,
		[]apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true})},
		[]apiInstanceRecord{rec("a.test", nil)},
	)
	err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
	if err == nil || !strings.Contains(err.Error(), "a.test: "+watchPendingDark) {
		t.Fatalf("err = %v", err)
	}
}

// A final state survives a missing observation, never an explicit later
// report (Astra plan 6 R1 #3; Review Focus 4). a.test swaps and its box goes
// dark: it stays swapped. d.test swaps and its box then reports b3 installed
// and serving: B no longer runs there, so d.test is pending again, named by
// what is installed, and the watch must not succeed for B once x2's canary
// settles.
func TestWatchRolloutFinalSurvivesADarkBoxButNotANewerBuild(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	list := sweeps(&calls,
		[]apiInstanceRecord{
			rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
			rec("d.test", &apiRollout{Box: "x3", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
			rec("b.test", &apiRollout{Box: "x2", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}),
		},
		[]apiInstanceRecord{
			rec("a.test", nil),
			rec("d.test", &apiRollout{Box: "x3", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true}),
			rec("b.test", &apiRollout{Box: "x2", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}),
		},
		[]apiInstanceRecord{
			rec("a.test", nil),
			rec("d.test", &apiRollout{Box: "x3", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true}),
			rec("b.test", &apiRollout{Box: "x2", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
		},
	)
	err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test", "d.test"))
	if err == nil {
		t.Fatalf("the watch succeeded for %s though d.test's box reports b3:\n%s", B, out.String())
	}
	if !strings.Contains(err.Error(), "\n  d.test: "+fmt.Sprintf(watchPendingOther, "b3")) {
		t.Errorf("timeout does not name d.test by what is installed: %v", err)
	}
	for _, h := range []string{"a.test", "b.test"} {
		if strings.Contains(err.Error(), "\n  "+h+": ") {
			t.Errorf("timeout names %s, which is final: %v", h, err)
		}
		if got := strings.Count(out.String(), watchWordSwapped+" https://"+h+"\n"); got != 1 {
			t.Errorf("%s printed swapped %d times:\n%s", h, got, out.String())
		}
	}
}

// A host watched by URL that a later report follows is reclassified by the
// report (Astra plan 6 R1 #3). c.test answers live by URL while b.test's
// canary settles; the next list reports c.test's box with b3 installed, so
// the watch must not count c.test done for B.
func TestWatchRolloutLaterReportReclassifiesAURLLiveHost(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	var pl pollLog
	settling := rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B})
	list := sweeps(&calls,
		[]apiInstanceRecord{settling, rec("c.test", nil)},
		[]apiInstanceRecord{settling, rec("c.test", &apiRollout{Box: "x2", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true})},
		[]apiInstanceRecord{
			rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
			rec("c.test", &apiRollout{Box: "x2", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true}),
		},
	)
	err := testWatch(&out, list, pl.check(map[string]bool{"c.test": true})).run(targetsFor("b.test", "c.test"))
	if err == nil || !strings.Contains(err.Error(), "\n  c.test: "+fmt.Sprintf(watchPendingOther, "b3")) {
		t.Fatalf("err = %v\n%s", err, out.String())
	}
	if got := pl.hosts()["c.test"]; got != 1 {
		t.Errorf("c.test polled %d times, want once (before its report)", got)
	}
}

// R3: absence from a truncated list keeps a host where it was. b.test is
// missing from the first (truncated) sweep and the list then fails once;
// it is never polled over HTTP, and its report, when it comes, decides it.
// a.test is missing from a later truncated sweep and keeps its state.
func TestWatchRolloutTruncatedListNeverSwitchesAHostToURL(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	busyA := rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B})
	list := scripted(&calls,
		sweep{recs: []apiInstanceRecord{busyA}, truncated: true},
		sweep{err: errors.New("instances list: 429")},
		sweep{recs: []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x2", Installed: B, Good: "b1", SchedulerIdle: true})}, truncated: true},
		sweep{recs: []apiInstanceRecord{
			rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
			rec("b.test", &apiRollout{Box: "x2", Installed: B, Good: "b1", SchedulerIdle: true}),
		}},
	)
	if err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test")); err != nil {
		t.Fatalf("watch: %v\n%s", err, out.String())
	}
	if calls.Load() < 4 {
		t.Fatalf("done after %d sweeps: a.test was still settling when the list cut it off", calls.Load())
	}
}

// R3: a host no truncated list ever names stays unknown, and the timeout
// says why.
func TestWatchRolloutAlwaysTruncatedNamesTheUnlistedHost(t *testing.T) {
	var calls atomic.Int64
	list := scripted(&calls, sweep{recs: []apiInstanceRecord{
		rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
	}, truncated: true})
	err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test"))
	if err == nil || !strings.Contains(err.Error(), "b.test: "+watchPendingUnlisted) {
		t.Fatalf("err = %v", err)
	}
}

// R4: a host its box did not report on at first (created since the box's
// last snapshot) is watched by URL until a report comes, then by the report.
func TestWatchRolloutALateReportMovesAHostOffURL(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	var pl pollLog
	list := sweeps(&calls,
		[]apiInstanceRecord{
			rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
			rec("c.test", nil),
		},
		[]apiInstanceRecord{
			rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
			rec("c.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true}),
		},
	)
	if err := testWatch(&out, list, pl.check(nil)).run(targetsFor("a.test", "c.test")); err != nil {
		t.Fatalf("watch: %v\n%s", err, out.String())
	}
	if got := pl.hosts(); got["c.test"] != 1 || got["a.test"] != 0 {
		t.Errorf("polled %v, want c.test once (before its report), a.test never", got)
	}
	if !strings.Contains(out.String(), watchWordInstalled+" https://c.test ("+watchAsleepTail+")") {
		t.Errorf("c.test not followed through its report:\n%s", out.String())
	}
}

func TestWatchRolloutMixedFleetPollsOnlyTheUnreported(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	var pl pollLog
	list := sweeps(&calls, []apiInstanceRecord{
		rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
		rec("c.test", nil),
	})
	rw := testWatch(&out, list, pl.check(map[string]bool{"c.test": true}))
	onURL := 0
	rw.onURL = func() { onURL++ }
	if err := rw.run(targetsFor("a.test", "c.test")); err != nil {
		t.Fatal(err)
	}
	if got := pl.hosts(); len(got) != 1 || got["c.test"] != 1 || onURL != 1 || !strings.Contains(out.String(), "live https://c.test") {
		t.Fatalf("polled %v, onURL %d, output:\n%s", got, onURL, out.String())
	}
}

// R6: a box that refuses every managed start holds its instances pending,
// never polled, and the timeout names why in the box's words.
func TestWatchRolloutRefusingBoxIsPendingWithItsReason(t *testing.T) {
	var calls atomic.Int64
	list := sweeps(&calls, []apiInstanceRecord{
		rec("a.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "no-release-key"}),
	})
	err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
	if err == nil || !strings.Contains(err.Error(), "a.test: "+rolloutwords.Unavailable("no-release-key")) {
		t.Fatalf("err = %v", err)
	}
}

// R6: a box that declined overlap runs today's stop-then-start paths, so its
// instances are watched by URL, like a flag-off box's.
func TestWatchRolloutDeclinedBoxIsWatchedByURL(t *testing.T) {
	var out bytes.Buffer
	var calls atomic.Int64
	var pl pollLog
	list := sweeps(&calls, []apiInstanceRecord{
		rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
		rec("c.test", &apiRollout{Box: "x2", Installed: B, Unavailable: "generations-off"}),
	})
	if err := testWatch(&out, list, pl.check(map[string]bool{"c.test": true})).run(targetsFor("a.test", "c.test")); err != nil {
		t.Fatalf("watch: %v\n%s", err, out.String())
	}
	if got := pl.hosts(); got["c.test"] != 1 || got["a.test"] != 0 || !strings.Contains(out.String(), "live https://c.test") {
		t.Fatalf("polled %v, output:\n%s", got, out.String())
	}
}

// R5: the instance list is read on its own cadence (5 s in production), not
// on every URL poll.
func TestWatchRolloutListsOnItsOwnCadence(t *testing.T) {
	var calls atomic.Int64
	var pl pollLog
	list := sweeps(&calls, []apiInstanceRecord{
		rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true, SchedulerIdle: true}),
		rec("c.test", nil),
	})
	rw := testWatch(io.Discard, list, pl.check(nil))
	rw.timeout, rw.listEvery = 80*time.Millisecond, time.Hour
	if err := rw.run(targetsFor("a.test", "c.test")); err == nil {
		t.Fatal("watch ended, want a timeout")
	}
	if calls.Load() != 1 || pl.hosts()["c.test"] < 3 {
		t.Fatalf("%d lists for %d URL polls: the list must keep its own cadence", calls.Load(), pl.hosts()["c.test"])
	}
}

func TestDeployWatchCadence(t *testing.T) {
	rw := newDeployWatch(nil, nil, apiAcct{Seg: "acme"}, "shop", B, "edge", "", time.Minute)
	if rw.listEvery != 5*time.Second || rw.interval != 2*time.Second {
		t.Fatalf("list every %s, poll every %s: want 5s and 2s", rw.listEvery, rw.interval)
	}
}

// R10: with no box reporting, the watch is today's watchVersion, byte for
// byte, its timeout report included.
func TestWatchRolloutWithNoReportsIsTheOldWatch(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("X-Carlos-Version", B)
	}))
	defer srv.Close()
	var out bytes.Buffer
	var calls atomic.Int64
	onURL := 0
	rw := testWatch(&out, sweeps(&calls, []apiInstanceRecord{rec("x.test", nil)}), nil)
	rw.onURL = func() { onURL++ }
	if err := rw.run([]watchTarget{{Host: "x.test", Origin: srv.URL}}); err != nil {
		t.Fatal(err)
	}
	if onURL != 1 || !strings.Contains(out.String(), "live https://x.test") || strings.Contains(out.String(), watchWordSwapped) {
		t.Fatalf("onURL %d, output:\n%s", onURL, out.String())
	}
}

func TestWatchRolloutWithNoReportsTimesOutAsTheOldWatch(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("X-Carlos-Version", "b1")
	}))
	defer srv.Close()
	targets := []watchTarget{{Host: "x.test", Origin: srv.URL}, {Host: "y.test", Origin: srv.URL}}
	diag := &watchDiagnostics{logsHint: "carlos logs --app shop"}
	var oldOut bytes.Buffer
	oldErr := watchVersion(&oldOut, targets, B, 30*time.Millisecond, 5*time.Millisecond, diag)
	for name, recs := range map[string][]apiInstanceRecord{
		"flag off": {rec("x.test", nil), rec("y.test", nil)},
		// R6: every box declined overlap.
		"declined": {rec("x.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "no-cgroup-delegation"}), rec("y.test", nil)},
		// A claims-on box declines overlap too (#272): URL-watched like any declined box.
		"claims-on": {rec("x.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "host-claims-on"}), rec("y.test", nil)},
		// A report on an instance the deploy does not target is not one.
		"untargeted report": {rec("x.test", nil), rec("y.test", nil), rec("z.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true})},
	} {
		var out bytes.Buffer
		var calls atomic.Int64
		rw := testWatch(&out, sweeps(&calls, recs), nil)
		rw.timeout, rw.diag = 30*time.Millisecond, diag
		err := rw.run(targets)
		if err == nil || oldErr == nil || err.Error() != oldErr.Error() || out.String() != oldOut.String() {
			t.Errorf("%s: err = %v\nwant %v\nout %q want %q", name, err, oldErr, out.String(), oldOut.String())
		}
	}
}

// R9: the until-about form only for a held_until in the future; otherwise
// today's line, byte for byte.
func TestDeployHeldLine(t *testing.T) {
	now := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC)
	today := "shop is a production app: the stable bake window applies — the fleet adopts when it expires; console fast-track if this is an emergency"
	future := now.Add(3 * time.Hour)
	for _, tc := range []struct {
		name, until, want string
	}{
		{"absent", "", today},
		{"past", now.Add(-time.Minute).Format(time.RFC3339), today},
		{"now", now.Format(time.RFC3339), today},
		{"unparseable", "soon", today},
		{"future", future.Format(time.RFC3339), fmt.Sprintf(deployHeldUntil, "shop", "stable", future.Local().Format("Mon 2 Jan 15:04 MST"))},
	} {
		if got := deployHeldLine("shop", "stable", tc.until, now); got != tc.want {
			t.Errorf("%s: %q\nwant %q", tc.name, got, tc.want)
		}
	}
}

// hitServer is an instance's origin that counts the HTTP requests reaching
// it, so a test sees a poll from either path: the rollout watch's check or a
// fallback to watchVersion, which dials the origin itself.
func hitServer(t *testing.T) (string, *atomic.Int64) {
	var hits atomic.Int64
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		hits.Add(1)
		w.Header().Set("X-Carlos-Version", B)
	}))
	t.Cleanup(srv.Close)
	return srv.URL, &hits
}

// Fix round 1: a failed first list is retried, not a reason to URL-poll a
// fleet whose boxes report (that wakes the sleeper here).
func TestWatchRolloutFirstListErrorIsRetried(t *testing.T) {
	origin, hits := hitServer(t)
	var calls atomic.Int64
	list := scripted(&calls,
		sweep{err: errors.New("instances list: 429")},
		sweep{recs: []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})}},
	)
	rw := testWatch(io.Discard, list, noURL(t))
	onURL := 0
	rw.onURL = func() { onURL++ }
	if err := rw.run([]watchTarget{{Host: "b.test", Origin: origin}}); err != nil {
		t.Fatal(err)
	}
	if hits.Load() != 0 || onURL != 0 || calls.Load() != 2 {
		t.Fatalf("%d HTTP polls, onURL %d, %d lists: want 0, 0, 2", hits.Load(), onURL, calls.Load())
	}
}

// Astra plan 6 R1 #4 (reverses fix round 1's three-tries fallback): failed
// lists are no evidence that the boxes do not report, so they are retried
// until a list answers. Here the fourth does, with a report: the sleeper is
// never requested.
func TestWatchRolloutFailedFirstListsNeverFallBack(t *testing.T) {
	origin, hits := hitServer(t)
	var calls atomic.Int64
	var out bytes.Buffer
	failed := sweep{err: &listStatusError{status: http.StatusServiceUnavailable, err: errors.New("instances list: 503")}}
	list := scripted(&calls, failed, failed, failed,
		sweep{recs: []apiInstanceRecord{rec("x.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})}},
	)
	rw := testWatch(&out, list, nil)
	onURL := 0
	rw.onURL = func() { onURL++ }
	if err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}}); err != nil {
		t.Fatalf("watch: %v\n%s", err, out.String())
	}
	if calls.Load() != 4 || onURL != 0 || hits.Load() != 0 || !strings.Contains(out.String(), watchWordInstalled+" https://x.test") {
		t.Fatalf("%d lists, onURL %d, %d HTTP polls, output:\n%s", calls.Load(), onURL, hits.Load(), out.String())
	}
}

// Astra plan 6 R1 #4: a list that keeps failing with a 5xx, a 429 or a
// network error is retried until the deadline. No origin is requested, and
// the timeout names the list failure.
func TestWatchRolloutListFailuresTimeOutNamingTheList(t *testing.T) {
	for name, lerr := range map[string]error{
		"503": &listStatusError{status: http.StatusServiceUnavailable, err: errors.New("could not list instances")},
		"429": &listStatusError{status: http.StatusTooManyRequests, err: errors.New("too many requests")},
		// Astra plan 6 R2 #4: the deploy just authenticated with this
		// token, so a 401 says nothing about the boxes.
		"401":     &listStatusError{status: http.StatusUnauthorized, err: errors.New("not signed in")},
		"network": errors.New("dial tcp: connection refused"),
	} {
		origin, hits := hitServer(t)
		var calls atomic.Int64
		rw := testWatch(io.Discard, scripted(&calls, sweep{err: lerr}), nil)
		onURL := 0
		rw.onURL = func() { onURL++ }
		err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}})
		if hits.Load() != 0 || onURL != 0 {
			t.Errorf("%s: %d HTTP polls, onURL %d: want none", name, hits.Load(), onURL)
		}
		want := fmt.Sprintf(watchTimeout, B) + "\n  " + watchListFailed
		if err == nil || !strings.HasPrefix(err.Error(), want) || !strings.Contains(err.Error(), lerr.Error()) {
			t.Errorf("%s: err = %v\nwant it to start %q and carry the list's error", name, err, want)
		}
		if calls.Load() < 2 {
			t.Errorf("%s: %d lists: a failed list must be retried", name, calls.Load())
		}
	}
}

// Astra plan 6 R1 #4, R2 #4: a 404 is a console with no list for the app,
// and a 403 a principal that may deploy but not read the list (a
// Publish-only token): either way the watch is today's, as before.
func TestWatchRolloutList404And403FallBack(t *testing.T) {
	for _, lerr := range []*listStatusError{
		{status: http.StatusNotFound, err: notFoundErr()},
		{status: http.StatusForbidden, err: errors.New("forbidden")},
	} {
		origin, hits := hitServer(t)
		var calls atomic.Int64
		var out bytes.Buffer
		rw := testWatch(&out, scripted(&calls, sweep{err: lerr}), nil)
		onURL := 0
		rw.onURL = func() { onURL++ }
		if err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}}); err != nil {
			t.Fatalf("%d: %v", lerr.status, err)
		}
		if calls.Load() != 1 || onURL != 1 || hits.Load() == 0 || out.String() != "live https://x.test\n" {
			t.Fatalf("%d: %d lists, onURL %d, %d HTTP polls, output %q", lerr.status, calls.Load(), onURL, hits.Load(), out.String())
		}
	}
}

// Astra plan 6 R1 #4: a list whose rollout join was incomplete is no
// evidence that a host has no report: it is not URL-polled, and the timeout
// says why its state is unknown.
func TestWatchRolloutIncompleteJoinIsNotNoReports(t *testing.T) {
	origin, hits := hitServer(t)
	var calls atomic.Int64
	rw := testWatch(io.Discard, scripted(&calls, sweep{recs: []apiInstanceRecord{rec("x.test", nil)}, incomplete: true}), nil)
	onURL := 0
	rw.onURL = func() { onURL++ }
	err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}})
	if hits.Load() != 0 || onURL != 0 {
		t.Fatalf("%d HTTP polls, onURL %d: want none", hits.Load(), onURL)
	}
	if err == nil || !strings.Contains(err.Error(), "\n  x.test: "+watchPendingUnread) {
		t.Fatalf("err = %v", err)
	}
}

// Astra plan 6 R1 #7: an app-wide truncated list on which every target is
// listed with no report is the old watch, byte for byte, on success and on
// timeout.
func TestWatchRolloutTruncatedListWithEveryTargetUnreportedIsTheOldWatch(t *testing.T) {
	for _, served := range []string{B, "b1"} {
		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			w.Header().Set("X-Carlos-Version", served)
		}))
		targets := []watchTarget{{Host: "x.test", Origin: srv.URL}, {Host: "y.test", Origin: srv.URL}}
		diag := &watchDiagnostics{logsHint: "carlos logs --app shop"}
		var oldOut bytes.Buffer
		oldErr := watchVersion(&oldOut, targets, B, 30*time.Millisecond, 5*time.Millisecond, diag)
		var out bytes.Buffer
		var calls atomic.Int64
		// The production check, so a watch that wrongly stays on the reports
		// path prints what it would print, rather than panicking.
		client := &http.Client{Timeout: time.Second}
		rw := testWatch(&out, scripted(&calls, sweep{recs: []apiInstanceRecord{rec("x.test", nil), rec("y.test", nil)}, truncated: true}),
			func(t watchTarget, n int64) (bool, string) { return checkTarget(client, t, B, n) })
		rw.timeout, rw.diag = 30*time.Millisecond, diag
		err := rw.run(targets)
		srv.Close()
		if fmt.Sprint(err) != fmt.Sprint(oldErr) || out.String() != oldOut.String() {
			t.Errorf("serving %s: err = %v\nwant %v\nout %q\nwant %q", served, err, oldErr, out.String(), oldOut.String())
		}
	}
}

// Astra plan 6 R1 #2: one deadline governs discovery. A list that keeps
// failing at the production cadence (5 s between lists) does not hold a
// 50 ms watch for ten seconds.
func TestWatchRolloutTinyTimeoutHonouredThroughDiscovery(t *testing.T) {
	var calls atomic.Int64
	rw := testWatch(io.Discard, scripted(&calls, sweep{err: &listStatusError{status: http.StatusBadGateway, err: errors.New("could not list instances")}}), nil)
	rw.timeout, rw.listEvery, rw.interval = 50*time.Millisecond, deployWatchListEvery, deployWatchPollEvery
	start := time.Now()
	err := rw.run(targetsFor("x.test"))
	if took := time.Since(start); took > time.Second {
		t.Fatalf("a 50ms watch took %s through a failing list", took)
	}
	if err == nil || !strings.Contains(err.Error(), watchListFailed) {
		t.Fatalf("err = %v", err)
	}
}

// Astra plan 6 R1 #2: the fallback gets what is left of the deadline, not
// the whole timeout again. The list takes 300 ms to answer 404 inside a
// 400 ms watch; today's watch then has about 100 ms, not 400.
func TestWatchRolloutFallbackGetsTheRemainingTime(t *testing.T) {
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("X-Carlos-Version", "b1")
	}))
	defer srv.Close()
	rw := testWatch(io.Discard, func(time.Time) (watchList, error) {
		time.Sleep(300 * time.Millisecond)
		return watchList{}, &listStatusError{status: http.StatusNotFound, err: notFoundErr()}
	}, nil)
	rw.timeout = 400 * time.Millisecond
	start := time.Now()
	err := rw.run([]watchTarget{{Host: "x.test", Origin: srv.URL}})
	if took := time.Since(start); took > 600*time.Millisecond {
		t.Fatalf("a 400ms watch took %s: the fallback was given the whole timeout again", took)
	}
	if err == nil || !strings.Contains(err.Error(), "timed out waiting for x.test to serve "+B) {
		t.Fatalf("err = %v", err)
	}
}

// The production list reader, against a console: a 503 is retried to the
// deadline with no origin request; a 404 with no JSON error falls back; and
// rollout_incomplete is decoded, so an incomplete join is never read as no
// reports.
func TestDeployWatchReadsTheConsolesList(t *testing.T) {
	for _, tc := range []struct {
		name     string
		answer   func(w http.ResponseWriter)
		wantHits bool
		wantErr  string
	}{
		{"503", func(w http.ResponseWriter) {
			w.WriteHeader(http.StatusServiceUnavailable)
			io.WriteString(w, `{"error":"could not list instances"}`)
		}, false, watchListFailed + ": could not list instances"},
		{"404", func(w http.ResponseWriter) { w.WriteHeader(http.StatusNotFound) }, true, ""},
		{"401", func(w http.ResponseWriter) {
			w.WriteHeader(http.StatusUnauthorized)
			io.WriteString(w, `{"error":"not signed in"}`)
		}, false, watchListFailed + ": not signed in"},
		{"incomplete join", func(w http.ResponseWriter) {
			io.WriteString(w, `{"instances":[{"host":"x.test"}],"truncated":false,"rollout_incomplete":true}`)
		}, false, "x.test: " + watchPendingUnread},
	} {
		console := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			if !strings.HasSuffix(r.URL.Path, "/instances") {
				http.NotFound(w, r)
				return
			}
			tc.answer(w)
		}))
		origin, hits := hitServer(t)
		sess := &apiSession{c: newConsoleClient(console.URL, "tok"), console: console.URL}
		rw := newDeployWatch(context.Background(), sess, apiAcct{Seg: "acme"}, "shop", B, "edge", "", 300*time.Millisecond)
		rw.w, rw.onURL, rw.diag = io.Discard, nil, nil
		start := time.Now()
		err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}})
		took := time.Since(start)
		console.Close()
		if (hits.Load() > 0) != tc.wantHits {
			t.Errorf("%s: %d origin requests", tc.name, hits.Load())
		}
		if tc.wantErr == "" && err != nil || tc.wantErr != "" && (err == nil || !strings.Contains(err.Error(), tc.wantErr)) {
			t.Errorf("%s: err = %v, want %q", tc.name, err, tc.wantErr)
		}
		if took > 2*time.Second {
			t.Errorf("%s: a 300ms watch took %s", tc.name, took)
		}
	}
}

// Fix round 1 (R3 on the first sweep): a truncated first list with no
// report among the listed targets is no evidence that no box reports.
// c.test (listed, unreported) is watched by URL; b.test (cut off, then
// reported asleep) is never requested.
func TestWatchRolloutTruncatedFirstListDoesNotFallBack(t *testing.T) {
	bOrigin, bHits := hitServer(t)
	var calls atomic.Int64
	var pl pollLog
	list := scripted(&calls,
		sweep{recs: []apiInstanceRecord{rec("c.test", nil)}, truncated: true},
		sweep{recs: []apiInstanceRecord{
			rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true}),
			rec("c.test", nil),
		}},
	)
	rw := testWatch(io.Discard, list, pl.check(map[string]bool{"c.test": true}))
	targets := []watchTarget{{Host: "b.test", Origin: bOrigin}, {Host: "c.test", Origin: "http://127.0.0.1:1"}}
	if err := rw.run(targets); err != nil {
		t.Fatal(err)
	}
	if got := pl.hosts(); bHits.Load() != 0 || got["b.test"] != 0 || got["c.test"] != 1 {
		t.Fatalf("b.test origin hit %d times, check polled %v", bHits.Load(), got)
	}
}

// Fix round 1 (R6): a code this binary has no phrase for is a newer box's
// problem; its instances are followed and never requested.
func TestWatchRolloutUnknownUnavailableCodeIsNeverPolled(t *testing.T) {
	origin, hits := hitServer(t)
	var calls atomic.Int64
	list := sweeps(&calls, []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "future-code"})})
	err := testWatch(io.Discard, list, noURL(t)).run([]watchTarget{{Host: "a.test", Origin: origin}})
	if hits.Load() != 0 || err == nil || !strings.Contains(err.Error(), "a.test: "+rolloutwords.Unavailable("future-code")) {
		t.Fatalf("%d HTTP polls, err = %v", hits.Load(), err)
	}
}

// Astra plan 6 R3: a target missing from a complete, untruncated list is not
// evidence that it uses the old path (the console may simply not have been
// able to read its record). It is never polled, and the timeout names it.
func TestWatchRolloutNeverPollsATargetMissingFromACompleteList(t *testing.T) {
	var calls atomic.Int64
	settling := &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}
	list := scripted(&calls, sweep{recs: []apiInstanceRecord{rec("a.test", settling)}})
	rw := testWatch(io.Discard, list, noURL(t))
	onURL := 0
	rw.onURL = func() { onURL++ }
	err := rw.run(targetsFor("a.test", "b.test"))
	if onURL != 0 {
		t.Fatalf("onURL ran %d times: a missing host was watched by URL", onURL)
	}
	if err == nil || !strings.Contains(err.Error(), "\n  b.test: "+watchPendingUnread) {
		t.Fatalf("err = %v, want a timeout naming b.test", err)
	}
}

// A host first listed with no report on a complete join is URL-watched. If it
// then drops off the list (its record became unreadable), the watch must stop
// requesting it: the request would wake it if it is asleep.
func TestWatchRolloutStopsPollingAURLHostThatDropsOffTheList(t *testing.T) {
	var calls, polls, atDrop atomic.Int64
	settling := &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}
	list := func(time.Time) (watchList, error) {
		if calls.Add(1) == 1 {
			return watchList{recs: []apiInstanceRecord{rec("a.test", settling), rec("b.test", nil)}}, nil
		}
		atDrop.CompareAndSwap(0, polls.Load()+1) // +1 so a snapshot of zero polls is not mistaken for unset
		return watchList{recs: []apiInstanceRecord{rec("a.test", settling)}}, nil
	}
	check := func(tg watchTarget, _ int64) (bool, string) {
		if tg.Host != "b.test" {
			t.Errorf("%s was polled over HTTP", tg.Host)
		}
		polls.Add(1)
		return false, "not yet"
	}
	err := testWatch(io.Discard, list, check).run(targetsFor("a.test", "b.test"))
	if err == nil {
		t.Fatal("the watch succeeded; a was never final")
	}
	if atDrop.Load() < 2 {
		t.Fatalf("b was never polled before it dropped off the list (snapshot %d)", atDrop.Load())
	}
	if got, want := polls.Load(), atDrop.Load()-1; got != want {
		t.Fatalf("%d polls, %d when b dropped off the list: b was polled after it was gone", got, want)
	}
}