package hostactors

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/boxid"
	"github.com/carlosframework/platform/internal/boxid/boxidtest"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/status"
)

// enrollWorld is plan9Fixture's published deployment (boxes a, b and w,
// active in ie under a published host floor) with ie's group, and a box
// (instance i-new, reference newRef) that the group launched: its
// identity record is published, and nothing else. The console reads the
// cloud through cloud.
type enrollWorld struct {
	plan9World
	e      *AutoEnroller
	cloud  *fakeMembership
	signer *boxidtest.Signer
	clock  time.Time
	logs   []string
}

func newEnrollWorld(t *testing.T, role string) *enrollWorld {
	t.Helper()
	w, _ := plan9Fixture(t, func(r *Roster) {
		r.Regions = map[string]Region{"ie": {Account: "123", AWSRegion: "eu-west-1", GroupName: "routers-ie",
			GroupARN: "arn:aws:autoscaling:eu-west-1:123:autoScalingGroup:x:autoScalingGroupName/routers-ie", Namespaces: []string{"platform"}, Pools: []string{"default"}, Role: role}}
	})
	signer := boxidtest.NewSigner(t)
	_, raw := signer.Record(t, newRef, "123", "eu-west-1", "i-new")
	if _, err := w.d.Store.Put(context.Background(), boxid.Key(newRef), raw, ""); err != nil {
		t.Fatal(err)
	}
	ew := &enrollWorld{plan9World: w, clock: time.Now().UTC(), signer: signer,
		cloud: &fakeMembership{members: map[string]Membership{"i-new": member("routers-ie")}}}
	ew.e = &AutoEnroller{Directory: w.d, Signer: w.sg, Evidence: ExactEvidence{PublicKey: w.d.PublicKey, Store: w.d.Store, Boxes: w.d.PrivateBoxReport},
		Members: ew.cloud, Certs: signer.Certs(), Namespace: "platform", Pool: "default", Now: func() time.Time { return ew.clock }, Patience: 2 * time.Minute,
		Logf: func(format string, args ...any) { ew.logs = append(ew.logs, fmt.Sprintf(format, args...)) }}
	return ew
}

// pass runs the enrollment pass and returns new's step.
func (w *enrollWorld) pass(t *testing.T) string {
	t.Helper()
	steps, err := w.e.EnrollPass(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range steps {
		if s.Box == newRef {
			return s.Step
		}
	}
	return ""
}

// guardReady is the joining box, its guard armed, publishing the report
// every hosting box publishes.
func (w *enrollWorld) guardReady(t *testing.T, enrollment string, ready bool) {
	t.Helper()
	box := status.Box{GeneratedAt: time.Now().UTC(), Metrics: &status.BoxMetrics{Enrollment: enrollment, InstanceID: "i-new", Release: "v1.0.0", ReleaseTag: "v1.0.0",
		Platform: "linux-amd64", BinarySHA256: w.hash, MovesProtocol: 1, BootID: "boot", FloorSHA256: w.floor.Digest, FloorGuardReady: ready, SystemChannel: "stable"}}
	raw, _ := json.Marshal(box)
	_, etag, _ := w.d.Store.Get(context.Background(), "status/boxes/"+newRef+".json")
	if _, err := w.d.Store.Put(context.Background(), "status/boxes/"+newRef+".json", raw, etag); err != nil {
		t.Fatal(err)
	}
}

func (w *enrollWorld) invitation(t *testing.T) (Invitation, bool) {
	t.Helper()
	ctx := context.Background()
	wire, _, err := w.d.Store.Get(ctx, "control/host-actor-invitations/"+newRef+".json")
	if errors.Is(err, os.ErrNotExist) {
		return Invitation{}, false
	}
	if err != nil {
		t.Fatal(err)
	}
	var got Invitation
	if err := release.VerifyRootEnvelope(w.d.PublicKey, wire, &got); err != nil {
		t.Fatal(err)
	}
	// Read only, never Invite (get-or-create): the invitation verifies
	// for the box's record, and nothing is written.
	raw, err := boxid.Load(ctx, w.d.Store, newRef)
	if err != nil {
		t.Fatal(err)
	}
	nonce, err := w.d.enrollmentNonce(ctx, EnrollmentRequest{Reference: newRef, Namespace: "platform"}, raw)
	if err != nil || nonce != got.Enrollment {
		t.Fatalf("invitation %+v: %v", got, err)
	}
	return got, true
}

func (w *enrollWorld) attempt(t *testing.T) EnrollmentAttempt {
	t.Helper()
	o, err := w.d.Read(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	return o.Roster.EnrollmentAttempts[newRef]
}

// The whole sequence (spec 2 §5 step 5, §11): attempt, the console's own
// membership read and the invitation, guard-ready, the console's second
// read and the enrollment. The deployment's floor is published, so the
// enrolling write passes membership.go's rule only because the guard-ready
// report is there; the box takes its group's role in it.
func TestAutomaticEnrollmentFromRecordToRoster(t *testing.T) {
	w := newEnrollWorld(t, RoleLegacy)
	if got := w.pass(t); got != "began an attempt" {
		t.Fatalf("first pass: %q", got)
	}
	if _, ok := w.invitation(t); ok || len(w.cloud.calls) != 0 {
		t.Fatalf("invited before membership was read: calls %v", w.cloud.calls)
	}
	if got := w.pass(t); got != "invited" {
		t.Fatalf("second pass: %q", got)
	}
	inv, ok := w.invitation(t)
	if !ok || len(w.cloud.calls) != 1 {
		t.Fatalf("invitation %v, calls %v", ok, w.cloud.calls)
	}
	// Invited, but the guard is not armed: nothing more, and no read.
	// First no report at all, then a report from a box whose guard is not
	// ready.
	if got := w.pass(t); got != "waiting for its guard-ready report" {
		t.Fatalf("invited, no report: %q", got)
	}
	w.guardReady(t, inv.Enrollment, false)
	if got := w.pass(t); got != "waiting for its guard-ready report" || len(w.cloud.calls) != 1 {
		t.Fatalf("invited, guard not ready: %q, calls %v", got, w.cloud.calls)
	}
	w.guardReady(t, inv.Enrollment, true)
	w.clock = w.clock.Add(5 * time.Minute)
	if got := w.pass(t); !strings.HasPrefix(got, "enrolled in ie") {
		t.Fatalf("ready: %q", got)
	}
	// The enrollment read membership again, itself.
	if len(w.cloud.calls) != 2 || len(w.logs) != 2 || !strings.Contains(w.logs[1], "attempt "+w.attempt(t).AttemptID+", instance i-new, request ids asg-req, ec2-req") || !strings.HasSuffix(w.logs[1], "): enrolled") {
		t.Fatalf("calls %v, logs %q", w.cloud.calls, w.logs)
	}
	o, err := w.d.Read(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	b, ok := o.Roster.Boxes[newRef]
	if !ok || b.State != "active" || b.Enrollment != inv.Enrollment || b.Cloud.InstanceID != "i-new" || b.Group != "routers-ie" || b.Role != RoleLegacy || b.Namespace != "platform" || b.Pool != "default" {
		t.Fatalf("roster entry %+v", b)
	}
	if !o.Roster.EnrollmentAttempts[newRef].Consumed {
		t.Fatal("the attempt was not consumed")
	}
	// Enrolled: the pass has nothing left to do for it.
	if got := w.pass(t); got != "" {
		t.Fatalf("enrolled box stepped again: %q", got)
	}
}

// R1: an instance in the deployment's account that the cloud does not
// place in the region's group is never invited and never enrolled, however
// long it waits; and one that leaves the group after its invitation is not
// enrolled on a guard-ready report.
func TestAnInstanceOutsideTheGroupIsNeverInvitedOrEnrolled(t *testing.T) {
	for name, m := range map[string]Membership{
		"another group": member("routers-uk"),
		"no group":      {Owner: "123", InstanceState: "running", AvailabilityZoneRegion: "eu-west-1", RequestIDs: []string{"a", "b"}},
	} {
		t.Run(name, func(t *testing.T) {
			w := newEnrollWorld(t, "")
			w.cloud.members["i-new"] = m
			for i := 0; i < 12; i++ {
				w.pass(t)
				w.clock = w.clock.Add(time.Minute)
			}
			if _, ok := w.invitation(t); ok {
				t.Fatal("an instance outside the group was invited")
			}
			if o, _ := w.d.Read(context.Background()); o.Roster.Boxes[newRef].Enrollment != "" {
				t.Fatal("an instance outside the group was enrolled")
			}
		})
	}
	w := newEnrollWorld(t, "")
	w.pass(t)
	if got := w.pass(t); got != "invited" {
		t.Fatal(got)
	}
	inv, _ := w.invitation(t)
	w.guardReady(t, inv.Enrollment, true)
	w.cloud.members["i-new"] = func() Membership { m := member("routers-ie"); m.LifecycleState = "Detaching"; return m }()
	if got := w.pass(t); !strings.Contains(got, "not confirmed") {
		t.Fatalf("left the group: %q", got)
	}
	if o, _ := w.d.Read(context.Background()); o.Roster.Boxes[newRef].Enrollment != "" {
		t.Fatal("an instance that left the group was enrolled")
	}
}

// R1 (#7): a read that fails is unknown. The attempt stands, its wait does
// not grow, and the next read that answers goes on from it.
func TestAMembershipReadThatFailsKeepsTheAttempt(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.pass(t)
	first := w.attempt(t).AttemptID
	w.cloud.err = errors.New("Throttling: rate exceeded")
	for i := 0; i < 10; i++ {
		w.clock = w.clock.Add(time.Hour)
		if got := w.pass(t); !strings.Contains(got, "could not be read") {
			t.Fatalf("failed read: %q", got)
		}
	}
	if got := w.attempt(t).AttemptID; got != first || w.e.attempts[newRef].tries != 1 {
		t.Fatalf("a failed read superseded the attempt (%s, was %s) or grew its wait (%d)", got, first, w.e.attempts[newRef].tries)
	}
	w.cloud.err = nil
	if got := w.pass(t); got != "invited" || w.attempt(t).AttemptID != first {
		t.Fatalf("after the outage: %q", got)
	}
}

// stalledMembership is a describe whose connection stalls for one
// instance: Member returns only when its context ends. Every other
// instance is answered by the cloud underneath.
type stalledMembership struct {
	stalled  string
	cloud    *fakeMembership
	deadline bool
}

func (s *stalledMembership) Member(ctx context.Context, awsRegion, instance string) (Membership, error) {
	if instance != s.stalled {
		return s.cloud.Member(ctx, awsRegion, instance)
	}
	_, s.deadline = ctx.Deadline()
	<-ctx.Done()
	return Membership{}, ctx.Err()
}

// A membership read that stalls is cut off at storeCallTimeout and is
// unknown: the pass returns within the bound, the stalled box's attempt
// stands with its wait unchanged, and the box after it in the same pass is
// still invited.
func TestAStalledMembershipReadIsCutOffAndThePassGoesOn(t *testing.T) {
	was := storeCallTimeout
	storeCallTimeout = 100 * time.Millisecond
	t.Cleanup(func() { storeCallTimeout = was })
	w := newEnrollWorld(t, "")
	// A second box the group launched, whose reference sorts after
	// newRef's, so it waits in the pass behind the stalled read.
	twoRef := GroupBoxReference("123", "i-two")
	if twoRef <= newRef {
		t.Fatalf("%s does not sort after %s", twoRef, newRef)
	}
	_, raw := w.signer.Record(t, twoRef, "123", "eu-west-1", "i-two")
	if _, err := w.d.Store.Put(context.Background(), boxid.Key(twoRef), raw, ""); err != nil {
		t.Fatal(err)
	}
	w.cloud.members["i-two"] = member("routers-ie")
	stalled := &stalledMembership{stalled: "i-new", cloud: w.cloud}
	w.e.Members = stalled
	steps, err := w.e.EnrollPass(context.Background())
	if err != nil || len(steps) != 2 || steps[0].Step != "began an attempt" || steps[1].Step != "began an attempt" {
		t.Fatalf("first pass: %v, %v", steps, err)
	}
	first := w.attempt(t).AttemptID

	done := make(chan []EnrollStep, 1)
	start := time.Now()
	go func() {
		steps, err := w.e.EnrollPass(context.Background())
		if err != nil {
			t.Error(err)
		}
		done <- steps
	}()
	select {
	case steps = <-done:
	case <-time.After(10 * time.Second):
		t.Fatal("the pass is still waiting on the stalled read")
	}
	if took := time.Since(start); took > storeCallTimeout+5*time.Second {
		t.Fatalf("the pass took %s", took)
	}
	if !stalled.deadline {
		t.Fatal("the read was given no deadline")
	}
	if len(steps) != 2 || steps[0].Box != newRef || !strings.Contains(steps[0].Step, "could not be read") || !strings.Contains(steps[0].Step, "did not finish in time") {
		t.Fatalf("stalled box: %v", steps)
	}
	if got := w.attempt(t).AttemptID; got != first || w.e.attempts[newRef].tries != 1 || w.e.attempts[newRef].refused != "" {
		t.Fatalf("the stalled read superseded the attempt (%s, was %s), grew its wait (%d) or refused it (%q)", got, first, w.e.attempts[newRef].tries, w.e.attempts[newRef].refused)
	}
	if _, ok := w.invitation(t); ok {
		t.Fatal("the stalled box was invited")
	}
	if steps[1].Box != twoRef || steps[1].Step != "invited" {
		t.Fatalf("the box behind the stalled read: %v", steps[1])
	}
	if _, _, err := w.d.Store.Get(context.Background(), "control/host-actor-invitations/"+twoRef+".json"); err != nil {
		t.Fatalf("the box behind the stalled read has no invitation: %v", err)
	}
}

// C2: the invitation is written only while the console's read is under a
// minute old. Here every look at the clock takes 31 seconds: the read is
// fresh when checkConfirmed has it, and over a minute old by the time the
// invitation would be written. That is unknown: the attempt and its wait
// stand.
func TestAnInvitationNeedsAReadUnderAMinuteOldAtTheWrite(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.pass(t)
	first := w.attempt(t).AttemptID
	w.e.Now = func() time.Time { w.clock = w.clock.Add(31 * time.Second); return w.clock }
	if got := w.pass(t); !strings.Contains(got, "more than a minute after its membership read") {
		t.Fatalf("slow pass: %q", got)
	}
	if _, ok := w.invitation(t); ok {
		t.Fatal("invited on a read over a minute old")
	}
	if w.attempt(t).AttemptID != first || w.e.attempts[newRef].tries != 1 || w.e.attempts[newRef].refused != "" {
		t.Fatalf("a missed deadline counted against the attempt: %+v", w.e.attempts[newRef])
	}
}

// slowStore is the deployment bucket with one slow read: the first Get of
// slow after armed is set moves the console's clock on by exactly 60
// seconds, to the deadline itself, as a bucket that answers late would. It honours its context, as the S3
// store does, and records every Put.
type slowStore struct {
	release.Store
	slow  string
	armed bool
	late  func()
	puts  []string
}

func (s *slowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	if err := ctx.Err(); err != nil {
		return nil, "", err
	}
	if s.armed && key == s.slow {
		s.armed = false
		s.late()
	}
	return s.Store.Get(ctx, key)
}

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

// slowAfterRead makes the store slow on key after the console's next
// membership read.
func (w *enrollWorld) slowAfterRead(key string) *slowStore {
	ss := &slowStore{Store: w.e.Directory.Store, slow: key, late: func() { w.clock = w.clock.Add(60 * time.Second) }}
	w.e.Directory.Store = ss
	w.cloud.during = func() { ss.armed = true }
	return ss
}

// S1: a read inside the sealed Invite that answers late (here, its own
// read of the invitation key) leaves the invitation unwritten: no Put is
// issued after the deadline. The attempt and its wait stand, and the next
// pass, with a fresh read, invites.
func TestASlowReadInsideInviteWritesNothing(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.pass(t)
	first := w.attempt(t).AttemptID
	ss := w.slowAfterRead("control/host-actor-invitations/" + newRef + ".json")
	if got := w.pass(t); !strings.Contains(got, "more than a minute after its membership read") {
		t.Fatalf("slow invite: %q", got)
	}
	if len(ss.puts) != 0 {
		t.Fatalf("written after the deadline: %v", ss.puts)
	}
	if _, ok := w.invitation(t); ok || w.attempt(t).AttemptID != first || w.e.attempts[newRef].tries != 1 {
		t.Fatalf("invited %v, attempt %+v", ok, w.e.attempts[newRef])
	}
	w.cloud.during = nil
	if got := w.pass(t); got != "invited" || w.attempt(t).AttemptID != first {
		t.Fatalf("next pass: %q", got)
	}
}

// S1: a read inside the sealed checkMembership that answers late (the
// guard-ready report Replace checks before its CAS) leaves the roster
// unwritten. The attempt stays unconsumed, and the next pass enrolls.
func TestASlowReadInsideTheRosterCheckWritesNothing(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.pass(t)
	if got := w.pass(t); got != "invited" {
		t.Fatal(got)
	}
	inv, _ := w.invitation(t)
	w.guardReady(t, inv.Enrollment, true)
	first, tries := w.attempt(t).AttemptID, w.e.attempts[newRef].tries
	ss := w.slowAfterRead("status/boxes/" + newRef + ".json")
	if got := w.pass(t); !strings.Contains(got, "more than a minute after its membership read") {
		t.Fatalf("slow enrolment: %q", got)
	}
	if len(ss.puts) != 0 {
		t.Fatalf("written after the deadline: %v", ss.puts)
	}
	if a := w.attempt(t); a.AttemptID != first || a.Consumed || w.e.attempts[newRef].tries != tries || w.e.attempts[newRef].refused != "" {
		t.Fatalf("attempt %+v, seen %+v", a, w.e.attempts[newRef])
	}
	if o, _ := w.d.Read(context.Background()); o.Roster.Boxes[newRef].Enrollment != "" {
		t.Fatal("enrolled after the deadline")
	}
	w.cloud.during = nil
	w.guardReady(t, inv.Enrollment, true)
	if got := w.pass(t); !strings.HasPrefix(got, "enrolled in ie") {
		t.Fatalf("next pass: %q", got)
	}
}

// The ruling's second test: a box cannot satisfy another box's report
// check. A report under newRef's key that another instance wrote is not
// newRef's guard-ready report, and a record under a key that is not its
// instance's group reference is never invited.
func TestABoxCannotStandInForAnother(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.pass(t)
	if got := w.pass(t); got != "invited" {
		t.Fatal(got)
	}
	inv, _ := w.invitation(t)
	box := status.Box{GeneratedAt: time.Now().UTC(), Metrics: &status.BoxMetrics{Enrollment: inv.Enrollment, InstanceID: "i-other", Release: "v1.0.0", ReleaseTag: "v1.0.0",
		Platform: "linux-amd64", BinarySHA256: w.hash, MovesProtocol: 1, BootID: "boot", FloorSHA256: w.floor.Digest, FloorGuardReady: true, SystemChannel: "stable"}}
	raw, _ := json.Marshal(box)
	if _, err := w.d.Store.Put(context.Background(), "status/boxes/"+newRef+".json", raw, ""); err != nil {
		t.Fatal(err)
	}
	if got := w.pass(t); got != "waiting for its guard-ready report" {
		t.Fatalf("another instance's report: %q", got)
	}
	// A record its box published under a key of its own choosing.
	v := newEnrollWorld(t, "")
	_, chosen := v.signer.Record(t, "chosen", "123", "eu-west-1", "i-chosen")
	if _, err := v.d.Store.Put(context.Background(), boxid.Key("chosen"), chosen, ""); err != nil {
		t.Fatal(err)
	}
	steps, err := v.e.EnrollPass(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range steps {
		if s.Box == "chosen" && !strings.Contains(s.Step, "not under its group reference") {
			t.Fatalf("chosen key: %q", s.Step)
		}
	}
	if _, _, err := v.d.Store.Get(context.Background(), "control/host-actor-invitations/chosen.json"); err == nil {
		t.Fatal("a box with a chosen key was invited")
	}
}

// inflightStore is the bucket with one slow Put: the first Put of key
// after armed is set commits (when commit is set) and then, by mode,
// waits for its context as an S3 request cut off by the deadline does
// ("cancelled"), ignores its context and never answers ("ignores"), or
// answers success only after the console's clock has passed the deadline
// ("late"). In mode "delayed" it commits nothing until hang is closed,
// whatever its context says, then commits and reports on done: a write
// the console gave up on that the bucket applies later.
type inflightStore struct {
	release.Store
	key           string
	armed, commit bool
	mode          string
	late          func()
	hang          chan struct{}
	done          chan error
}

func (s *inflightStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if s.armed && key == s.key {
		s.armed = false
		if s.mode == "delayed" {
			<-s.hang
			etag, err := s.Store.Put(ctx, key, body, ifMatch)
			s.done <- err
			return etag, err
		}
		etag := ""
		if s.commit {
			var err error
			if etag, err = s.Store.Put(ctx, key, body, ifMatch); err != nil {
				return "", err
			}
		}
		switch s.mode {
		case "ignores":
			<-s.hang
			return etag, nil
		case "late":
			s.late()
			return etag, nil
		}
		<-ctx.Done()
		return "", ctx.Err()
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// inflightAfterRead makes key's next Put slow in mode: the next
// membership read takes 59.95 seconds, so its write has 50 milliseconds,
// and a late answer comes a second after that.
func (w *enrollWorld) inflightAfterRead(key string, commit bool, mode string) *inflightStore {
	is := &inflightStore{Store: w.e.Directory.Store, key: key, commit: commit, mode: mode, hang: make(chan struct{}), done: make(chan error, 1),
		late: func() { w.clock = w.clock.Add(time.Second) }}
	w.e.Directory.Store = is
	w.cloud.during = func() { w.clock = w.clock.Add(59950 * time.Millisecond); is.armed = true }
	return is
}

// S1: a Put still in flight at the deadline is cancelled there, and its
// result is unknown: the attempt and its wait stand, and the next pass
// reads the bucket as it does every pass. Here the invitation did not
// land, then (a second time) it did, and the next pass finds it.
func TestAPutInFlightAtTheDeadlineIsUnknown(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.pass(t)
	first, tries := w.attempt(t).AttemptID, w.e.attempts[newRef].tries
	w.inflightAfterRead("control/host-actor-invitations/"+newRef+".json", false, "cancelled")
	if got := w.pass(t); !strings.Contains(got, "in flight at the deadline") {
		t.Fatalf("in flight: %q", got)
	}
	if _, ok := w.invitation(t); ok || w.attempt(t).AttemptID != first || w.e.attempts[newRef].tries != tries || w.e.attempts[newRef].refused != "" {
		t.Fatalf("invited %v, seen %+v", ok, w.e.attempts[newRef])
	}
	w.inflightAfterRead("control/host-actor-invitations/"+newRef+".json", true, "cancelled")
	if got := w.pass(t); !strings.Contains(got, "in flight at the deadline") {
		t.Fatalf("in flight, landed: %q", got)
	}
	w.cloud.during = nil
	if got := w.pass(t); got != "waiting for its guard-ready report" {
		t.Fatalf("read back: %q", got)
	}
}

// U2: a store that ignores its context and never answers is cut off by
// the hard per-call timeout, well before the deadline: the outcome is
// unknown, the attempt and its wait stand, and nothing was invited.
func TestAStoreThatIgnoresItsContextIsCutOffAtTheHardTimeout(t *testing.T) {
	was := storeCallTimeout
	storeCallTimeout = 100 * time.Millisecond
	t.Cleanup(func() { storeCallTimeout = was })
	w := newEnrollWorld(t, "")
	w.pass(t)
	first, tries := w.attempt(t).AttemptID, w.e.attempts[newRef].tries
	is := &inflightStore{Store: w.e.Directory.Store, key: "control/host-actor-invitations/" + newRef + ".json", mode: "ignores", hang: make(chan struct{}), armed: true}
	t.Cleanup(func() { close(is.hang) })
	w.e.Directory.Store = is
	start := time.Now()
	got := w.pass(t)
	if !strings.Contains(got, "did not answer in time") || time.Since(start) > 5*time.Second {
		t.Fatalf("after %s: %q", time.Since(start), got)
	}
	if _, ok := w.invitation(t); ok || w.attempt(t).AttemptID != first || w.e.attempts[newRef].tries != tries || w.e.attempts[newRef].refused != "" {
		t.Fatalf("invited %v, seen %+v", ok, w.e.attempts[newRef])
	}
}

// U1, U2: an enrolling write whose outcome is unknown is settled by the
// next pass's roster read, with nothing more. Not landed: the same
// attempt, its wait unchanged, enrols on the next pass. Landed, whether
// cut off in flight or answering success only after the deadline: the
// roster lists the box, and the pass has nothing left to do for it.
func TestAnUnknownEnrolmentIsSettledByTheRoster(t *testing.T) {
	ready := func(t *testing.T) *enrollWorld {
		w := newEnrollWorld(t, "")
		w.pass(t)
		if got := w.pass(t); got != "invited" {
			t.Fatal(got)
		}
		inv, _ := w.invitation(t)
		w.guardReady(t, inv.Enrollment, true)
		return w
	}
	t.Run("did not land", func(t *testing.T) {
		w := ready(t)
		first, tries := w.attempt(t).AttemptID, w.e.attempts[newRef].tries
		w.inflightAfterRead(Key, false, "cancelled")
		if got := w.pass(t); !strings.Contains(got, "in flight at the deadline") {
			t.Fatalf("in flight: %q", got)
		}
		if a := w.attempt(t); a.Consumed || a.AttemptID != first || w.e.attempts[newRef].tries != tries {
			t.Fatalf("attempt %+v, seen %+v", a, w.e.attempts[newRef])
		}
		w.cloud.during = nil
		if got := w.pass(t); !strings.HasPrefix(got, "enrolled in ie") {
			t.Fatalf("next pass: %q", got)
		}
	})
	for name, mode := range map[string]string{"landed, cut off in flight": "cancelled", "landed, answered after the deadline": "late"} {
		t.Run(name, func(t *testing.T) {
			w := ready(t)
			w.inflightAfterRead(Key, true, mode)
			want := map[string]string{"cancelled": "in flight at the deadline", "late": "answered only after the deadline"}[mode]
			if got := w.pass(t); !strings.Contains(got, want) {
				t.Fatalf("unknown: %q", got)
			}
			w.cloud.during = nil
			calls := len(w.cloud.calls)
			steps, err := w.e.EnrollPass(context.Background())
			if err != nil || len(steps) != 0 || len(w.cloud.calls) != calls {
				t.Fatalf("settled by the roster: %+v %v, calls %v", steps, err, w.cloud.calls)
			}
			o, _ := w.d.Read(context.Background())
			if b := o.Roster.Boxes[newRef]; b.State != "active" || !o.Roster.EnrollmentAttempts[newRef].Consumed {
				t.Fatalf("roster %+v", b)
			}
		})
	}
}

// V1, the accepted exposure: the console waits for no write past its
// bound, but a write it gave up on can still commit. Here the enrolling
// Put is issued before the deadline and blocks; the pass returns unknown;
// the instance is detached from its group (still running); the next pass
// finds the box absent from the roster, and its fresh describe refuses
// without writing anything; then the bucket applies the delayed Put, on
// the roster revision it was built against, which nothing has moved. The
// pass after that finds the box enrolled and does nothing more. Such an
// enrolment rests on a membership read that was true when its write was
// issued: the same exposure as an instance detached after enrolment,
// which spec 3's regional worker reconciles.
func TestAWriteTheConsoleGaveUpOnCanCommitLater(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.pass(t)
	if got := w.pass(t); got != "invited" {
		t.Fatal(got)
	}
	inv, _ := w.invitation(t)
	w.guardReady(t, inv.Enrollment, true)
	before, _ := w.d.Read(context.Background())
	is := w.inflightAfterRead(Key, false, "delayed")
	if got := w.pass(t); !strings.Contains(got, "in flight at the deadline") {
		t.Fatalf("unknown: %q", got)
	}
	// Detached: the group no longer lists the instance, which still runs.
	w.cloud.during = nil
	w.cloud.members["i-new"] = member("")
	calls := len(w.cloud.calls)
	if got := w.pass(t); !strings.Contains(got, "not confirmed") || len(w.cloud.calls) != calls+1 {
		t.Fatalf("absent, then refused: %q, calls %v", got, w.cloud.calls)
	}
	if o, _ := w.d.Read(context.Background()); o.ETag != before.ETag {
		t.Fatalf("the refusing pass wrote the roster: %s, was %s", o.ETag, before.ETag)
	}
	close(is.hang)
	if err := <-is.done; err != nil {
		t.Fatalf("the delayed write did not commit: %v", err)
	}
	calls = len(w.cloud.calls)
	steps, err := w.e.EnrollPass(context.Background())
	if err != nil || len(steps) != 0 || len(w.cloud.calls) != calls {
		t.Fatalf("enrolled: %+v %v, calls %v", steps, err, w.cloud.calls)
	}
	o, _ := w.d.Read(context.Background())
	if b := o.Roster.Boxes[newRef]; b.State != "active" || o.Roster.Revision != before.Roster.Revision+1 {
		t.Fatalf("roster %+v, revision %d", b, o.Roster.Revision)
	}
}

// A refused attempt is kept, with no more reads, for its wait, then begun
// again; each refusal doubles the wait, up to 32 times Patience.
func TestARefusedAttemptIsBegunAgainWithADoublingWait(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.cloud.members["i-new"] = member("")
	if got := w.pass(t); got != "began an attempt" {
		t.Fatal(got)
	}
	first := w.attempt(t).AttemptID
	if got := w.pass(t); !strings.Contains(got, "not confirmed") {
		t.Fatal(got)
	}
	w.clock = w.clock.Add(time.Minute + 59*time.Second)
	if got := w.pass(t); !strings.Contains(got, "waiting before another attempt") || len(w.cloud.calls) != 1 {
		t.Fatalf("inside the wait: %q, calls %v", got, w.cloud.calls)
	}
	w.clock = w.clock.Add(time.Second)
	if got := w.pass(t); got != "began an attempt" || w.attempt(t).AttemptID == first {
		t.Fatalf("after two minutes: %q", got)
	}
	if got := w.pass(t); !strings.Contains(got, "not confirmed") {
		t.Fatal(got)
	}
	// The second refusal waits four minutes.
	w.clock = w.clock.Add(3*time.Minute + 59*time.Second)
	if got := w.pass(t); !strings.Contains(got, "waiting before another attempt") {
		t.Fatalf("second attempt before its four minutes: %q", got)
	}
	w.clock = w.clock.Add(time.Second)
	if got := w.pass(t); got != "began an attempt" {
		t.Fatalf("second attempt after four minutes: %q", got)
	}
	if got := w.e.wait(10); got != 64*time.Minute {
		t.Fatalf("the wait is not capped: %s", got)
	}
}

func TestAutomaticEnrollmentRefusesWhatItCannotProve(t *testing.T) {
	// A box from an account or region with no group is left alone.
	w := newEnrollWorld(t, "")
	_, raw := boxidtest.NewSigner(t).Record(t, "elsewhere", "999", "eu-west-1", "i-x")
	if _, err := w.d.Store.Put(context.Background(), boxid.Key("elsewhere"), raw, ""); err != nil {
		t.Fatal(err)
	}
	steps, err := w.e.EnrollPass(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range steps {
		if s.Box == "elsewhere" && !strings.Contains(s.Step, "no region's group") {
			t.Fatalf("elsewhere: %q", s.Step)
		}
	}
	// A record whose signature does not verify under the certificates.
	w.e.Certs = boxidtest.NewSigner(t).Certs()
	if got := w.pass(t); !strings.Contains(got, "does not verify") {
		t.Fatalf("unverified record: %q", got)
	}
	// A deployment with no published floor enrolls nothing.
	d, sg, _ := fixture(t)
	unpublished := &AutoEnroller{Directory: d, Signer: sg, Namespace: "platform", Pool: "default", Now: time.Now}
	if _, err := unpublished.EnrollPass(context.Background()); !errors.Is(err, ErrNoFloor) {
		t.Fatalf("unpublished: %v", err)
	}
}

// S5: every membership read is logged with the box, its attempt, its
// instance and the request ids the read's own confirmation kept, whatever
// came of it: a failed read, unknown (the ids the cloud gave, "none" for a
// call it never answered), a refusal, and an invitation.
func TestEveryMembershipReadIsLoggedWithItsIds(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.pass(t)
	id := w.attempt(t).AttemptID
	head := "membership read for " + newRef + " (attempt " + id + ", instance i-new, request ids "
	w.cloud.err, w.cloud.failIDs = errors.New("Throttling: rate exceeded"), []string{"asg-req", NoRequestID}
	w.pass(t)
	w.cloud.err, w.cloud.failIDs = nil, nil
	w.cloud.members["i-new"] = member("routers-uk")
	w.pass(t)
	if len(w.logs) != 2 || !strings.Contains(w.logs[0], head+"asg-req, none, read at ") || !strings.Contains(w.logs[0], "): unknown: "+ErrMembershipUnknown.Error()) ||
		!strings.Contains(w.logs[1], head+"asg-req, ec2-req, read at ") || !strings.Contains(w.logs[1], "): refused: "+ErrNotMember.Error()) {
		t.Fatalf("logs %q", w.logs)
	}
	// A check that failed before any read was made is not a membership
	// read, and is not logged as one.
	w.e.logRead(EnrollmentRequest{Reference: newRef, AttemptID: id}, "i-new", MembershipEvidence{}, "invited", ErrRoster)
	if len(w.logs) != 2 {
		t.Fatalf("a check with no read was logged: %q", w.logs[2:])
	}
	// A describe that took a minute is a refusal too.
	x := newEnrollWorld(t, "")
	x.pass(t)
	x.cloud.during = func() { x.clock = x.clock.Add(time.Minute) }
	if got := x.pass(t); !strings.Contains(got, "not confirmed") || len(x.logs) != 1 || !strings.Contains(x.logs[0], "): refused: "+ErrMembershipStale.Error()) {
		t.Fatalf("stale: %q, logs %q", got, x.logs)
	}
	v := newEnrollWorld(t, "")
	v.pass(t)
	if got := v.pass(t); got != "invited" {
		t.Fatal(got)
	}
	vhead := "membership read for " + newRef + " (attempt " + v.attempt(t).AttemptID + ", instance i-new, request ids asg-req, ec2-req, read at "
	if len(v.logs) != 1 || !strings.Contains(v.logs[0], vhead) || !strings.HasSuffix(v.logs[0], "): invited") {
		t.Fatalf("logs %q", v.logs)
	}
}

// conflictStore refuses the next Put of key with the store's own conflict,
// as a conditional write another writer got to first is refused.
type conflictStore struct {
	release.Store
	key   string
	armed bool
}

func (s *conflictStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if s.armed && key == s.key {
		s.armed = false
		return "", release.ErrCASConflict
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// Only a positive answer counts against an attempt. A store that refuses
// the invitation's or the enrolment's write with its own error (a
// conflict, which a call failing after the deadline also returns) is
// unknown: the attempt stands, its wait does not grow, and the next pass
// goes on from it.
func TestAStoreErrorIsUnknownAndKeepsTheAttempt(t *testing.T) {
	for _, key := range []string{"control/host-actor-invitations/" + newRef + ".json", Key} {
		t.Run(key, func(t *testing.T) {
			w := newEnrollWorld(t, "")
			w.pass(t)
			if key == Key {
				if got := w.pass(t); got != "invited" {
					t.Fatal(got)
				}
				inv, _ := w.invitation(t)
				w.guardReady(t, inv.Enrollment, true)
			}
			first, tries := w.attempt(t).AttemptID, w.e.attempts[newRef].tries
			w.e.Directory.Store = &conflictStore{Store: w.e.Directory.Store, key: key, armed: true}
			if got := w.pass(t); !strings.Contains(got, release.ErrCASConflict.Error()) {
				t.Fatalf("conflict: %q", got)
			}
			if a := w.attempt(t); a.AttemptID != first || a.Consumed || w.e.attempts[newRef].tries != tries || w.e.attempts[newRef].refused != "" {
				t.Fatalf("attempt %+v, seen %+v", a, w.e.attempts[newRef])
			}
			want := map[string]string{Key: "enrolled in ie"}[key]
			if want == "" {
				want = "invited"
			}
			if got := w.pass(t); !strings.HasPrefix(got, want) || w.attempt(t).AttemptID != first {
				t.Fatalf("next pass: %q", got)
			}
		})
	}
}

// vanishingStore answers the next Get of key, once armed, with the stored
// object and then removes it, as an invitation deleted just after the
// pass read it would be. It records each Put of key in events.
type vanishingStore struct {
	release.Store
	key    string
	armed  bool
	events *[]string
}

func (s *vanishingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
	body, etag, err := s.Store.Get(ctx, key)
	if s.armed && key == s.key && err == nil {
		s.armed = false
		if err := s.Store.Delete(ctx, key); err != nil {
			return nil, "", err
		}
	}
	return body, etag, err
}

func (s *vanishingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
	if key == s.key {
		*s.events = append(*s.events, "put")
	}
	return s.Store.Put(ctx, key, body, ifMatch)
}

// An invitation is never written without a membership read just before
// it. Here the invitation vanishes right after the pass reads it: that
// pass waits for the box's report and writes nothing, and the next one
// reads membership and only then invites again.
func TestAVanishedInvitationIsNeverWrittenWithoutARead(t *testing.T) {
	w := newEnrollWorld(t, "")
	w.pass(t)
	var events []string
	w.cloud.during = func() { events = append(events, "read") }
	key := "control/host-actor-invitations/" + newRef + ".json"
	vs := &vanishingStore{Store: w.e.Directory.Store, key: key, events: &events}
	w.e.Directory.Store = vs
	if got := w.pass(t); got != "invited" {
		t.Fatal(got)
	}
	vs.armed = true
	if got := w.pass(t); got != "waiting for its guard-ready report" {
		t.Fatalf("read, then vanished: %q", got)
	}
	if _, ok := w.invitation(t); ok {
		t.Fatal("the invitation was written again in the pass that read it")
	}
	if got := w.pass(t); got != "invited" {
		t.Fatalf("after it vanished: %q", got)
	}
	if got := w.pass(t); got != "waiting for its guard-ready report" {
		t.Fatalf("invited again: %q", got)
	}
	last := ""
	for _, ev := range events {
		if ev == "put" && last != "read" {
			t.Fatalf("an invitation Put with no membership read just before it: %v", events)
		}
		last = ev
	}
	if fmt.Sprint(events) != "[read put read put]" {
		t.Fatalf("events %v", events)
	}
}