package edge

import (
	"bytes"
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/http/httptest"
	"os"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"golang.org/x/crypto/acme"
	"golang.org/x/crypto/acme/autocert"

	"github.com/carlosframework/platform/internal/release"
)

// certStoreGet is the two-value read the tests want: the Store's third
// return (the etag) is never interesting here.
func certStoreGet(t *testing.T, s release.Store, key string) ([]byte, error) {
	t.Helper()
	b, _, err := s.Get(context.Background(), key)
	return b, err
}

func TestCertKeyShape(t *testing.T) {
	// Pinned: Task 5's mirror lists this exact prefix and Task 6's door ACL
	// test names these exact keys. autocert's cache names go through
	// UNCHANGED — per-host PEM, the account key, HTTP-01 tokens — and so do
	// the wildcard manager's (wildcard.go), which is what lets one mirror of
	// one prefix rebuild a whole cache.
	for name, want := range map[string]string{
		"example.com":             "control/certs/example.com",
		"example.com+rsa":         "control/certs/example.com+rsa",
		"acme_account+key":        "control/certs/acme_account+key",
		"tok+http-01":             "control/certs/tok+http-01",
		"wildcard-go.tito.io.pem": "control/certs/wildcard-go.tito.io.pem",
		"wildcard-account.key":    "control/certs/wildcard-account.key",
	} {
		if got := certKey(name); got != want {
			t.Fatalf("certKey(%q) = %q, want %q", name, got, want)
		}
	}
}

func TestPublishingCacheMirrorsPutsToBucket(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	pc := &publishingCache{inner: autocert.DirCache(t.TempDir()), s: s}
	if err := pc.Put(context.Background(), "example.com", []byte("PEM")); err != nil {
		t.Fatal(err)
	}
	got, err := certStoreGet(t, s, "control/certs/example.com")
	if err != nil || string(got) != "PEM" {
		t.Fatalf("bucket copy = %q, %v", got, err)
	}
	// Local cache also holds it (autocert's own read path keeps working).
	if b, err := pc.Get(context.Background(), "example.com"); err != nil || string(b) != "PEM" {
		t.Fatalf("local copy = %q, %v", b, err)
	}
}

func TestPublishingCacheDeleteRemovesBucketCopy(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	pc := &publishingCache{inner: autocert.DirCache(t.TempDir()), s: s}
	_ = pc.Put(context.Background(), "tok+http-01", []byte("challenge"))
	if err := pc.Delete(context.Background(), "tok+http-01"); err != nil {
		t.Fatal(err)
	}
	if _, err := certStoreGet(t, s, "control/certs/tok+http-01"); err == nil {
		t.Fatal("token must be deleted from bucket after validation")
	}
	if _, err := pc.Get(context.Background(), "tok+http-01"); !errors.Is(err, autocert.ErrCacheMiss) {
		t.Fatalf("local copy after Delete: want ErrCacheMiss, got %v", err)
	}
}

// A renewal writes the SAME key again. A plain create-only Put would lose
// every renewal to a CAS conflict, so the publish is last-writer-wins.
func TestPublishingCacheReplacesAnExistingBucketCopy(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	pc := &publishingCache{inner: autocert.DirCache(t.TempDir()), s: s}
	if err := pc.Put(context.Background(), "example.com", []byte("OLD")); err != nil {
		t.Fatal(err)
	}
	if err := pc.Put(context.Background(), "example.com", []byte("NEW")); err != nil {
		t.Fatal(err)
	}
	got, err := certStoreGet(t, s, "control/certs/example.com")
	if err != nil || string(got) != "NEW" {
		t.Fatalf("bucket copy after renewal = %q, %v — the publish must replace, not create-only", got, err)
	}
}

// failingStore fails every write; reads are the real store's.
type failingStore struct {
	release.Store
	mu     sync.Mutex
	puts   int
	dels   int
	putErr error
}

func (f *failingStore) Put(context.Context, string, []byte, string) (string, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.puts++
	return "", f.putErr
}

func (f *failingStore) Delete(context.Context, string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.dels++
	return errors.New("bucket down")
}

// Local issuance must survive a bucket blip: the cert is on disk, the
// failure is loud, and autocert never sees an error (a returned error would
// abort an issuance that actually succeeded).
func TestPublishingCacheBucketFailureNeverFailsTheLocalOperation(t *testing.T) {
	var buf bytes.Buffer
	prev := log.Writer()
	log.SetOutput(&buf)
	t.Cleanup(func() { log.SetOutput(prev) })

	s := &failingStore{Store: release.NewDirStore(t.TempDir()), putErr: errors.New("bucket down")}
	dir := t.TempDir()
	pc := &publishingCache{inner: autocert.DirCache(dir), s: s}

	if err := pc.Put(context.Background(), "example.com", []byte("PEM")); err != nil {
		t.Fatalf("Put must not fail on a bucket error: %v", err)
	}
	if b, err := pc.Get(context.Background(), "example.com"); err != nil || string(b) != "PEM" {
		t.Fatalf("local copy = %q, %v — local durability comes first", b, err)
	}
	if err := pc.Delete(context.Background(), "example.com"); err != nil {
		t.Fatalf("Delete must not fail on a bucket error: %v", err)
	}
	if out := buf.String(); !strings.Contains(out, "certsync: bucket publish failed") ||
		!strings.Contains(out, "certsync: bucket delete failed") {
		t.Fatalf("a bucket failure must be loud; log was:\n%s", out)
	}
}

// The gate is the whole point: only the housekeeper publishes. A non-issuer
// keeps its local cache (it must still serve) and writes nothing.
func TestPublishingCacheNonIssuerPublishesNothing(t *testing.T) {
	s := &failingStore{Store: release.NewDirStore(t.TempDir())}
	pc := &publishingCache{
		inner:  autocert.DirCache(t.TempDir()),
		s:      s,
		issuer: func() (context.Context, bool) { return nil, false },
	}
	if err := pc.Put(context.Background(), "example.com", []byte("PEM")); err != nil {
		t.Fatal(err)
	}
	if err := pc.Delete(context.Background(), "example.com"); err != nil {
		t.Fatal(err)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.puts != 0 || s.dels != 0 {
		t.Fatalf("non-issuer touched the bucket: %d put(s), %d delete(s)", s.puts, s.dels)
	}
}

// hangingStore blocks every write until its context ends — a bucket that has
// stopped answering rather than one that refuses.
type hangingStore struct {
	release.Store
	mu      sync.Mutex
	entered int
}

func (h *hangingStore) enter() {
	h.mu.Lock()
	h.entered++
	h.mu.Unlock()
}

func (h *hangingStore) Get(ctx context.Context, _ string) ([]byte, string, error) {
	h.enter()
	<-ctx.Done()
	return nil, "", ctx.Err()
}

func (h *hangingStore) Put(ctx context.Context, _ string, _ []byte, _ string) (string, error) {
	h.enter()
	<-ctx.Done()
	return "", ctx.Err()
}

func (h *hangingStore) Delete(ctx context.Context, _ string) error {
	h.enter()
	<-ctx.Done()
	return ctx.Err()
}

// autocert calls Delete while holding challengeMu for write, and serves
// tls-alpn-01 handshakes and :80 challenge requests under that same mutex —
// so a publish that waits on a stalled bucket wedges the issuer's front
// door. Every bucket call gets its own budget.
func TestPublishingCacheBoundsHowLongABucketCanStall(t *testing.T) {
	prev := certPublishTimeout
	certPublishTimeout = 40 * time.Millisecond
	t.Cleanup(func() { certPublishTimeout = prev })
	log.SetOutput(io.Discard)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	s := &hangingStore{}
	dir := t.TempDir()
	pc := &publishingCache{inner: autocert.DirCache(dir), s: s}

	for _, step := range []struct {
		name string
		run  func() error
	}{
		{"Put", func() error { return pc.Put(context.Background(), "tok+http-01", []byte("challenge")) }},
		{"Delete", func() error { return pc.Delete(context.Background(), "tok+http-01") }},
	} {
		start := time.Now()
		if err := step.run(); err != nil {
			t.Fatalf("%s on a stalled bucket: %v", step.name, err)
		}
		if took := time.Since(start); took > 2*time.Second {
			t.Fatalf("%s waited %s on a stalled bucket — the publish must carry its own budget", step.name, took)
		}
	}
	if s.entered == 0 {
		t.Fatal("test bug: the bucket was never called")
	}
	// The local half completed both times, which is the half that matters.
	if _, err := pc.Get(context.Background(), "tok+http-01"); !errors.Is(err, autocert.ErrCacheMiss) {
		t.Fatalf("local delete did not happen: %v", err)
	}
}

// ctxAwareStore refuses work on a dead context, the way a real S3 client
// does and the local DirStore (which ignores its context) does not.
type ctxAwareStore struct{ release.Store }

func (c ctxAwareStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if err := ctx.Err(); err != nil {
		return nil, "", err
	}
	return c.Store.Get(ctx, key)
}

func (c ctxAwareStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if err := ctx.Err(); err != nil {
		return "", err
	}
	return c.Store.Put(ctx, key, body, ifMatch)
}

func (c ctxAwareStore) Delete(ctx context.Context, key string) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	return c.Store.Delete(ctx, key)
}

// A cancelled order still has to clean up its challenge token: the delete
// runs on a context detached from the caller's.
func TestPublishingCacheDeletesEvenUnderACancelledCaller(t *testing.T) {
	s := ctxAwareStore{release.NewDirStore(t.TempDir())}
	pc := &publishingCache{inner: autocert.DirCache(t.TempDir()), s: s}
	if err := pc.Put(context.Background(), "tok+http-01", []byte("challenge")); err != nil {
		t.Fatal(err)
	}
	dead, cancel := context.WithCancel(context.Background())
	cancel()
	_ = pc.Delete(dead, "tok+http-01")
	if _, err := certStoreGet(t, s, "control/certs/tok+http-01"); err == nil {
		t.Fatal("an abandoned order must still take its token out of the bucket")
	}
}

// ---- the issuance gate ------------------------------------------------------

func testIssuerEdge(hosts ...string) *Edge {
	e := &Edge{}
	m := map[string]upstream{}
	for _, h := range hosts {
		m[h] = upstream{}
	}
	e.upstreams.Store(&m)
	return e
}

// seedCachedCert writes a cert for host into cache in autocert's own entry
// shape (private key PEM, then the chain) — what a previous process, or Task
// 5's mirror, leaves on a box's disk.
func seedCachedCert(t *testing.T, cache autocert.Cache, parent, host string) {
	t.Helper()
	cert, key := selfSignedWildcard(t, parent, time.Now().Add(60*24*time.Hour))
	raw, err := encodeWildcardPEM(cert, key)
	if err != nil {
		t.Fatal(err)
	}
	if err := cache.Put(context.Background(), host, raw); err != nil {
		t.Fatal(err)
	}
}

// ecdsaHello is a handshake that supports ECDSA, so autocert looks under the
// bare host name rather than "<host>+rsa".
func ecdsaHello(host string) *tls.ClientHelloInfo {
	return &tls.ClientHelloInfo{
		ServerName:       host,
		SignatureSchemes: []tls.SignatureScheme{tls.ECDSAWithP256AndSHA256},
		SupportedCurves:  []tls.CurveID{tls.CurveP256},
		CipherSuites:     []uint16{tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256},
	}
}

// offlineManager is a real autocert.Manager pointed at a dead directory, so
// a test that accidentally reaches the ordering path fails offline and fast
// instead of dialling Let's Encrypt.
func offlineManager(t *testing.T, cache autocert.Cache, pol autocert.HostPolicy) *autocert.Manager {
	return &autocert.Manager{
		Prompt:     autocert.AcceptTOS,
		Cache:      cache,
		HostPolicy: pol,
		Client:     &acme.Client{DirectoryURL: deadDirectory(t)},
	}
}

// A box with no deployment store has nothing to publish to AND nothing to
// mirror from — fleet boxes (barred from control/certs/** by the door ACL),
// boxes with no bucket or dir, and bucket-without-pubkey boxes. Gating those
// would leave them permanently certificate-less, because autocert consults
// HostPolicy BEFORE its cache lookup: no orders and no serving either.
func TestNoStoreEdgeKeepsIssuingAndServingWithTheGateShut(t *testing.T) {
	const parent, host = "bab.oncarlos.com", "app.bab.oncarlos.com"
	dir := autocert.DirCache(t.TempDir())
	seedCachedCert(t, dir, parent, host)

	e := testIssuerEdge(host)
	e.issuer = func() (context.Context, bool) { return nil, false } // never holds

	cache := e.armCertIssuance(dir) // no staticStore
	if _, wrapped := cache.(*publishingCache); wrapped {
		t.Fatal("no deployment store: there is nothing to publish through")
	}
	if _, ok := e.issuerGate(); !ok {
		t.Fatal("the gate must be cleared, so the wildcard path is ungated too")
	}
	got, err := offlineManager(t, cache, e.issuerHostPolicy()).GetCertificate(ecdsaHello(host))
	if err != nil {
		t.Fatalf("a box with no mirror must keep serving its own cache: %v", err)
	}
	if got.Leaf == nil || got.Leaf.VerifyHostname(host) != nil {
		t.Fatalf("served cert does not cover %s", host)
	}
}

// The same edge WITH a store stays gated: the refusal is the gate's, and the
// cert it declines to serve is sitting right there on disk — which is the
// whole reason Task 5's mirror layer has to sit behind this.
func TestStoreBackedEdgeStaysGated(t *testing.T) {
	const parent, host = "bab.oncarlos.com", "app.bab.oncarlos.com"
	dir := autocert.DirCache(t.TempDir())
	seedCachedCert(t, dir, parent, host)

	e := testIssuerEdge(host)
	e.staticStore = release.NewDirStore(t.TempDir())
	e.issuer = func() (context.Context, bool) { return nil, false }

	cache := e.armCertIssuance(dir)
	if _, wrapped := cache.(*publishingCache); !wrapped {
		t.Fatal("a deployment store means every artifact publishes")
	}
	if _, err := cache.Get(context.Background(), host); err != nil {
		t.Fatalf("test bug: the local cert should still be readable: %v", err)
	}
	_, err := offlineManager(t, cache, e.issuerHostPolicy()).GetCertificate(ecdsaHello(host))
	if !errors.Is(err, errNotIssuer) {
		t.Fatalf("non-issuer with a mirror behind it = %v, want errNotIssuer", err)
	}
}

func TestHostPolicyUngatedIsTodaysBehaviour(t *testing.T) {
	e := testIssuerEdge("app.bab.oncarlos.com")
	pol := e.issuerHostPolicy()
	if err := pol(context.Background(), "app.bab.oncarlos.com"); err != nil {
		t.Fatalf("no gate configured (carlos edge, single box): want issuance allowed, got %v", err)
	}
	if err := pol(context.Background(), "stranger.example"); err == nil {
		t.Fatal("unknown host must still be refused")
	} else if errors.Is(err, errNotIssuer) {
		t.Fatalf("unknown host must be refused as unknown, not as not-issuer: %v", err)
	}
}

func TestHostPolicyRefusesOrdersOnANonIssuer(t *testing.T) {
	e := testIssuerEdge("app.bab.oncarlos.com")
	e.issuer = func() (context.Context, bool) { return nil, false }
	err := e.issuerHostPolicy()(context.Background(), "app.bab.oncarlos.com")
	if !errors.Is(err, errNotIssuer) {
		t.Fatalf("non-issuer HostPolicy = %v, want errNotIssuer", err)
	}
}

func TestHostPolicyAllowsOrdersWhileHolding(t *testing.T) {
	e := testIssuerEdge("app.bab.oncarlos.com")
	e.issuer = func() (context.Context, bool) { return context.Background(), true }
	if err := e.issuerHostPolicy()(context.Background(), "app.bab.oncarlos.com"); err != nil {
		t.Fatalf("issuer HostPolicy = %v, want nil", err)
	}
}

// ---- the wildcard (DNS-01) gate --------------------------------------------

// countingCache counts Gets per key — the wildcard order path's first cache
// read is its ACME account key, so "did an order start?" is observable
// without a CA.
type countingCache struct {
	autocert.Cache
	mu   sync.Mutex
	gets map[string]int
}

func newCountingCache(t *testing.T) *countingCache {
	return &countingCache{Cache: autocert.DirCache(t.TempDir()), gets: map[string]int{}}
}

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

func (c *countingCache) count(key string) int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.gets[key]
}

// deadDirectory is an ACME directory URL nothing is listening on, so an
// order that DOES start fails immediately and offline.
func deadDirectory(t *testing.T) string {
	srv := httptest.NewServer(nil)
	url := srv.URL + "/directory"
	srv.Close()
	return url
}

func wildcardGateManager(t *testing.T, cache autocert.Cache) *wildcardManager {
	w := newWildcardManager([]string{"go.tito.io"}, "", func() []string { return nil }, nil, "", cache, &fakeDNS{})
	w.directory = deadDirectory(t)
	w.refresh()
	return w
}

func TestWildcardPassDoesNotOrderOnANonIssuer(t *testing.T) {
	cache := newCountingCache(t)
	w := wildcardGateManager(t, cache)
	w.issuer = func() (context.Context, bool) { return nil, false }

	// A stood-down pass comes back within the hour: the next one is what
	// loads the issuer's wildcard out of this box's cache and into memory.
	if delay := w.pass(); delay > time.Hour {
		t.Fatalf("non-issuer pass delay = %s, want <= 1h", delay)
	}

	if n := cache.count(wildcardAccountKeyName); n != 0 {
		t.Fatalf("a non-issuer started an ACME order (%d account-key read(s))", n)
	}
	if w.backingOff("go.tito.io", time.Now()) {
		t.Fatal("standing down is not a failure: it must not arm the per-parent backoff")
	}
}

func TestWildcardPassOrdersWhileHolding(t *testing.T) {
	cache := newCountingCache(t)
	w := wildcardGateManager(t, cache)
	w.issuer = func() (context.Context, bool) { return context.Background(), true }

	w.pass()

	if n := cache.count(wildcardAccountKeyName); n == 0 {
		t.Fatal("the issuer must attempt the order (no account-key read happened)")
	}
	if !w.backingOff("go.tito.io", time.Now()) {
		t.Fatal("a failed order must arm the per-parent backoff")
	}
}

// The order runs under the HOLDING's context, so a stand-down mid-order is
// observable inside the order rather than at the next 12h tick.
func TestWildcardOrderRunsUnderTheHoldingContext(t *testing.T) {
	cache := newCountingCache(t)
	w := wildcardGateManager(t, cache)
	held, standDown := context.WithCancel(context.Background())
	w.issuer = func() (context.Context, bool) { return held, true }
	standDown()

	w.pass()

	// The account key read is the first thing obtain does; with the holding
	// already cancelled the order must not have reached the CA. The cache is
	// context-free, so what this pins is that the cancelled context is the
	// one the order carries: obtain fails before any network call.
	if err := held.Err(); err == nil {
		t.Fatal("test bug: the holding should be cancelled")
	}
	w.mu.Lock()
	got := w.certs["go.tito.io"]
	w.mu.Unlock()
	if got != nil {
		t.Fatal("an order under a cancelled holding must not produce a cert")
	}
}

func TestWildcardUngatedOrdersAsToday(t *testing.T) {
	cache := newCountingCache(t)
	w := wildcardGateManager(t, cache) // no issuer gate at all
	w.pass()
	if n := cache.count(wildcardAccountKeyName); n == 0 {
		t.Fatal("with no gate configured (carlos edge, single box) issuance must run exactly as today")
	}
}

// ---- the mirror (Task 5) ----------------------------------------------------

// hostCertPEM mints a cert for host in autocert's own cache-entry shape (the
// private key PEM, then the chain) — byte-for-byte what the issuer publishes
// to control/certs/<host>.
func hostCertPEM(t *testing.T, host string, notAfter time.Time) []byte {
	t.Helper()
	cert, key := selfSignedLeaf(t, host, notAfter)
	raw, err := encodeWildcardPEM(cert, key)
	if err != nil {
		t.Fatal(err)
	}
	return raw
}

// putBucket writes one control/certs/ entry the way the issuer's
// publishingCache would.
func putBucket(t *testing.T, s release.Store, name string, body []byte) {
	t.Helper()
	if err := putCert(context.Background(), s, name, body); err != nil {
		t.Fatal(err)
	}
}

func TestCertMirrorServesBucketCert(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, "example.com", []byte("PEM"))
	m := newCertMirror(s, nil)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM" {
		t.Fatalf("mirror cert = %q, %v", b, ok)
	}
	if _, ok := m.Cert("nobody.example"); ok {
		t.Fatal("a name the bucket does not hold must miss")
	}
}

// The HTTP-01 cache key is autocert's, read from the vendored source
// (autocert.go:896): path.Base(tokenPath) + "+http-01" — a FLAT name, no
// /.well-known/ prefix (publishableCertName would refuse one anyway). Task 7
// fails loudly if this drifts.
func TestCertMirrorServesChallengeTokens(t *testing.T) {
	const token = "0123456789abcdefghij0123456789abcdefghijkl"
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, token+"+http-01", []byte("resp"))
	m := newCertMirror(s, nil)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if b, ok := m.Challenge(token); !ok || string(b) != "resp" {
		t.Fatalf("challenge = %q, %v", b, ok)
	}
}

// Account keys are ISSUER material: they must never enter the serving map,
// where a name lookup could hand one to a handshake.
func TestCertMirrorKeepsAccountKeysOutOfTheServingMap(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, "acme_account+key", []byte("KEY"))
	putBucket(t, s, wildcardAccountKeyName, []byte("KEY"))
	putBucket(t, s, "example.com", []byte("PEM"))
	m := newCertMirror(s, nil)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	for _, name := range []string{"acme_account+key", wildcardAccountKeyName} {
		if _, ok := m.Cert(name); ok {
			t.Fatalf("%s is issuer material and must not be servable", name)
		}
	}
	if _, ok := m.Cert("example.com"); !ok {
		t.Fatal("ordinary cert material must still be served")
	}
}

// listFailingStore fails List (and optionally Get) on demand — a bucket blip
// mid-poll.
type listFailingStore struct {
	release.Store
	mu   sync.Mutex
	fail bool
}

func (l *listFailingStore) setFail(v bool) {
	l.mu.Lock()
	l.fail = v
	l.mu.Unlock()
}

func (l *listFailingStore) failing() bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.fail
}

func (l *listFailingStore) List(ctx context.Context, prefix string) ([]string, error) {
	if l.failing() {
		return nil, errors.New("bucket down")
	}
	return l.Store.List(ctx, prefix)
}

func (l *listFailingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if l.failing() {
		return nil, "", errors.New("bucket down")
	}
	return l.Store.Get(ctx, key)
}

// Degrade, don't die: a failed poll keeps the last-good map. The alternative
// (an empty map) turns a bucket blip into a TLS outage on every non-issuer.
func TestCertMirrorRefreshErrorKeepsLastGood(t *testing.T) {
	s := &listFailingStore{Store: release.NewDirStore(t.TempDir())}
	putBucket(t, s.Store, "example.com", []byte("PEM"))
	m := newCertMirror(s, nil)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	s.setFail(true)
	if err := m.Refresh(context.Background()); err == nil {
		t.Fatal("a failed poll must be reported to the loop, which logs it")
	}
	if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM" {
		t.Fatalf("after a failed refresh the mirror = %q, %v — want the last-good copy", b, ok)
	}
}

// A withdrawn entry (a spent challenge token the issuer deleted) leaves the
// map on the next good poll.
func TestCertMirrorRefreshDropsWithdrawnEntries(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, "example.com", []byte("PEM"))
	m := newCertMirror(s, nil)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if err := s.Delete(context.Background(), certKey("example.com")); err != nil {
		t.Fatal(err)
	}
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if _, ok := m.Cert("example.com"); ok {
		t.Fatal("a deleted bucket entry must leave the mirror")
	}
}

// ---- R3: the mirror feeds this box's own autocert cache ---------------------

// localCert reads a name back out of a local autocert cache.
func localCert(t *testing.T, c autocert.Cache, name string) ([]byte, error) {
	t.Helper()
	return c.Get(context.Background(), name)
}

// autocert's renewal timers never consult HostPolicy, and renewal.go's do()
// checks the CACHE first — so a mirror-fed local cache is what stops a
// stood-down box's stale timers from ordering. It also carries a box through
// the boot window before the election has been decided.
func TestCertMirrorWritesFreshCertsIntoTheLocalCache(t *testing.T) {
	const host = "app.bab.oncarlos.com"
	s := release.NewDirStore(t.TempDir())
	fresh := hostCertPEM(t, host, time.Now().Add(80*24*time.Hour))
	putBucket(t, s, host, fresh)
	putBucket(t, s, "wildcard-bab.oncarlos.com.pem", hostCertPEM(t, "*.bab.oncarlos.com", time.Now().Add(80*24*time.Hour)))

	local := autocert.DirCache(t.TempDir())
	m := newCertMirror(s, local)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	got, err := localCert(t, local, host)
	if err != nil || !bytes.Equal(got, fresh) {
		t.Fatalf("local cache after refresh = %d byte(s), %v — want the mirrored cert", len(got), err)
	}
	// The wildcard PEM lands under wildcard.go's own cache name, which is what
	// the wildcard manager's loadCached reads.
	if _, err := localCert(t, local, "wildcard-bab.oncarlos.com.pem"); err != nil {
		t.Fatalf("wildcard PEM did not reach the local cache: %v", err)
	}
}

// A renewal in the bucket replaces an older local copy; a local copy that is
// already newer is left alone (this box may be the issuer, one poll ahead of
// what it published).
func TestCertMirrorLocalWriteRespectsFreshness(t *testing.T) {
	const host = "app.bab.oncarlos.com"
	s := release.NewDirStore(t.TempDir())
	local := autocert.DirCache(t.TempDir())

	old := hostCertPEM(t, host, time.Now().Add(10*24*time.Hour))
	newer := hostCertPEM(t, host, time.Now().Add(80*24*time.Hour))

	// Local holds the OLD one; the bucket holds a renewal.
	if err := local.Put(context.Background(), host, old); err != nil {
		t.Fatal(err)
	}
	putBucket(t, s, host, newer)
	m := newCertMirror(s, local)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if got, _ := localCert(t, local, host); !bytes.Equal(got, newer) {
		t.Fatal("a renewal in the bucket must replace the older local copy")
	}

	// Now the other way round: local is fresher than the bucket.
	local2 := autocert.DirCache(t.TempDir())
	if err := local2.Put(context.Background(), host, newer); err != nil {
		t.Fatal(err)
	}
	s2 := release.NewDirStore(t.TempDir())
	putBucket(t, s2, host, old)
	m2 := newCertMirror(s2, local2)
	if err := m2.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if got, _ := localCert(t, local2, host); !bytes.Equal(got, newer) {
		t.Fatal("a local cert that is already fresher must not be downgraded by the mirror")
	}
}

// Account keys reach the local cache for issuer-takeover warmth (one ACME
// account per deployment, one rate-limit bucket) — but only when this box has
// none: clobbering a registered account key is never an improvement.
func TestCertMirrorSeedsAccountKeysOnlyWhenAbsent(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, "acme_account+key", []byte("BUCKET-KEY"))
	putBucket(t, s, wildcardAccountKeyName, []byte("BUCKET-WILDCARD-KEY"))

	local := autocert.DirCache(t.TempDir())
	m := newCertMirror(s, local)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if got, err := localCert(t, local, "acme_account+key"); err != nil || string(got) != "BUCKET-KEY" {
		t.Fatalf("account key not seeded: %q, %v", got, err)
	}

	local2 := autocert.DirCache(t.TempDir())
	if err := local2.Put(context.Background(), "acme_account+key", []byte("MINE")); err != nil {
		t.Fatal(err)
	}
	m2 := newCertMirror(s, local2)
	if err := m2.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if got, _ := localCert(t, local2, "acme_account+key"); string(got) != "MINE" {
		t.Fatalf("this box's own account key was clobbered: %q", got)
	}
}

// Challenge tokens are ephemeral and the issuer owns their deletion: copying
// them onto every box's disk would litter caches nothing ever sweeps.
func TestCertMirrorNeverWritesChallengeTokensLocally(t *testing.T) {
	const token = "0123456789abcdefghij0123456789abcdefghijkl"
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, token+"+http-01", []byte("resp"))
	local := autocert.DirCache(t.TempDir())
	m := newCertMirror(s, local)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if _, err := localCert(t, local, token+"+http-01"); !errors.Is(err, autocert.ErrCacheMiss) {
		t.Fatalf("a challenge token must not land in the local cache: %v", err)
	}
}

// A steady state costs no local I/O: an unchanged entry is not rewritten.
func TestCertMirrorSkipsUnchangedLocalWrites(t *testing.T) {
	const host = "app.bab.oncarlos.com"
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, host, hostCertPEM(t, host, time.Now().Add(80*24*time.Hour)))
	local := &countingWriteCache{Cache: autocert.DirCache(t.TempDir())}
	m := newCertMirror(s, local)
	for range 3 {
		if err := m.Refresh(context.Background()); err != nil {
			t.Fatal(err)
		}
	}
	if n := local.puts(); n != 1 {
		t.Fatalf("local writes = %d, want exactly 1 — an unchanged mirror must not rewrite the cache", n)
	}
}

// countingWriteCache counts Puts.
type countingWriteCache struct {
	autocert.Cache
	mu sync.Mutex
	n  int
}

func (c *countingWriteCache) Put(ctx context.Context, key string, data []byte) error {
	c.mu.Lock()
	c.n++
	c.mu.Unlock()
	return c.Cache.Put(ctx, key, data)
}

func (c *countingWriteCache) puts() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.n
}

// ---- R4: a wildcard takeover propagates on the poll -------------------------

func TestCertMirrorWakesTheWildcardManagerOnChangedWildcardMaterial(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, "app.bab.oncarlos.com", hostCertPEM(t, "app.bab.oncarlos.com", time.Now().Add(80*24*time.Hour)))
	local := autocert.DirCache(t.TempDir())
	m := newCertMirror(s, local)
	var wakes int
	m.wake = func() { wakes++ }

	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if wakes != 0 {
		t.Fatalf("a per-host cert is not wildcard material: %d wake(s)", wakes)
	}

	putBucket(t, s, "wildcard-bab.oncarlos.com.pem", hostCertPEM(t, "*.bab.oncarlos.com", time.Now().Add(80*24*time.Hour)))
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if wakes != 1 {
		t.Fatalf("a new wildcard must wake the manager: %d wake(s)", wakes)
	}
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if wakes != 1 {
		t.Fatalf("an unchanged poll must not wake the manager: %d wake(s)", wakes)
	}
	putBucket(t, s, "wildcard-bab.oncarlos.com.pem", hostCertPEM(t, "*.bab.oncarlos.com", time.Now().Add(85*24*time.Hour)))
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if wakes != 2 {
		t.Fatalf("a renewed wildcard must wake the manager: %d wake(s)", wakes)
	}
}

// ---- R1: the serving chain --------------------------------------------------

const mirrorHost = "app.bab.oncarlos.com"

// gatedChain builds what RunWithActivatorReady builds for a NON-ISSUER: a
// gated autocert manager with m in front of it and autocert's own cache
// (counted, so "did autocert serve this?" is observable) behind it.
//
// The mirror is passed in rather than built here so that a test can give it
// NO local cache: the mirror layer and the last-resort local-cache layer both
// serve the same bytes, and a test that means to pin one must not be able to
// pass on the other.
func gatedChain(t *testing.T, m *certMirror, local autocert.Cache, hosts ...string) (func(*tls.ClientHelloInfo) (*tls.Certificate, error), *countingCache) {
	t.Helper()
	return gatedChainFor(t, testIssuerEdge(hosts...), m, local)
}

// gatedChainFor is gatedChain over an edge the caller built — the form a test
// needs when the edge carries something beyond its route table, such as the
// mirrored home map the cert gates now consult (deploymentHost, #277).
func gatedChainFor(t *testing.T, e *Edge, m *certMirror, local autocert.Cache) (func(*tls.ClientHelloInfo) (*tls.Certificate, error), *countingCache) {
	t.Helper()
	e.issuer = func() (context.Context, bool) { return nil, false } // never the issuer
	counting := &countingCache{Cache: local, gets: map[string]int{}}
	mgr := offlineManager(t, counting, e.issuerHostPolicy())
	return m.wrapGetCertificate(mgr.GetCertificate), counting
}

// The whole point of Task 5: a gated box serves the issuer's cert, and does
// so BEFORE autocert — which would refuse the name outright.
func TestMirrorServesTheGatedNonIssuersHandshake(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, mirrorHost, hostCertPEM(t, mirrorHost, time.Now().Add(80*24*time.Hour)))
	// No local cache on the mirror: only the mirror layer itself can answer.
	m := newCertMirror(s, nil)
	get, counting := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	got, err := get(ecdsaHello(mirrorHost))
	if err != nil {
		t.Fatalf("a non-issuer must serve the mirrored cert: %v", err)
	}
	if got.Leaf == nil || got.Leaf.VerifyHostname(mirrorHost) != nil {
		t.Fatalf("served cert does not cover %s", mirrorHost)
	}
	if n := counting.count(mirrorHost); n != 0 {
		t.Fatalf("autocert was consulted %d time(s) — a valid mirror cert must short-circuit before it", n)
	}
}

// Each layer's miss falls through cleanly. With no gate (a single-box
// deployment) and an empty mirror, autocert serves from its own cache exactly
// as it does today.
func TestMirrorFallsThroughToAutocert(t *testing.T) {
	local := autocert.DirCache(t.TempDir())
	if err := local.Put(context.Background(), mirrorHost, hostCertPEM(t, mirrorHost, time.Now().Add(80*24*time.Hour))); err != nil {
		t.Fatal(err)
	}
	e := testIssuerEdge(mirrorHost) // ungated
	m := newCertMirror(release.NewDirStore(t.TempDir()), local)
	get := m.wrapGetCertificate(offlineManager(t, local, e.issuerHostPolicy()).GetCertificate)
	if _, err := get(ecdsaHello(mirrorHost)); err != nil {
		t.Fatalf("an empty mirror must fall through to autocert: %v", err)
	}
}

// Last resort: gated autocert AND an empty mirror (a bucket outage across a
// restart), but the cert is on this box's disk from before. autocert refuses
// it — HostPolicy runs before its cache lookup — so the chain reads the
// DirCache directly rather than dropping the handshake.
func TestMirrorLastResortServesTheLocalCacheEvenWhenGated(t *testing.T) {
	local := autocert.DirCache(t.TempDir())
	if err := local.Put(context.Background(), mirrorHost, hostCertPEM(t, mirrorHost, time.Now().Add(80*24*time.Hour))); err != nil {
		t.Fatal(err)
	}
	m := newCertMirror(release.NewDirStore(t.TempDir()), local)
	get, _ := gatedChain(t, m, local, mirrorHost)
	got, err := get(ecdsaHello(mirrorHost))
	if err != nil {
		t.Fatalf("the local cache is the last resort, and it has the cert: %v", err)
	}
	if got.Leaf.VerifyHostname(mirrorHost) != nil {
		t.Fatal("served cert does not cover the host")
	}
}

// Nothing anywhere: the error the caller sees is autocert's own, so the
// journal still says WHY (not-issuer, or unknown host) rather than blaming
// the mirror.
func TestMirrorChainKeepsAutocertsError(t *testing.T) {
	get, _ := gatedChain(t, newCertMirror(release.NewDirStore(t.TempDir()), nil), autocert.DirCache(t.TempDir()), mirrorHost)
	_, err := get(ecdsaHello(mirrorHost))
	if !errors.Is(err, errNotIssuer) {
		t.Fatalf("chain error = %v, want errNotIssuer", err)
	}
}

// A mirrored entry that is expired, or does not cover the name asked for, is
// not served: the chain falls through instead. Serving a dead cert is worse
// than serving none — the handshake fails either way, but a fall-through can
// still find a good one.
func TestMirrorRefusesUnusableEntries(t *testing.T) {
	for _, tc := range []struct {
		name string
		key  string
		pem  []byte
	}{
		{"expired", mirrorHost, hostCertPEM(t, mirrorHost, time.Now().Add(-time.Hour))},
		{"wrong host", mirrorHost, hostCertPEM(t, "other.bab.oncarlos.com", time.Now().Add(80*24*time.Hour))},
	} {
		t.Run(tc.name, func(t *testing.T) {
			s := release.NewDirStore(t.TempDir())
			putBucket(t, s, tc.key, tc.pem)
			m := newCertMirror(s, nil)
			get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
			if err := m.Refresh(context.Background()); err != nil {
				t.Fatal(err)
			}
			if _, err := get(ecdsaHello(mirrorHost)); !errors.Is(err, errNotIssuer) {
				t.Fatalf("an unusable mirrored entry must fall through, got %v", err)
			}
		})
	}
}

// autocert keys legacy clients' certs under "<host>+rsa", so the mirror must
// look under the SAME name for the SAME client — with both entries present,
// the choice is the one autocert would have made.
func TestMirrorPicksTheEntryAutocertWouldHave(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	ecdsaPEM := hostCertPEM(t, mirrorHost, time.Now().Add(80*24*time.Hour))
	rsaPEM := hostCertPEM(t, mirrorHost, time.Now().Add(70*24*time.Hour))
	putBucket(t, s, mirrorHost, ecdsaPEM)
	putBucket(t, s, mirrorHost+"+rsa", rsaPEM)
	m := newCertMirror(s, nil)
	get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	rsaOnly := &tls.ClientHelloInfo{
		ServerName:       mirrorHost,
		SignatureSchemes: []tls.SignatureScheme{tls.PKCS1WithSHA256},
		CipherSuites:     []uint16{tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256},
	}
	for _, tc := range []struct {
		name  string
		hello *tls.ClientHelloInfo
		want  []byte
	}{
		{"ecdsa client", ecdsaHello(mirrorHost), ecdsaPEM},
		{"rsa-only client", rsaOnly, rsaPEM},
	} {
		got, err := get(tc.hello)
		if err != nil {
			t.Fatalf("%s: %v", tc.name, err)
		}
		want, err := parseCertEntry(tc.want)
		if err != nil {
			t.Fatal(err)
		}
		if !bytes.Equal(got.Certificate[0], want.Certificate[0]) {
			t.Fatalf("%s was served the other entry", tc.name)
		}
	}
}

// ---- tls-alpn-01: the challenge autocert PREFERS ----------------------------

func alpnHello(host string) *tls.ClientHelloInfo {
	h := ecdsaHello(host)
	h.SupportedProtos = []string{acme.ALPNProto}
	return h
}

// autocert tries tls-alpn-01 first and only falls back to http-01 after a
// whole failed order, so an N-edge deployment would burn a failed validation
// on (N-1)/N of its orders if only the issuer could answer these.
func TestMirrorAnswersAnotherEdgesALPNValidation(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, mirrorHost+"+token", hostCertPEM(t, mirrorHost, time.Now().Add(24*time.Hour)))
	m := newCertMirror(s, nil)
	get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	m.allowed = func(h string) bool { return h == mirrorHost }
	// Deliberately NOT refreshed: a validation is live, and the poll is 30s.
	if _, err := get(alpnHello(mirrorHost)); err != nil {
		t.Fatalf("a non-issuer must answer another edge's tls-alpn-01 validation: %v", err)
	}
}

// The SNI here is attacker-chosen, so the live lookup is bounded to the names
// this DEPLOYMENT serves — this box's own routes, or another box's
// exactly-named ones (#277). A stranger is refused before any bucket read, as
// it always was.
func TestMirrorALPNReadThroughOnlyForServedOrRemoteHomedHosts(t *testing.T) {
	s := &bucketGetCounter{Store: release.NewDirStore(t.TempDir())}
	m := newCertMirror(s, nil)
	get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	m.allowed = func(string) bool { return false }
	if _, err := get(alpnHello("stranger.example")); err == nil {
		t.Fatal("there is no token cert for a host this deployment does not serve")
	}
	if n := s.gets(); n != 0 {
		t.Fatalf("bucket gets = %d, want 0 for an unserved host", n)
	}
}

// A token cert is NOT a certificate for the host: it is self-signed and only
// meaningful to a validating CA. It must never leak into an ordinary
// handshake.
func TestMirrorNeverServesATokenCertToAnOrdinaryHello(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, mirrorHost+"+token", hostCertPEM(t, mirrorHost, time.Now().Add(24*time.Hour)))
	m := newCertMirror(s, nil)
	get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	m.allowed = func(string) bool { return true }
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	if _, err := get(ecdsaHello(mirrorHost)); !errors.Is(err, errNotIssuer) {
		t.Fatalf("an ordinary hello must not be served the tls-alpn-01 token cert: %v", err)
	}
}

// The wildcard stays in front of everything: a covered name never reaches the
// mirror at all (this is the composition RunWithActivatorReady builds).
func TestWildcardWinsAheadOfTheMirror(t *testing.T) {
	const parent = "bab.oncarlos.com"
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, mirrorHost, hostCertPEM(t, mirrorHost, time.Now().Add(80*24*time.Hour)))
	m := newCertMirror(s, nil)
	inner, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	w := newWildcardManager([]string{parent}, "", func() []string { return nil }, nil, "", autocert.DirCache(t.TempDir()), &fakeDNS{})
	wildcard, _ := selfSignedWildcard(t, parent, time.Now().Add(80*24*time.Hour))
	w.refresh()
	w.mu.Lock()
	w.certs[parent] = wildcard
	w.mu.Unlock()

	got, err := wrapGetCertificate(w, inner)(ecdsaHello(mirrorHost))
	if err != nil {
		t.Fatal(err)
	}
	if got != wildcard {
		t.Fatal("a wildcard-covered name must be served the wildcard, not the mirror's per-host cert")
	}
}

// ---- R2: challenges, in front of autocert's own handler ---------------------

const mirrorTestToken = "0123456789abcdefghij0123456789abcdefghijkl"

func challengeRequest(t *testing.T, h http.Handler, token string) *httptest.ResponseRecorder {
	t.Helper()
	return challengeRequestHost(t, h, mirrorHost, token)
}

// gatedHTTPHandler is the :80 chain of a non-issuer: the mirror's handler in
// front of autocert's, which 403s every challenge on this box.
func gatedHTTPHandler(t *testing.T, m *certMirror) http.Handler {
	t.Helper()
	return gatedHTTPHandlerFor(t, testIssuerEdge(mirrorHost), m)
}

// gatedHTTPHandlerFor is gatedHTTPHandler over an edge the caller built, for
// the same reason gatedChainFor exists.
func gatedHTTPHandlerFor(t *testing.T, e *Edge, m *certMirror) http.Handler {
	t.Helper()
	e.issuer = func() (context.Context, bool) { return nil, false }
	mgr := offlineManager(t, autocert.DirCache(t.TempDir()), e.issuerHostPolicy())
	fallback := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("X-Fallback", "yes")
		w.WriteHeader(http.StatusTeapot)
	})
	return m.challengeHandler(mgr.HTTPHandler(fallback))
}

// autocert's HTTPHandler runs its HostPolicy — the issuance gate — before it
// ever looks a token up, so a non-issuer 403s a challenge it is holding the
// answer to. The mirror answers first.
func TestChallengeHandlerAnswersInFrontOfAutocert(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, mirrorTestToken+"+http-01", []byte("keyauth"))
	m := newCertMirror(s, nil)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	got := challengeRequest(t, gatedHTTPHandler(t, m), mirrorTestToken)
	if got.Code != 200 || got.Body.String() != "keyauth" {
		t.Fatalf("challenge = %d %q, want 200 \"keyauth\"", got.Code, got.Body.String())
	}
}

// Everything that is not a challenge path is somebody else's request.
func TestChallengeHandlerDelegatesEverythingElse(t *testing.T) {
	m := newCertMirror(release.NewDirStore(t.TempDir()), nil)
	r := httptest.NewRequest("GET", "http://"+mirrorHost+"/whatever", nil)
	w := httptest.NewRecorder()
	gatedHTTPHandler(t, m).ServeHTTP(w, r)
	if w.Header().Get("X-Fallback") != "yes" {
		t.Fatalf("non-challenge request = %d, want the fallback handler", w.Code)
	}
}

// A live validation is faster than a 30s poll: the token was written to the
// bucket a second ago and Accept() has already been sent. A miss reads
// through — once, then from a short-lived cache.
func TestChallengeReadsThroughForLiveTokens(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, mirrorTestToken+"+http-01", []byte("keyauth"))
	counting := &bucketGetCounter{Store: s}
	m := newCertMirror(counting, nil) // deliberately never refreshed
	h := gatedHTTPHandler(t, m)
	for range 3 {
		got := challengeRequest(t, h, mirrorTestToken)
		if got.Code != 200 || got.Body.String() != "keyauth" {
			t.Fatalf("read-through challenge = %d %q", got.Code, got.Body.String())
		}
	}
	if n := counting.gets(); n != 1 {
		t.Fatalf("bucket gets = %d, want 1 — a live token is cached briefly after the first read", n)
	}
}

// A miss is negative-cached: a scanner walking /.well-known/acme-challenge/
// must not turn into one bucket GET per request.
func TestChallengeReadThroughNegativeCachesMisses(t *testing.T) {
	counting := &bucketGetCounter{Store: release.NewDirStore(t.TempDir())}
	m := newCertMirror(counting, nil)
	h := gatedHTTPHandler(t, m)
	for range 5 {
		if got := challengeRequest(t, h, mirrorTestToken); got.Code == 200 {
			t.Fatal("there is no such token; the handler must not answer 200")
		}
	}
	if n := counting.gets(); n != 1 {
		t.Fatalf("bucket gets = %d, want 1 — misses are negative-cached", n)
	}
}

// Only names shaped like an ACME token are ever fetched: the read-through is
// a bucket read driven by an unauthenticated URL path.
func TestChallengeReadThroughRefusesNonTokenPaths(t *testing.T) {
	counting := &bucketGetCounter{Store: release.NewDirStore(t.TempDir())}
	m := newCertMirror(counting, nil)
	h := gatedHTTPHandler(t, m)
	for _, token := range []string{"short", "../../control/serial", "wildcard-account.key", strings.Repeat("a", 300)} {
		if got := challengeRequest(t, h, token); got.Code == 200 {
			t.Fatalf("%q answered 200", token)
		}
	}
	if n := counting.gets(); n != 0 {
		t.Fatalf("bucket gets = %d, want 0 — a name that cannot be an ACME token is never fetched", n)
	}
}

// Distinct token names defeat the negative cache, so the read-through carries
// its own budget: a flood degrades to "no read-through", never to a bucket
// GET per request.
func TestChallengeReadThroughIsRateLimited(t *testing.T) {
	counting := &bucketGetCounter{Store: release.NewDirStore(t.TempDir())}
	m := newCertMirror(counting, nil)
	h := gatedHTTPHandler(t, m)
	for i := range 200 {
		challengeRequest(t, h, fmt.Sprintf("%s%03d", strings.Repeat("z", 39), i))
	}
	if n := counting.gets(); n > readThroughBurst {
		t.Fatalf("bucket gets = %d under a flood, want <= %d", n, readThroughBurst)
	}
	if counting.gets() == 0 {
		t.Fatal("test bug: the read-through never ran at all")
	}
}

// bucketGetCounter counts Gets.
type bucketGetCounter struct {
	release.Store
	mu sync.Mutex
	n  int
}

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

func (c *bucketGetCounter) gets() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.n
}

// slowGetStore blocks every Get until release is closed.
type slowGetStore struct {
	release.Store
	release chan struct{}
	mu      sync.Mutex
	n       int
}

func (s *slowGetStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.mu.Lock()
	s.n++
	s.mu.Unlock()
	<-s.release
	return s.Store.Get(ctx, key)
}

func (s *slowGetStore) gets() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.n
}

// A CA opens several connections for one validation. They share one bucket
// read: the slow one is in flight, the rest wait on it.
func TestChallengeReadThroughIsSingleFlighted(t *testing.T) {
	inner := release.NewDirStore(t.TempDir())
	putBucket(t, inner, mirrorTestToken+"+http-01", []byte("keyauth"))
	s := &slowGetStore{Store: inner, release: make(chan struct{})}
	m := newCertMirror(s, nil)
	h := gatedHTTPHandler(t, m)

	var wg sync.WaitGroup
	got := make([]int, 6)
	for i := range got {
		wg.Add(1)
		go func() {
			defer wg.Done()
			got[i] = challengeRequest(t, h, mirrorTestToken).Code
		}()
	}
	// Let the waiters pile up behind the leader, then let it through.
	time.Sleep(50 * time.Millisecond)
	close(s.release)
	wg.Wait()

	for i, code := range got {
		if code != 200 {
			t.Fatalf("request %d = %d, want 200", i, code)
		}
	}
	if n := s.gets(); n != 1 {
		t.Fatalf("bucket gets = %d, want 1 — concurrent lookups share one read", n)
	}
}

// The loop is the whole delivery mechanism: it must pick material up without
// anybody asking, keep going after a failure, and stop when told.
func TestCertMirrorLoopPollsAndStops(t *testing.T) {
	prev := certMirrorInterval
	certMirrorInterval = 5 * time.Millisecond
	t.Cleanup(func() { certMirrorInterval = prev })
	log.SetOutput(io.Discard)
	t.Cleanup(func() { log.SetOutput(os.Stderr) })

	s := &listFailingStore{Store: release.NewDirStore(t.TempDir())}
	s.setFail(true) // the loop starts against a broken bucket
	m := newCertMirror(s, nil)
	stop := make(chan struct{})
	done := make(chan struct{})
	go func() {
		m.run(stop)
		close(done)
	}()

	putBucket(t, s.Store, "example.com", []byte("PEM"))
	s.setFail(false)
	deadline := time.Now().Add(2 * time.Second)
	for {
		if _, ok := m.Cert("example.com"); ok {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("the loop never picked the cert up")
		}
		time.Sleep(2 * time.Millisecond)
	}
	close(stop)
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("the loop did not stop")
	}
}

// autocert arms its renewal timers LAZILY, from GetCertificate: m.cert()
// loads the cache and calls startRenew (autocert.go:464), and nothing else
// ever does. A mirror that short-circuited every hello would therefore
// silently stop an ISSUER from renewing anything it had not ordered in this
// process — the certs would simply expire in 90 days.
//
// So the short-circuit is for the GATED case only, which is the case it
// exists for: an ungated box's autocert can serve everything the mirror holds
// anyway, because syncLocal has already put it in its cache.
func TestMirrorLetsAnIssuersAutocertSeeTheHello(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, mirrorHost, hostCertPEM(t, mirrorHost, time.Now().Add(80*24*time.Hour)))

	local := autocert.DirCache(t.TempDir())
	counting := &countingCache{Cache: local, gets: map[string]int{}}
	m := newCertMirror(s, local)
	m.gated = func() bool { return false } // this box IS the issuer
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	e := testIssuerEdge(mirrorHost) // ungated host policy
	get := m.wrapGetCertificate(offlineManager(t, counting, e.issuerHostPolicy()).GetCertificate)
	if _, err := get(ecdsaHello(mirrorHost)); err != nil {
		t.Fatalf("the issuer must serve the mirrored cert (through autocert): %v", err)
	}
	if n := counting.count(mirrorHost); n == 0 {
		t.Fatal("autocert never saw the hello — its renewal timer for this host is never armed")
	}
}

// The mirror is still there behind autocert on an ungated box: a name
// autocert cannot serve (and, being offline, cannot order) is rescued.
func TestMirrorStillRescuesAnUngatedBox(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, mirrorHost, hostCertPEM(t, mirrorHost, time.Now().Add(80*24*time.Hour)))
	m := newCertMirror(s, nil)
	m.gated = func() bool { return false }
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	e := testIssuerEdge(mirrorHost)
	get := m.wrapGetCertificate(offlineManager(t, autocert.DirCache(t.TempDir()), e.issuerHostPolicy()).GetCertificate)
	if _, err := get(ecdsaHello(mirrorHost)); err != nil {
		t.Fatalf("an ungated box whose order fails must still fall back to the mirror: %v", err)
	}
}

// A mirror with no gate wired (every test, and any future caller that
// forgets) behaves as though this box were gated: it serves, which is always
// safe, rather than assuming an issuance authority it was never given.
func TestMirrorWithNoGateDefaultsToServingFirst(t *testing.T) {
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, mirrorHost, hostCertPEM(t, mirrorHost, time.Now().Add(80*24*time.Hour)))
	m := newCertMirror(s, nil) // no gate
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	local := autocert.DirCache(t.TempDir())
	if err := local.Put(context.Background(), mirrorHost, hostCertPEM(t, mirrorHost, time.Now().Add(80*24*time.Hour))); err != nil {
		t.Fatal(err)
	}
	counting := &countingCache{Cache: local, gets: map[string]int{}}
	e := testIssuerEdge(mirrorHost) // ungated host policy: autocert WOULD serve
	get := m.wrapGetCertificate(offlineManager(t, counting, e.issuerHostPolicy()).GetCertificate)
	if _, err := get(ecdsaHello(mirrorHost)); err != nil {
		t.Fatal(err)
	}
	if n := counting.count(mirrorHost); n != 0 {
		t.Fatalf("autocert was consulted %d time(s) — with no gate the mirror answers first", n)
	}
}

// ---- review fixes -----------------------------------------------------------

// keyFailingStore fails Get for exactly the keys named (all of them when
// `all` is set), and counts every Get.
type keyFailingStore struct {
	release.Store
	mu   sync.Mutex
	fail map[string]bool
	all  bool
	n    int
}

func (k *keyFailingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	k.mu.Lock()
	k.n++
	bad := k.all || k.fail[key]
	k.mu.Unlock()
	if bad {
		return nil, "", errors.New("object unreadable")
	}
	return k.Store.Get(ctx, key)
}

func (k *keyFailingStore) setFail(keys ...string) {
	k.mu.Lock()
	k.fail = map[string]bool{}
	for _, key := range keys {
		k.fail[key] = true
	}
	k.mu.Unlock()
}

func (k *keyFailingStore) setAll(v bool) {
	k.mu.Lock()
	k.all = v
	k.mu.Unlock()
}

func (k *keyFailingStore) gets() int {
	k.mu.Lock()
	defer k.mu.Unlock()
	return k.n
}

// FINDING 3: one unreadable object used to discard the WHOLE rebuild — every
// unrelated new cert with it, plus syncLocal and the wildcard wake. A poll
// fails only when the listing fails, or when nothing at all could be read.
func TestCertMirrorPollCarriesUnreadableObjectsForward(t *testing.T) {
	inner := release.NewDirStore(t.TempDir())
	s := &keyFailingStore{Store: inner}
	putBucket(t, inner, "a.example.com", []byte("A"))
	putBucket(t, inner, "b.example.com", []byte("B"))
	local := autocert.DirCache(t.TempDir())
	m := newCertMirror(s, local)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}

	// a.example.com goes unreadable; a brand-new cert lands at the same time.
	s.setFail(certKey("a.example.com"))
	putBucket(t, inner, "c.example.com", hostCertPEM(t, "c.example.com", time.Now().Add(80*24*time.Hour)))
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatalf("one bad object must not fail the whole poll: %v", err)
	}
	if b, ok := m.Cert("a.example.com"); !ok || string(b) != "A" {
		t.Fatalf("the unreadable object's previous copy = %q, %v — want it carried forward", b, ok)
	}
	if _, ok := m.Cert("c.example.com"); !ok {
		t.Fatal("the new cert must land even though a sibling object was unreadable")
	}
	if _, err := local.Get(context.Background(), "c.example.com"); err != nil {
		t.Fatalf("syncLocal must still run on a partially failed poll: %v", err)
	}

	// Everything unreadable IS a failed poll — and the last-good map stands.
	s.setAll(true)
	if err := m.Refresh(context.Background()); err == nil {
		t.Fatal("a poll where every object failed must be reported")
	}
	if _, ok := m.Cert("b.example.com"); !ok {
		t.Fatal("a failed poll must keep the last-good map")
	}
}

// FINDING 1: tls-alpn-01 token certs are keyed by HOSTNAME, not by a unique
// token, so a previous order's entry for the same host is a live collision.
// The snapshot is up to a poll stale (and, if a gated Delete leaked, far
// staler than that) — token certs are read live, never served from the poll.
func TestALPNTokenCertIsReadLiveNotFromTheSnapshot(t *testing.T) {
	inner := release.NewDirStore(t.TempDir())
	s := &bucketGetCounter{Store: inner}
	stale := hostCertPEM(t, mirrorHost, time.Now().Add(24*time.Hour))
	putBucket(t, inner, mirrorHost+"+token", stale)

	m := newCertMirror(s, nil)
	m.allowed = func(h string) bool { return h == mirrorHost }
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	// A new order for the same host replaces the token cert in the bucket.
	fresh := hostCertPEM(t, mirrorHost, time.Now().Add(23*time.Hour))
	putBucket(t, inner, mirrorHost+"+token", fresh)

	get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	got, err := get(alpnHello(mirrorHost))
	if err != nil {
		t.Fatal(err)
	}
	want, err := parseCertEntry(fresh)
	if err != nil {
		t.Fatal(err)
	}
	if !bytes.Equal(got.Certificate[0], want.Certificate[0]) {
		t.Fatal("the CA was handed the PREVIOUS order's token cert — the validation fails and the order burns")
	}
	if s.gets() == 0 {
		t.Fatal("a token cert must be read live, not served out of the poll's snapshot")
	}
}

// ...and a token read-through is remembered for seconds, not for a poll: two
// orders for one host can be minutes apart, but they can also be seconds.
func TestALPNTokenReadThroughTTLIsShort(t *testing.T) {
	if readThroughTTL(mirrorHost+alpnTokenSuffix) > 5*time.Second {
		t.Fatalf("token TTL = %s, want a few seconds at most", readThroughTTL(mirrorHost+alpnTokenSuffix))
	}
	if got := readThroughTTL("tok" + httpTokenSuffix); got != readThroughHitTTL {
		t.Fatalf("HTTP-01 token TTL = %s, want %s — those names are unique per order", got, readThroughHitTTL)
	}

	prev := readThroughTokenTTL
	readThroughTokenTTL = 0
	t.Cleanup(func() { readThroughTokenTTL = prev })
	inner := release.NewDirStore(t.TempDir())
	s := &bucketGetCounter{Store: inner}
	putBucket(t, inner, mirrorHost+"+token", hostCertPEM(t, mirrorHost, time.Now().Add(24*time.Hour)))
	m := newCertMirror(s, nil)
	m.allowed = func(string) bool { return true }
	get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	for range 2 {
		if _, err := get(alpnHello(mirrorHost)); err != nil {
			t.Fatal(err)
		}
	}
	if n := s.gets(); n != 2 {
		t.Fatalf("bucket gets = %d, want 2 — an expired token memo must be re-read", n)
	}
}

// FINDING 2: some clients (curl, and anything not doing IDNA itself) send
// unicode SNI. autocert converts with idna.Lookup.ToASCII before it looks in
// its cache (autocert.go:307); a mirror that did not would hard-fail a
// handshake the same box serves today.
func TestMirrorConvertsUnicodeSNIToPunycode(t *testing.T) {
	const unicodeHost, asciiHost = "bücher.example", "xn--bcher-kva.example"
	s := release.NewDirStore(t.TempDir())
	putBucket(t, s, asciiHost, hostCertPEM(t, asciiHost, time.Now().Add(80*24*time.Hour)))
	m := newCertMirror(s, nil)
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), asciiHost)
	got, err := get(ecdsaHello(unicodeHost))
	if err != nil {
		t.Fatalf("unicode SNI must find the punycode cache name: %v", err)
	}
	if got.Leaf.VerifyHostname(asciiHost) != nil {
		t.Fatal("served cert does not cover the punycode host")
	}
	// The last-resort local read looks under the same converted name.
	local := autocert.DirCache(t.TempDir())
	if err := local.Put(context.Background(), asciiHost, hostCertPEM(t, asciiHost, time.Now().Add(80*24*time.Hour))); err != nil {
		t.Fatal(err)
	}
	m2 := newCertMirror(release.NewDirStore(t.TempDir()), local)
	get2, _ := gatedChain(t, m2, local, asciiHost)
	if _, err := get2(ecdsaHello(unicodeHost)); err != nil {
		t.Fatalf("the local last resort must convert too: %v", err)
	}
}

// A name IDNA cannot convert (and an empty one) must not panic or fetch: it
// falls through as an ordinary miss.
func TestMirrorSurvivesUnconvertibleSNI(t *testing.T) {
	m := newCertMirror(release.NewDirStore(t.TempDir()), nil)
	get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	for _, sni := range []string{"", "-.-", strings.Repeat("x", 300) + ".example"} {
		if _, err := get(ecdsaHello(sni)); err == nil {
			t.Fatalf("%q: want a miss, got a certificate", sni)
		}
	}
}

// challengeRequestHost is challengeRequest with an explicit Host header.
func challengeRequestHost(t *testing.T, h http.Handler, host, token string) *httptest.ResponseRecorder {
	t.Helper()
	r := httptest.NewRequest("GET", "http://"+host+"/.well-known/acme-challenge/"+token, nil)
	w := httptest.NewRecorder()
	h.ServeHTTP(w, r)
	return w
}

// FINDING 4a: the live lookup on :80 is bounded by the host under
// validation, exactly as autocert's own handler bounds its token lookup. A
// real CA always sends it; a flood against hosts this box does not serve
// never reaches the bucket.
func TestChallengeReadThroughIsBoundedByHost(t *testing.T) {
	inner := release.NewDirStore(t.TempDir())
	putBucket(t, inner, mirrorTestToken+"+http-01", []byte("keyauth"))
	s := &bucketGetCounter{Store: inner}
	m := newCertMirror(s, nil)
	m.allowed = func(h string) bool { return h == mirrorHost }
	h := gatedHTTPHandler(t, m)

	if got := challengeRequestHost(t, h, "stranger.example", mirrorTestToken); got.Code == 200 {
		t.Fatal("a host this box does not serve must not drive a bucket read")
	}
	if n := s.gets(); n != 0 {
		t.Fatalf("bucket gets for an unserved host = %d, want 0", n)
	}
	if got := challengeRequestHost(t, h, mirrorHost+":80", mirrorTestToken); got.Code != 200 {
		t.Fatalf("the host under validation must be answered: %d", got.Code)
	}
}

// FINDING 4b: :80 and 443 do not share one budget. An unauthenticated flood
// of challenge requests must not starve a tls-alpn-01 rescue on 443.
func TestChallengeFloodDoesNotStarveALPNRescue(t *testing.T) {
	inner := release.NewDirStore(t.TempDir())
	putBucket(t, inner, mirrorHost+"+token", hostCertPEM(t, mirrorHost, time.Now().Add(24*time.Hour)))
	s := &bucketGetCounter{Store: inner}
	m := newCertMirror(s, nil)
	m.allowed = func(string) bool { return true }
	h := gatedHTTPHandler(t, m)
	for i := range 200 {
		challengeRequest(t, h, fmt.Sprintf("%s%03d", strings.Repeat("z", 39), i))
	}
	get, _ := gatedChain(t, m, autocert.DirCache(t.TempDir()), mirrorHost)
	if _, err := get(alpnHello(mirrorHost)); err != nil {
		t.Fatalf("a :80 flood starved the 443 validation path: %v", err)
	}
}

// ---- #277: the three gates admit the deployment's exactly-named hosts -------
//
// DNS for a home-boxed route can point at ANY edge. The edge it points at
// backhauls to the home box with the real SNI, the home box (as issuer)
// orders — and the CA then validates against the box DNS names, which is the
// one whose registry does not carry the host. Bounding the gates by Allowed
// meant every such validation failed, five failures burned the name's Let's
// Encrypt per-hostname budget in minutes, and the route was down throughout.
//
// The gates now ask deploymentHost: this box's routes, OR an exact key of the
// mirrored home map. Exact is the whole discipline — every test below that
// says "wildcard label" is PR #180's rule standing on the new surface.

// remoteHomedEdge is one box of a two-box deployment: it serves served from
// its own registry, and mirrors an edges file in which box "betty" homes
// homed (pass "*.<parent>" to mirror a catch-all row, which Task 1 publishes
// verbatim). The handle it issues and mirrors through COUNTS bucket Gets, so
// "did this gate spend a read-through?" is observable; the home map polls the
// same directory through its own handle, so the map's own reads never show up
// in that count.
func remoteHomedEdge(t *testing.T, served []string, homed ...string) (*Edge, *bucketGetCounter, release.Store) {
	t.Helper()
	base := release.NewDirStore(t.TempDir())
	counting := &bucketGetCounter{Store: base}
	e := testIssuerEdge(served...)
	e.staticStore = counting
	m := newHomeMap(base, "annie")
	homedOn(t, m, "betty", "2.2.2.2", homed...)
	e.homes.Store(m)
	return e, counting, base
}

const (
	// remoteHomedHost is homed on the OTHER box: absent from this box's
	// registry, exactly named in its home map.
	remoteHomedHost = "app.remote.test"
	// remoteCatchAll is the catch-all row that box publishes, and
	// remoteWildcardLabel is a label under it — dialable through Home()'s
	// one-label rule, and admissible by none of the three cert gates.
	remoteCatchAll      = "*.remote.test"
	remoteWildcardLabel = "invented.remote.test"
)

// HTTP-01, the failing half of #277: the CA arrives at a box that does not
// serve the name, and the token is in the bucket. It is answered.
func TestChallengeReadThroughAdmitsRemoteHomedHost(t *testing.T) {
	e, counting, base := remoteHomedEdge(t, []string{mirrorHost}, remoteHomedHost)
	putBucket(t, base, mirrorTestToken+"+http-01", []byte("keyauth"))
	m := e.newCertMirror(nil) // production wiring: m.allowed = e.deploymentHost
	if m == nil {
		t.Fatal("a store-backed edge must have a mirror")
	}
	if e.Allowed(remoteHomedHost) {
		t.Fatal("test bug: the host is supposed to be absent from THIS box's registry")
	}
	got := challengeRequestHost(t, gatedHTTPHandlerFor(t, e, m), remoteHomedHost, mirrorTestToken)
	if got.Code != 200 || got.Body.String() != "keyauth" {
		t.Fatalf("challenge for a remote-homed host = %d %q, want 200 \"keyauth\"", got.Code, got.Body.String())
	}
	if n := counting.gets(); n != 1 {
		t.Fatalf("bucket gets = %d, want exactly 1 — the live read-through", n)
	}
}

// ...and the anti-stranger bound the widening had to survive: a name in
// neither the registry nor the map spends nothing at all, and neither does an
// invented label under a MIRRORED catch-all (#180).
func TestChallengeReadThroughStillRefusesStrangers(t *testing.T) {
	for _, host := range []string{"stranger.example", remoteWildcardLabel, remoteCatchAll} {
		t.Run(host, func(t *testing.T) {
			e, counting, base := remoteHomedEdge(t, []string{mirrorHost}, remoteHomedHost, remoteCatchAll)
			putBucket(t, base, mirrorTestToken+"+http-01", []byte("keyauth"))
			m := e.newCertMirror(nil)
			if got := challengeRequestHost(t, gatedHTTPHandlerFor(t, e, m), host, mirrorTestToken); got.Code == 200 {
				t.Fatalf("%s was answered a live token body", host)
			}
			if n := counting.gets(); n != 0 {
				t.Fatalf("bucket gets = %d, want 0 — %s must never drive a bucket read", n, host)
			}
		})
	}
}

// tls-alpn-01, the half autocert tries FIRST: the validation cert for another
// edge's order is served for a remote-homed name.
func TestMirrorALPNReadThroughAdmitsRemoteHomedHost(t *testing.T) {
	e, _, base := remoteHomedEdge(t, []string{mirrorHost}, remoteHomedHost)
	putBucket(t, base, remoteHomedHost+"+token", hostCertPEM(t, remoteHomedHost, time.Now().Add(24*time.Hour)))
	m := e.newCertMirror(nil)
	get, _ := gatedChainFor(t, e, m, autocert.DirCache(t.TempDir()))
	// Deliberately NOT refreshed: a validation is live, and the poll is 30s.
	if _, err := get(alpnHello(remoteHomedHost)); err != nil {
		t.Fatalf("an edge DNS points at must answer the home box's tls-alpn-01 validation: %v", err)
	}
}

// The #180 pin on the 443 surface: SNI is attacker-chosen, and a mirrored
// catch-all must not turn an invented label into a bucket read.
func TestALPNReadThroughRefusesWildcardLabelUnderRemoteCatchAll(t *testing.T) {
	e, counting, base := remoteHomedEdge(t, []string{mirrorHost}, remoteCatchAll)
	putBucket(t, base, remoteWildcardLabel+"+token", hostCertPEM(t, remoteWildcardLabel, time.Now().Add(24*time.Hour)))
	if _, _, ok := e.homes.Load().Home(remoteWildcardLabel); !ok {
		t.Fatal("test bug: the label is supposed to be DIALABLE through the one-label rule")
	}
	m := e.newCertMirror(nil)
	get, _ := gatedChainFor(t, e, m, autocert.DirCache(t.TempDir()))
	if _, err := get(alpnHello(remoteWildcardLabel)); err == nil {
		t.Fatal("a label under a mirrored catch-all must not be served a token cert")
	}
	if n := counting.gets(); n != 0 {
		t.Fatalf("bucket gets = %d, want 0 — the label must not spend the ALPN budget", n)
	}
}

// The order gate: the issuer mints for any exactly-named route in the
// deployment, whichever box homes it. Without this the home-boxed route can
// only ever mint while the housekeeper lease sits on its own box.
func TestIssuerOrdersForRemoteHomedHost(t *testing.T) {
	e, _, _ := remoteHomedEdge(t, []string{mirrorHost}, remoteHomedHost)
	e.issuer = func() (context.Context, bool) { return context.Background(), true }
	if err := e.issuerHostPolicy()(context.Background(), remoteHomedHost); err != nil {
		t.Fatalf("HostPolicy for a remote-homed host = %v, want nil", err)
	}
	if err := e.issuerHostPolicy()(context.Background(), mirrorHost); err != nil {
		t.Fatalf("HostPolicy for this box's own route = %v, want nil", err)
	}
}

// The #180 pin on the order gate, and the one that matters most: an
// attacker-chosen label must never drive a fresh per-host ACME order against
// the deployment's single shared account and rate-limit bucket.
func TestHostPolicyRefusesWildcardLabelUnderRemoteCatchAll(t *testing.T) {
	e, _, _ := remoteHomedEdge(t, []string{mirrorHost}, remoteCatchAll)
	e.issuer = func() (context.Context, bool) { return context.Background(), true }
	for _, host := range []string{remoteWildcardLabel, remoteCatchAll, "stranger.example"} {
		err := e.issuerHostPolicy()(context.Background(), host)
		if err == nil {
			t.Fatalf("%s: HostPolicy allowed an order", host)
		}
		if !strings.Contains(err.Error(), "not in registry") {
			t.Fatalf("%s: HostPolicy = %v, want the unchanged not-in-registry refusal", host, err)
		}
	}
}

// Widening the NAMES a policy admits does not widen the BOXES that may order:
// a non-issuer still refuses every name, remote-homed ones included.
func TestHostPolicyStillGatesNonIssuer(t *testing.T) {
	e, _, _ := remoteHomedEdge(t, []string{mirrorHost}, remoteHomedHost)
	e.issuer = func() (context.Context, bool) { return nil, false }
	for _, host := range []string{remoteHomedHost, mirrorHost} {
		if err := e.issuerHostPolicy()(context.Background(), host); !errors.Is(err, errNotIssuer) {
			t.Fatalf("%s on a non-issuer = %v, want errNotIssuer", host, err)
		}
	}
}

// Single-box deployments are byte-for-byte unchanged: no home map armed (or an
// empty one), so deploymentHost reduces to Allowed.
func TestDeploymentHostIsAllowedOnASingleBoxDeployment(t *testing.T) {
	e := testIssuerEdge(mirrorHost) // e.homes never stored: nil pointer
	if !e.deploymentHost(mirrorHost) || e.deploymentHost("stranger.example") {
		t.Fatal("with no home map at all, deploymentHost must be exactly Allowed")
	}
	e2, _, _ := remoteHomedEdge(t, []string{mirrorHost}) // map armed, folds to nothing
	if !e2.deploymentHost(mirrorHost) || e2.deploymentHost("stranger.example") {
		t.Fatal("with an empty home map, deploymentHost must be exactly Allowed")
	}
}

// slowStore adds a fixed delay to every Get, standing in for a box that
// reads the deployment bucket from the other side of the planet.
type slowStore struct {
	release.Store
	delay          time.Duration
	mu             sync.Mutex
	peak           int // greatest number of gets in flight at once
	cur            int
	peakGoroutines int
}

func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.mu.Lock()
	s.cur++
	if s.cur > s.peak {
		s.peak = s.cur
	}
	// Sampled while the queue is deepest, which is what tells a fixed worker
	// pool from one goroutine per object: the first is flat in the number of
	// objects, the second is not.
	if g := runtime.NumGoroutine(); g > s.peakGoroutines {
		s.peakGoroutines = g
	}
	s.mu.Unlock()
	time.Sleep(s.delay)
	s.mu.Lock()
	s.cur--
	s.mu.Unlock()
	return s.Store.Get(ctx, key)
}

// TestCertMirrorRefreshFetchesConcurrently is amadan #118's second half as a
// test. The mirror's boot refresh runs BEFORE either listener binds, so its
// cost is the edge's time-to-serve; the self-update gate allows 20s for the
// edge to answer on loopback. Serially, the pass costs count x latency, and
// the Sydney box (101 certs, ~250ms to eu-west-1) therefore took ~25s to
// start serving and failed every gate from 2026-09-07 to 2026-09-19.
//
// The property is that the pass does NOT cost count x latency. Asserted as
// wall-clock against a deliberately slow store, and by the peak number of
// gets in flight, so a change that quietly re-serialises it is caught by
// both.
func TestCertMirrorRefreshFetchesConcurrently(t *testing.T) {
	const (
		certs = 40
		delay = 20 * time.Millisecond
	)
	inner := release.NewDirStore(t.TempDir())
	for i := range certs {
		putBucket(t, inner, fmt.Sprintf("host%02d.example.com", i), []byte("PEM"))
	}
	s := &slowStore{Store: inner, delay: delay}
	m := newCertMirror(s, nil)

	base := runtime.NumGoroutine()
	start := time.Now()
	if err := m.Refresh(context.Background()); err != nil {
		t.Fatal(err)
	}
	took := time.Since(start)

	// Serial would be at least certs*delay (800ms).
	if serial := certs * delay; took > serial/2 {
		t.Errorf("refresh took %s, serial would be ~%s — the gets are not concurrent", took, serial)
	}
	// The wall-clock bound above is weak on its own: it admits any
	// concurrency of about three or more, so a silent drop from ten to four
	// would pass it while putting a far-from-bucket box back over the update
	// gate.
	//
	// The floor is a LITERAL, deliberately. Deriving it from
	// certRefreshConcurrency makes the assertion move with the thing it is
	// meant to pin, and a change of the constant to 4 then passes — which is
	// exactly what happened to the first version of this line. Eight is the
	// arithmetic that matters: the update gate allows 20s, a far box is
	// ~250ms a round trip, so the pass must not exceed ~80 waves, and at a
	// few hundred published certs that needs most of ten in flight.
	const floor = 8
	if s.peak < floor {
		t.Errorf("peak gets in flight = %d, want at least %d — too little concurrency to fit the update gate on a far-from-bucket box", s.peak, floor)
	}
	if s.peak > certRefreshConcurrency {
		t.Errorf("peak gets in flight = %d, above the %d bound", s.peak, certRefreshConcurrency)
	}
	// And the pool is FIXED: goroutines must not scale with the number of
	// objects. A goroutine per object costs one stack per cert before a
	// single one is read, which is the shape that breaks at a million certs
	// rather than at a hundred.
	if s.peakGoroutines > base+certRefreshConcurrency+8 {
		t.Errorf("peak goroutines = %d with %d baseline and %d objects — the pool is not fixed", s.peakGoroutines, base, certs)
	}
	// And it still mirrors everything it was asked for.
	for i := range certs {
		if _, ok := m.Cert(fmt.Sprintf("host%02d.example.com", i)); !ok {
			t.Fatalf("host%02d.example.com missing from the mirror", i)
		}
	}
}

// ---- #336: the change index gates the mirror's reads -------------------------

// changeIndexStore is a DirStore with the change index on that records every
// read by method and key. Embedding the concrete store forwards
// ConditionalGetter and ETagLister, so the mirror and the watch see the store
// production hands them, and every write bumps its scope as in production.
type changeIndexStore struct {
	*release.DirStore
	mu      sync.Mutex
	calls   []string
	failGet string // Get of exactly this key fails
	// block, when set, parks every GetIfChanged until it is closed: a hung
	// index GET holding the shared watch's poll lock. entered says one has.
	block   chan struct{}
	entered chan struct{}
}

// blockIndex makes the next index reads hang; closing the returned channel
// (done at cleanup too) releases them.
func (s *changeIndexStore) blockIndex(t *testing.T) (entered <-chan struct{}, release func()) {
	t.Helper()
	s.mu.Lock()
	s.block, s.entered = make(chan struct{}), make(chan struct{}, 16)
	block, ent := s.block, s.entered
	s.mu.Unlock()
	var once sync.Once
	release = func() {
		once.Do(func() {
			s.mu.Lock()
			s.block = nil
			s.mu.Unlock()
			close(block)
		})
	}
	t.Cleanup(release)
	return ent, release
}

func (s *changeIndexStore) setFailGet(key string) { s.mu.Lock(); s.failGet = key; s.mu.Unlock() }

func newChangeIndexStore(t *testing.T) *changeIndexStore {
	t.Helper()
	d := release.NewDirStore(t.TempDir())
	d.EnableChangeIndex()
	return &changeIndexStore{DirStore: d}
}

func (s *changeIndexStore) note(method, key string) {
	s.mu.Lock()
	s.calls = append(s.calls, method+" "+key)
	s.mu.Unlock()
}

func (s *changeIndexStore) reset() { s.mu.Lock(); s.calls = nil; s.mu.Unlock() }

// count is how many recorded calls of method name a key under prefix.
func (s *changeIndexStore) count(method, prefix string) int {
	s.mu.Lock()
	defer s.mu.Unlock()
	n := 0
	for _, c := range s.calls {
		if strings.HasPrefix(c, method+" "+prefix) {
			n++
		}
	}
	return n
}

func (s *changeIndexStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	s.note("Get", key)
	s.mu.Lock()
	fail := s.failGet != "" && s.failGet == key
	s.mu.Unlock()
	if fail {
		return nil, "", errors.New("get refused")
	}
	return s.DirStore.Get(ctx, key)
}

func (s *changeIndexStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
	s.note("GetIfChanged", key)
	s.mu.Lock()
	block, entered := s.block, s.entered
	s.mu.Unlock()
	if block != nil {
		entered <- struct{}{}
		<-block
	}
	return s.DirStore.GetIfChanged(ctx, key, etag)
}

func (s *changeIndexStore) List(ctx context.Context, prefix string) ([]string, error) {
	s.note("List", prefix)
	return s.DirStore.List(ctx, prefix)
}

func (s *changeIndexStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
	s.note("ListETags", prefix)
	return s.DirStore.ListETags(ctx, prefix)
}

// certReads is every bucket read the mirror made under control/certs/.
func certReads(s *changeIndexStore) (list, listETags, get int) {
	return s.count("List", certPrefix), s.count("ListETags", certPrefix), s.count("Get", certPrefix)
}

// armedMirror is a mirror over s with the change index armed: one watch over
// the same store (the agent's, in production) and an injected clock.
func armedMirror(t *testing.T, s *changeIndexStore) (*certMirror, *testClock) {
	t.Helper()
	w := release.NewChangeWatch(s)
	clk := &testClock{t: time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)}
	m := newCertMirror(s, nil)
	m.watch = func() *release.ChangeWatch { return w }
	m.now = clk.now
	return m, clk
}

func mirrorPoll(t *testing.T, m *certMirror) {
	t.Helper()
	if err := m.poll(context.Background()); err != nil {
		t.Fatalf("poll: %v", err)
	}
}

// failingPutCache refuses every Put while fail is set, and counts the Puts
// it lets through.
type failingPutCache struct {
	autocert.Cache
	fail atomic.Bool
	puts atomic.Int32
}

func (c *failingPutCache) Put(ctx context.Context, key string, data []byte) error {
	if c.fail.Load() {
		return errors.New("disk full")
	}
	c.puts.Add(1)
	return c.Cache.Put(ctx, key, data)
}

// Final review R1 finding 7 (amadan #336): a copy into the local cert cache
// that failed is retried on the next poll even when the index says nothing
// moved, from the snapshot the mirror already holds, with no bucket read.
// Once it lands, an unchanged poll writes nothing.
func TestCertMirrorRetriesAFailedLocalCopyOnAnUnchangedPoll(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	pem := hostCertPEM(t, "shop.example.com", time.Now().Add(80*24*time.Hour))
	putBucket(t, s, "shop.example.com", pem)
	m, clk := armedMirror(t, s)
	local := &failingPutCache{Cache: autocert.DirCache(t.TempDir())}
	m.local = local
	local.fail.Store(true)
	mirrorPoll(t, m)
	if _, err := local.Get(context.Background(), "shop.example.com"); err == nil {
		t.Fatal("fixture: the local copy landed while Puts fail")
	}

	local.fail.Store(false)
	s.reset()
	clk.advance(time.Second)
	mirrorPoll(t, m)
	if body, err := local.Get(context.Background(), "shop.example.com"); err != nil || !bytes.Equal(body, pem) {
		t.Fatalf("an unchanged poll after a failed local copy left the local cache without it: %v", err)
	}
	if list, listETags, get := certReads(s); list+listETags+get != 0 {
		t.Fatalf("the retry read the bucket: list %d listETags %d get %d", list, listETags, get)
	}
	n := local.puts.Load()
	clk.advance(time.Second)
	mirrorPoll(t, m)
	if got := local.puts.Load(); got != n {
		t.Fatalf("a settled unchanged poll wrote the local cache again (%d puts, want %d)", got, n)
	}
}

func TestCertMirrorSkipsWhenUnmoved(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putBucket(t, s, "example.com", []byte("PEM"))
	putBucket(t, s, "shop.example.com", []byte("PEM2"))
	m, clk := armedMirror(t, s)
	mirrorPoll(t, m)
	if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM" {
		t.Fatalf("first poll: mirror cert = %q, %v", b, ok)
	}
	s.reset()
	clk.advance(certMirrorInterval)
	mirrorPoll(t, m)
	if l, le, g := certReads(s); l+le+g != 0 {
		t.Fatalf("unmoved poll read the bucket: List %d, ListETags %d, Get %d; want none", l, le, g)
	}
	if n := s.count("GetIfChanged", release.ChangeIndexKey); n != 1 {
		t.Fatalf("unmoved poll: %d index reads, want 1", n)
	}
	if b, ok := m.Cert("shop.example.com"); !ok || string(b) != "PEM2" {
		t.Fatalf("after a skipped poll the mirror = %q, %v — want the same copy", b, ok)
	}
}

func TestCertMirrorFetchesOnlyChanged(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	local := newCountingCache(t)
	pemA := hostCertPEM(t, "a.example.com", time.Now().Add(60*24*time.Hour))
	pemB := hostCertPEM(t, "b.example.com", time.Now().Add(60*24*time.Hour))
	putBucket(t, s, "a.example.com", pemA)
	putBucket(t, s, "b.example.com", pemB)
	putBucket(t, s, "c.example.com", []byte("PEM-C"))
	m, clk := armedMirror(t, s)
	m.local = local
	mirrorPoll(t, m)

	renewed := hostCertPEM(t, "b.example.com", time.Now().Add(90*24*time.Hour))
	putBucket(t, s, "b.example.com", renewed)
	s.reset()
	clk.advance(certMirrorInterval)
	mirrorPoll(t, m)
	l, le, g := certReads(s)
	if l != 0 || le != 1 || g != 1 {
		t.Fatalf("one rewritten cert: List %d, ListETags %d, Get %d; want 0, 1, 1", l, le, g)
	}
	if n := s.count("Get", certKey("b.example.com")); n != 1 {
		t.Fatalf("the one Get must be the rewritten cert's, got %d of it", n)
	}
	if b, ok := m.Cert("b.example.com"); !ok || !bytes.Equal(b, renewed) {
		t.Fatal("the rewritten cert must be served from the new snapshot")
	}
	for _, name := range []string{"a.example.com", "c.example.com"} {
		if _, ok := m.Cert(name); !ok {
			t.Fatalf("%s did not move and must stay in the snapshot", name)
		}
	}
	if got, err := localCert(t, local, "b.example.com"); err != nil || !bytes.Equal(got, renewed) {
		t.Fatalf("a changed cert must still be copied into the local cache: %v", err)
	}
	if m.snapshot().certs["b.example.com"] == nil {
		t.Fatal("the changed cert must be parsed into the new snapshot")
	}
}

func TestCertMirrorDropsRemovedKey(t *testing.T) {
	t.Parallel()
	const token = "0123456789abcdefghij0123456789abcdefghijkl"
	s := newChangeIndexStore(t)
	putBucket(t, s, "example.com", []byte("PEM"))
	putBucket(t, s, token+httpTokenSuffix, []byte("resp"))
	m, clk := armedMirror(t, s)
	mirrorPoll(t, m)
	if _, ok := m.Challenge(token); !ok {
		t.Fatal("precondition: the live token is mirrored")
	}
	if err := s.Delete(context.Background(), certKey(token+httpTokenSuffix)); err != nil {
		t.Fatal(err)
	}
	s.reset()
	clk.advance(certMirrorInterval)
	mirrorPoll(t, m)
	if _, ok := m.Challenge(token); ok {
		t.Fatal("a spent challenge token must leave the snapshot when its key leaves the listing")
	}
	if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM" {
		t.Fatalf("the unmoved cert = %q, %v — want it kept", b, ok)
	}
	if l, le, g := certReads(s); l != 0 || le != 1 || g != 0 {
		t.Fatalf("a deletion: List %d, ListETags %d, Get %d; want 0, 1, 0", l, le, g)
	}
}

func TestCertMirrorFullRebuildAtBackstop(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putBucket(t, s, "example.com", []byte("PEM"))
	putBucket(t, s, "shop.example.com", []byte("PEM2"))
	m, clk := armedMirror(t, s)
	mirrorPoll(t, m)

	// Inside the jitter's lower bound (-10%): still trusted.
	s.reset()
	clk.advance(release.ChangeIndexBackstop*9/10 - time.Minute)
	mirrorPoll(t, m)
	if l, le, g := certReads(s); l+le+g != 0 {
		t.Fatalf("a young snapshot with the index unmoved: List %d, ListETags %d, Get %d; want none", l, le, g)
	}

	// Past the jitter's upper bound (+10%): a full rebuild, index unmoved.
	s.reset()
	clk.advance(release.ChangeIndexBackstop*2/10 + 2*time.Minute)
	mirrorPoll(t, m)
	if l, le, g := certReads(s); l != 1 || le != 0 || g != 2 {
		t.Fatalf("at the backstop: List %d, ListETags %d, Get %d; want the full rebuild 1, 0, 2", l, le, g)
	}

	// The rebuild resets the age: the next unmoved poll reads nothing.
	s.reset()
	clk.advance(certMirrorInterval)
	mirrorPoll(t, m)
	if l, le, g := certReads(s); l+le+g != 0 {
		t.Fatalf("after the backstop rebuild: List %d, ListETags %d, Get %d; want none", l, le, g)
	}
}

func TestCertMirrorBootRefreshUnchanged(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putBucket(t, s, "example.com", []byte("PEM"))
	putBucket(t, s, "shop.example.com", []byte("PEM2"))
	m, clk := armedMirror(t, s)
	// The agent's tick has already polled the shared watch.
	if _, err := m.watch().Poll(context.Background()); err != nil {
		t.Fatal(err)
	}
	s.reset()
	// The boot path: a full read, whatever the index says, and twice over.
	for i := 0; i < 2; i++ {
		if err := m.Refresh(context.Background()); err != nil {
			t.Fatal(err)
		}
	}
	if l, le, g := certReads(s); l != 2 || le != 0 || g != 4 {
		t.Fatalf("two boot refreshes: List %d, ListETags %d, Get %d; want 2, 0, 4", l, le, g)
	}
	if n := s.count("GetIfChanged", ""); n != 0 {
		t.Fatalf("the boot refresh must not poll the index (%d reads): it runs before the listeners bind", n)
	}
	// The loop's first poll trusts the boot read.
	s.reset()
	clk.advance(certMirrorInterval)
	mirrorPoll(t, m)
	if l, le, g := certReads(s); l+le+g != 0 {
		t.Fatalf("first poll after boot, index unmoved: List %d, ListETags %d, Get %d; want none", l, le, g)
	}
}

// Disarmed (no watch handed down, or the agent never armed one), every poll
// is today's full Refresh.
func TestCertMirrorDisarmedPollIsAFullRefresh(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putBucket(t, s, "example.com", []byte("PEM"))
	for _, watch := range []func() *release.ChangeWatch{nil, func() *release.ChangeWatch { return nil }} {
		m := newCertMirror(s, nil)
		m.watch = watch
		for i := 0; i < 2; i++ {
			s.reset()
			mirrorPoll(t, m)
			if l, le, g := certReads(s); l != 1 || le != 0 || g != 1 {
				t.Fatalf("disarmed poll %d: List %d, ListETags %d, Get %d; want 1, 0, 1", i, l, le, g)
			}
			if n := s.count("GetIfChanged", ""); n != 0 {
				t.Fatalf("disarmed poll read the index %d time(s)", n)
			}
		}
	}
}

// A changed cert whose Get fails keeps its previous copy (degrade, don't
// die), and the next poll reads it again with the index unmoved.
func TestCertMirrorFailedGetOfAChangedCertKeepsThePreviousCopy(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putBucket(t, s, "example.com", []byte("PEM"))
	putBucket(t, s, "shop.example.com", []byte("PEM2"))
	m, clk := armedMirror(t, s)
	mirrorPoll(t, m)

	putBucket(t, s, "example.com", []byte("PEM-RENEWED"))
	putBucket(t, s, "new.example.com", []byte("PEM3"))
	s.setFailGet(certKey("example.com"))
	clk.advance(certMirrorInterval)
	mirrorPoll(t, m)
	if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM" {
		t.Fatalf("a failed Get of a changed cert: mirror = %q, %v — want the previous copy", b, ok)
	}
	if b, ok := m.Cert("new.example.com"); !ok || string(b) != "PEM3" {
		t.Fatalf("the other changed cert = %q, %v — want it landed", b, ok)
	}

	s.setFailGet("")
	s.reset()
	clk.advance(certMirrorInterval)
	mirrorPoll(t, m)
	if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM-RENEWED" {
		t.Fatalf("the retry: mirror = %q, %v — want the renewed copy", b, ok)
	}
	if l, le, g := certReads(s); l != 0 || le != 1 || g != 1 {
		t.Fatalf("the retry: List %d, ListETags %d, Get %d; want 0, 1, 1", l, le, g)
	}
	s.reset()
	clk.advance(certMirrorInterval)
	mirrorPoll(t, m)
	if l, le, g := certReads(s); l+le+g != 0 {
		t.Fatalf("settled: List %d, ListETags %d, Get %d; want none", l, le, g)
	}
}

// The scopes the mirror and the home map gate on are the ones their prefixes'
// writes bump.
func TestCertMirrorAndHomeMapScopesMatchTheirPrefixes(t *testing.T) {
	t.Parallel()
	if got, ok := release.ChangeScope(certKey("example.com")); !ok || got != certScope {
		t.Fatalf("a cert write bumps %q (%v), the mirror gates on %q", got, ok, certScope)
	}
	if got, ok := release.ChangeScope(edgesPrefix + "box.json"); !ok || got != edgesScope {
		t.Fatalf("an edges-file write bumps %q (%v), the home map gates on %q", got, ok, edgesScope)
	}
}

// WithChangeWatch reaches both loops, resolved per poll: a watch the agent
// arms after the edge started is still seen.
func TestCertMirrorWithChangeWatchReachesTheMirrorAndHomeMap(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	var armed atomic.Pointer[release.ChangeWatch]
	e, err := New(openReg(t), nil)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(e.probers.Stop)
	e.staticStore = s
	WithChangeWatch(armed.Load)(e)
	m, hm := e.newCertMirror(nil), e.newHomeMap()
	if m.changeWatch() != nil || hm.changeWatch() != nil {
		t.Fatal("nothing armed yet: both loops must be disarmed")
	}
	w := release.NewChangeWatch(s)
	armed.Store(w)
	if m.changeWatch() != w || hm.changeWatch() != w {
		t.Fatal("a watch armed after the edge started must reach both loops")
	}
}

// ONE bad object is not a bad poll, on the changed-only read too: a changed
// cert that cannot be read keeps its previous copy, a token deleted in the
// same window still leaves, and the poll succeeds (so the loop's backfill
// runs), on every poll until the object reads again.
func TestCertMirrorOneUnreadableChangedKeyIsNotABadPoll(t *testing.T) {
	t.Parallel()
	const token = "0123456789abcdefghij0123456789abcdefghijkl"
	s := newChangeIndexStore(t)
	putBucket(t, s, "example.com", []byte("PEM"))
	putBucket(t, s, token+httpTokenSuffix, []byte("resp"))
	m, clk := armedMirror(t, s)
	mirrorPoll(t, m)

	putBucket(t, s, "example.com", []byte("PEM-RENEWED"))
	s.setFailGet(certKey("example.com"))
	if err := s.Delete(context.Background(), certKey(token+httpTokenSuffix)); err != nil {
		t.Fatal(err)
	}
	for i := 0; i < 2; i++ {
		s.reset()
		clk.advance(certMirrorInterval)
		if err := m.poll(context.Background()); err != nil {
			t.Fatalf("poll %d with one unreadable changed object: %v — want a successful partial poll", i, err)
		}
		if _, ok := m.Challenge(token); ok {
			t.Fatalf("poll %d: the deleted token must leave even while another object is unreadable", i)
		}
		if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM" {
			t.Fatalf("poll %d: the unreadable cert = %q, %v — want the previous copy", i, b, ok)
		}
		if n := s.count("Get", certKey("example.com")); n != 1 {
			t.Fatalf("poll %d: %d Get(s) of the unreadable cert, want it retried once", i, n)
		}
	}
}

// A hung index GET elsewhere in the process (the agent's tick holding the
// shared watch's poll lock) costs the mirror its index read, not its poll:
// it gives up on the watch and reads in full.
func TestCertMirrorPollsPastAStuckWatch(t *testing.T) {
	t.Parallel()
	s := newChangeIndexStore(t)
	putBucket(t, s, "example.com", []byte("PEM"))
	m, clk := armedMirror(t, s)
	m.watchTimeout = 50 * time.Millisecond
	mirrorPoll(t, m)

	entered, release := s.blockIndex(t)
	go func() { _, _ = m.watch().Poll(context.Background()) }()
	<-entered
	putBucket(t, s, "example.com", []byte("PEM-RENEWED"))
	clk.advance(certMirrorInterval)
	done := make(chan error, 1)
	go func() { done <- m.poll(context.Background()) }()
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("poll: %v", err)
		}
	case <-time.After(5 * time.Second):
		release()
		t.Fatal("the mirror's poll waited on a stuck watch")
	}
	if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM-RENEWED" {
		t.Fatalf("a poll that could not read the index must read in full: mirror = %q, %v", b, ok)
	}
}