package host

import (
	"context"
	"crypto/ed25519"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"syscall"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// failing names the checks that did not pass, with their reasons.
func failing(r *moves.TargetReport) map[string]string {
	out := map[string]string{}
	for _, id := range moves.AllChecks {
		if c := r.Checks[id]; !c.OK {
			out[id] = c.Reason
		}
	}
	return out
}

// The target's report flags each check on its own (§7.3): from a staged
// move.json, a serving-then-moving claim that has not touched uk, and no
// archive, T9 fails on touched and T5 on the archive while T1 passes; once
// the claim touches uk and the source stages an archive, both pass. Each
// injectable hook, failing, fails its own check alone.
func TestTargetReportPassFlagsEachCheck(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newPhase1Harness(t)
	h.start(t, "manual")
	body, _ := json.Marshal(moves.MoveJSON{MoveID: p1ID, Incarnation: h.inc, Host: p1Host, From: "ie", To: "uk"})
	h.f.store("data-uk", moves.MoveJSONKey(p1Host, p1ID), fakeMoveObject{lease.MirrorObject{ETag: "m", Size: int64(len(body))}, string(body)})
	tc := h.targetCfg()
	h.targetPass(t, tc)
	r := h.targetReport(t)
	if r == nil {
		t.Fatal("no target report")
	}
	f := failing(r)
	if !strings.Contains(f[moves.CheckClaim], "touched") || f[moves.CheckClaim] != t9Untouched("uk") {
		t.Fatalf("T9 = %q, want the touched sentence for uk (%v)", f[moves.CheckClaim], f)
	}
	if !strings.Contains(f[moves.CheckArchive], "no archive") {
		t.Fatalf("T5 = %q, want the no-archive sentence", f[moves.CheckArchive])
	}
	if _, bad := f[moves.CheckProtocol]; bad {
		t.Fatalf("T1 failed: %v", f)
	}
	if r.Box != "uk-box" || r.Region != "uk" || r.MoveID != p1ID {
		t.Fatalf("report identity %+v", r)
	}

	// Touch uk and uk-box, then let the source stage its archive.
	obs, _ := hostclaim.Read(ctx, h.s, p1Host)
	if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.Touch(p1ID, []string{"uk"}, []string{"uk-box"})); err != nil {
		t.Fatal(err)
	}
	h.pass(t)
	h.targetPass(t, tc)
	if f := failing(h.targetReport(t)); len(f) != 0 {
		t.Fatalf("after touch and archive: %v", f)
	}

	// Each hook alone.
	hookErr := errors.New("injected")
	for name, c := range map[string]struct {
		check string
		edit  func(*moveCfg)
		want  string
	}{
		"ComposeHeldEnv": {moves.CheckEnv, func(c *moveCfg) {
			c.ComposeHeldEnv = func(context.Context, release.Store, provision.Record, []byte) (map[string]string, map[string]bool, []string, provision.MoveInputs, error) {
				return nil, nil, nil, provision.MoveInputs{}, hookErr
			}
		}, t3NoCompose},
		"ComposeHeldEnv differs": {moves.CheckEnv, func(c *moveCfg) {
			inner := c.ComposeHeldEnv
			c.ComposeHeldEnv = func(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, map[string]bool, []string, provision.MoveInputs, error) {
				env, inst, sh, in, err := inner(ctx, s, rec, carried)
				env["APP_MODE"] = "staging"
				return env, inst, sh, in, err
			}
		}, t3Differs([]string{"APP_MODE (differs)"})},
		"ProbeStore": {moves.CheckStorage, func(c *moveCfg) {
			c.ProbeStore = func(context.Context, map[string]string, string) (string, error) { return "MEDIA", hookErr }
		}, t4Failed("MEDIA")},
		"DiskFree": {moves.CheckDisk, func(c *moveCfg) {
			c.DiskFree = func(string) (uint64, uint64, error) { return 0, 0, hookErr }
		}, t6Unmeasured},
		"ReleaseReady": {moves.CheckRelease, func(c *moveCfg) {
			c.ReleaseReady = func(context.Context, release.Store, provision.Record, string) moves.Check {
				return moves.Check{Reason: t7Waiting}
			}
		}, t7Waiting},
		"LaneProbe": {moves.CheckLane, func(c *moveCfg) {
			c.LaneProbe = func(context.Context, string, string, string) error { return hookErr }
		}, t10Failed},
	} {
		t.Run(name, func(t *testing.T) {
			hc := tc
			c.edit(&hc)
			h.targetPass(t, hc)
			f := failing(h.targetReport(t))
			// T4 probes the stores bound in T3's composition, so without one
			// it fails closed with T3's reason: the one dependency.
			if name == "ComposeHeldEnv" && f[moves.CheckStorage] == t3NoCompose {
				delete(f, moves.CheckStorage)
			}
			if len(f) != 1 || f[c.check] != c.want {
				t.Fatalf("%s failing: %v, want only %s = %q", name, f, c.check, c.want)
			}
		})
	}

	// T9 owns move.json's validation: one naming another move fails it
	// with the lane sentence.
	other, _ := json.Marshal(moves.MoveJSON{MoveID: p1ID2, Incarnation: h.inc, Host: p1Host, From: "ie", To: "uk"})
	h.f.store("data-uk", moves.MoveJSONKey(p1Host, p1ID), fakeMoveObject{lease.MirrorObject{ETag: "o", Size: int64(len(other))}, string(other)})
	h.targetPass(t, tc)
	if f := failing(h.targetReport(t)); f[moves.CheckClaim] != t9OtherMove {
		t.Fatalf("T9 with another move's move.json: %v", f)
	}
	wrongTo, _ := json.Marshal(moves.MoveJSON{MoveID: p1ID, Incarnation: h.inc, Host: p1Host, From: "ie", To: "se"})
	h.f.store("data-uk", moves.MoveJSONKey(p1Host, p1ID), fakeMoveObject{lease.MirrorObject{ETag: "w", Size: int64(len(wrongTo))}, string(wrongTo)})
	h.targetPass(t, tc)
	if f := failing(h.targetReport(t)); f[moves.CheckClaim] != t9OtherMove {
		t.Fatalf("T9 with a move.json for another region: %v", f)
	}
	h.f.store("data-uk", moves.MoveJSONKey(p1Host, p1ID), fakeMoveObject{lease.MirrorObject{ETag: "m", Size: int64(len(body))}, string(body)})

	// T5 checks the archive against its manifest: a manifest whose entry
	// digest was changed under the same version is refused.
	sr := h.sourceReport(t)
	mkey := sr.Archive.ManifestKey
	var man map[string]any
	if err := json.Unmarshal([]byte(h.f.buckets["data-uk"][mkey].body), &man); err != nil {
		t.Fatal(err)
	}
	entries := man["entries"].([]any)
	entries[0].(map[string]any)["sha256"] = "sha256:" + strings.Repeat("0", 64)
	tampered, _ := json.Marshal(man)
	good := h.f.buckets["data-uk"][mkey]
	h.f.store("data-uk", mkey, fakeMoveObject{good.MirrorObject, string(tampered)})
	// A version is immutable in S3, so T5 keeps its verdict per version;
	// the fake rewrote one in place, so the target starts with no verdicts.
	h.worker2.archives = nil
	h.targetPass(t, tc)
	if f := failing(h.targetReport(t)); f[moves.CheckArchive] != t5Refused {
		t.Fatalf("T5 with a tampered manifest: %v", f)
	}
	h.f.store("data-uk", mkey, good)
	h.worker2.archives = nil

	// Only the named target writes a report.
	h3 := newPhase1Harness(t)
	h3.start(t, "manual")
	notTarget := h3.targetCfg()
	notTarget.Box = "uk-box-2"
	h3.targetPass(t, notTarget)
	if r := h3.targetReport(t); r != nil {
		t.Fatalf("a box that is not the move's target wrote a report: %+v", r)
	}
}

// T3 composes the flat held row's environment on a scratch directory
// (ruling F19): the rendered bundle and the carried .env, through
// LaunchEnvironment. A carried variable overriding a rendered one is
// reported as shadowed (round-6 finding 7), with no instance overlay.
func TestComposeHeldEnvReportsShadowedNames(t *testing.T) {
	ctx := context.Background()
	s := release.NewDirStore(t.TempDir())
	privB64, _, err := appconfig.GenerateKey()
	if err != nil {
		t.Fatal(err)
	}
	priv, _ := appconfig.ParsePrivateKey(privB64, "test")
	body, _ := json.Marshal(appconfig.Manifest{Schema: 1, App: "app", Vars: map[string]appconfig.Var{
		"APP_MODE": {Kind: "env", Value: "rendered"}, "ONLY_RENDERED": {Kind: "env", Value: "1"}}})
	etag, err := s.Put(ctx, appconfig.Key("bab", "app"), body, "")
	if err != nil {
		t.Fatal(err)
	}
	rec := provision.Record{Host: p1Host, Account: "bab", App: "app"}
	env, _, shadowed, in, err := composeHeldEnv(priv, "uk", true)(ctx, s, rec, []byte("APP_MODE=carried\nCARRIED=1\n"))
	if err != nil {
		t.Fatal(err)
	}
	if strings.Join(shadowed, ",") != "APP_MODE" {
		t.Fatalf("shadowed = %v", shadowed)
	}
	if env["APP_MODE"] != "carried" || env["ONLY_RENDERED"] != "1" || env["CARRIED"] != "1" || env["CARLOS_REGION"] != "uk" ||
		env["CARLOS_PURGE"] != "header" || env["CARLOS_CONTROL_SOCKET"] != provision.DefaultSocket(p1Host)+".control" {
		t.Fatalf("composed env lacks an expected variable (names: %v)", envNames(env))
	}
	if in.BundleETags["default"] != etag || in.OverlayETag != "" {
		t.Fatalf("inputs %+v, want the bundle etag %q", in, etag)
	}
	// No cache on the live box: no CARLOS_PURGE.
	env, _, _, _, err = composeHeldEnv(priv, "uk", false)(ctx, s, rec, nil)
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := env["CARLOS_PURGE"]; ok {
		t.Fatal("CARLOS_PURGE composed for a box with no cache")
	}

	// Through the report: the shadowed name is a note, never a failure.
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.pass(t)
	tc := h.targetCfg()
	inner := tc.ComposeHeldEnv
	tc.ComposeHeldEnv = func(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, map[string]bool, []string, provision.MoveInputs, error) {
		env, inst, _, in, err := inner(ctx, s, rec, carried)
		return env, inst, []string{"APP_MODE"}, in, err
	}
	h.targetPass(t, tc)
	r := h.targetReport(t)
	if !r.Checks[moves.CheckEnv].OK || len(r.Notes) != 1 || r.Notes[0] != t3Shadowed("APP_MODE") {
		t.Fatalf("T3 %+v notes %v", r.Checks[moves.CheckEnv], r.Notes)
	}
}

func envNames(env map[string]string) []string {
	var out []string
	for k := range env {
		out = append(out, k)
	}
	return out
}

// Round-6 finding 3: on a hardened box the agent cannot write the version
// store. From a MoveStart'ed record with no target registry row, T7 leaves
// a held-stage request naming the source report's Release and touches the
// stage-requested marker; the root pass (ServeHeldStageRequests under
// heldStageAuthority) materializes it into optDir; the next pass finds
// LiveVersion equal and T7 passes. A request naming another version, or an
// op the claim does not name, is refused and live/<host> is untouched.
func TestPreparingRelaysHeldStageOnAHardenedBox(t *testing.T) {
	ctx := context.Background()
	h := newPhase1Harness(t)
	h.start(t, "manual")
	shipPinnedV7(t, h.s)
	h.pass(t) // the source report names Release v7

	reg := testRegistry(t)
	targetData, targetOpt := t.TempDir(), t.TempDir()
	prev := heldMaterialize
	t.Cleanup(func() { heldMaterialize = prev })
	heldMaterialize = func(context.Context, release.Store, ed25519.PublicKey, *release.DigestRatchet, string, registry.Route, string, *release.ProofGate, func(func() error) error) (*release.Manifest, error) {
		return nil, fmt.Errorf("%w: %w", release.ErrStageBlocked, &os.PathError{Op: "mkdir", Path: "/opt/carlos/versions", Err: syscall.EROFS})
	}
	tc := h.targetCfg()
	tc.DataDir = targetData
	tc.ReleaseReady = heldReleaseReady(reg, targetOpt, targetData, "uk-box", hsPub, hsRatchet(t), offProofGate(t), stageRequester(targetData))
	h.targetPass(t, tc)
	if f := failing(h.targetReport(t)); f[moves.CheckRelease] != t7Waiting {
		t.Fatalf("T7 = %q, want %q", f[moves.CheckRelease], t7Waiting)
	}
	raw, err := os.ReadFile(filepath.Join(release.HeldStageDir(targetData), p1Host+".json"))
	if err != nil {
		t.Fatal(err)
	}
	var req release.HeldStageRequest
	if err := json.Unmarshal(raw, &req); err != nil || req.Version != "v7" || req.Op != p1ID || req.Box != "uk-box" || req.Account != "bab" || req.App != "app" {
		t.Fatalf("request %s (%v)", raw, err)
	}
	if !fileExistsForTest(filepath.Join(targetData, stageRequestMarker)) {
		t.Fatal("the stage-requested marker was not touched")
	}

	// The root pass, with the relay's own authority, materializes it.
	heldMaterialize = prev
	if err := release.ServeHeldStageRequestsWithPins(ctx, h.s, hsPub, hsRatchet(t), targetOpt, targetData, offProofGate(t), heldStageAuthority(reg, h.s, "uk-box", targetData)); err != nil {
		t.Fatal(err)
	}
	if v, err := release.LiveVersion(targetOpt, p1Host); err != nil || v != "v7" {
		t.Fatalf("LiveVersion = %q, %v", v, err)
	}
	h.targetPass(t, tc)
	if f := failing(h.targetReport(t)); f[moves.CheckRelease] != "" {
		t.Fatalf("T7 after the root stage: %q", f[moves.CheckRelease])
	}

	// Requests the authority refuses: another version, another op.
	for _, bad := range []release.HeldStageRequest{
		{Account: "bab", App: "app", Host: p1Host, Version: "v8", Op: p1ID, Box: "uk-box"},
		{Account: "bab", App: "app", Host: p1Host, Version: "v7", Op: p1ID2, Box: "uk-box"},
	} {
		opt := t.TempDir()
		if err := writeHeldStageRequest(targetData, bad); err != nil {
			t.Fatal(err)
		}
		if err := release.ServeHeldStageRequestsWithPins(ctx, h.s, hsPub, hsRatchet(t), opt, targetData, offProofGate(t), heldStageAuthority(reg, h.s, "uk-box", targetData)); err != nil {
			t.Fatal(err)
		}
		if v, _ := release.LiveVersion(opt, p1Host); v != "" {
			t.Fatalf("a refused request (%s, %s) repointed live to %q", bad.Version, bad.Op, v)
		}
	}
}

// T4 finds the store bindings in the composed environment by objstore's
// variable names; a prefix without its credentials is not a binding.
func TestStorePrefixesFollowTheDeliveredNames(t *testing.T) {
	env := map[string]string{
		"MEDIA_BUCKET": "b", "MEDIA_ACCESS_KEY": "k", "MEDIA_SECRET_KEY": "s",
		"STORE_BUCKET": "b2", "STORE_ACCESS_KEY": "k2", "STORE_SECRET_KEY": "s2", "STORE_REGION": "eu-west-1",
		"HALF_BUCKET": "b3", "HALF_ACCESS_KEY": "k3",
		"_BUCKET": "x",
	}
	if got := strings.Join(storePrefixes(env), ","); got != "MEDIA,STORE" {
		t.Fatalf("prefixes %q", got)
	}
}

// T10 in production: put-if-absent then delete by version; a probe object
// left behind is removed and the check fails for that pass only.
func TestLaneProbePutsAndDeletesByVersion(t *testing.T) {
	ctx := context.Background()
	f := newFakeMoveMirror()
	vs := &fakeVersioned{f: f}
	probe := laneProbe(func(context.Context, string, string) (lease.VersionedStore, error) { return vs, nil })
	if err := probe(ctx, "data-ie", "eu-west-1", p1ID); err != nil {
		t.Fatal(err)
	}
	if _, ok := f.buckets["data-ie"][moves.LaneKey(p1ID)]; ok {
		t.Fatal("the probe object was left behind")
	}
	f.put("data-ie", moves.LaneKey(p1ID), "left")
	f.setVersion("data-ie", moves.LaneKey(p1ID), "v-left")
	if err := probe(ctx, "data-ie", "eu-west-1", p1ID); err == nil {
		t.Fatal("a left-behind probe object passed")
	}
	if err := probe(ctx, "data-ie", "eu-west-1", p1ID); err != nil {
		t.Fatalf("the next pass did not start clean: %v", err)
	}
	vs.deny = func(string, string, string) error { return errors.New("AccessDenied") }
	if err := probe(ctx, "data-ie", "eu-west-1", p1ID); err == nil {
		t.Fatal("a denied lane passed")
	}
}

// The disk measurement (O_DIRECT on linux, synced elsewhere) and the
// per-file sync rate both produce a usable, dated measurement.
func TestMeasureDiskIn(t *testing.T) {
	t.Parallel()
	if testing.Short() {
		t.Skip("writes and reads 64 MiB")
	}
	dir := t.TempDir()
	disk, files, err := measureDiskIn(dir, time.Now)(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if disk.MiBPerS <= 0 || disk.At.IsZero() || disk.Bytes != 2*measureDiskSize {
		t.Fatalf("disk %+v", disk)
	}
	if files.PerS <= 0 || files.At.IsZero() || files.Files != measureFsyncFiles {
		t.Fatalf("fsync %+v", files)
	}
	if ents, _ := os.ReadDir(filepath.Join(dir, ".moves")); len(ents) != 0 {
		t.Fatalf("the measurement left %d entries behind", len(ents))
	}
}

// The regional S3 measurement writes the probe key outside moves/ with
// this box's own client and deletes it (F11).
func TestMeasureS3UsesTheProbeKeyAndCleansUp(t *testing.T) {
	t.Parallel()
	f := newFakeMoveMirror()
	m := measureS3(func(context.Context) (lease.MirrorStore, error) { return f, nil }, "data-uk", (&stepClock{now: time.Now()}).Now)
	r, err := m(context.Background(), moves.ProbeKey(p1Host, p1ID))
	if err != nil {
		t.Fatal(err)
	}
	if r.MiBPerS <= 0 || r.At.IsZero() {
		t.Fatalf("%+v", r)
	}
	if len(f.buckets["data-uk"]) != 0 {
		t.Fatalf("the probe object was left behind: %v", f.buckets["data-uk"])
	}
	if strings.Join(f.ops, ",") != "put data-uk/"+moves.ProbeKey(p1Host, p1ID)+",delete data-uk/"+moves.ProbeKey(p1Host, p1ID) {
		t.Fatalf("ops %v", f.ops)
	}
}

// T7 on a box that can write its version store: the release is staged and
// published directly, under the same authority the root relay uses, so a
// claim that names another op publishes nothing.
func TestHeldReleaseReadyOnAWritableBox(t *testing.T) {
	ctx := context.Background()
	h := newPhase1Harness(t)
	rec := h.start(t, "manual")
	shipPinnedV7(t, h.s)
	h.pass(t)
	reg := testRegistry(t)
	data, opt := t.TempDir(), t.TempDir()
	ready := heldReleaseReady(reg, opt, data, "uk-box", hsPub, hsRatchet(t), offProofGate(t), nil)
	if c := ready(ctx, h.s, rec, "v7"); !c.OK {
		t.Fatalf("T7 = %+v", c)
	}
	if v, _ := release.LiveVersion(opt, p1Host); v != "v7" {
		t.Fatalf("live = %q", v)
	}
	// The claim moves to another op while the release stages: the
	// publication's authority refuses, nothing is published.
	prev := heldMaterialize
	t.Cleanup(func() { heldMaterialize = prev })
	heldMaterialize = func(ctx context.Context, s release.Store, pub ed25519.PublicKey, r *release.DigestRatchet, optDir string, rt registry.Route, v string, proof *release.ProofGate, guard func(func() error) error) (*release.Manifest, error) {
		rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.ID = p1ID2 })
		return prev(ctx, s, pub, r, optDir, rt, v, proof, guard)
	}
	opt2 := t.TempDir()
	if c := heldReleaseReady(reg, opt2, data, "uk-box", hsPub, hsRatchet(t), offProofGate(t), nil)(ctx, h.s, rec, "v7"); c.OK || c.Reason != t7Failed {
		t.Fatalf("T7 when the op changed during staging = %+v", c)
	}
	if v, _ := release.LiveVersion(opt2, p1Host); v != "" {
		t.Fatalf("published after the op changed: %q", v)
	}
	// Under another op from the start: refused before anything is staged.
	heldMaterialize = prev
	opt3 := t.TempDir()
	if c := heldReleaseReady(reg, opt3, data, "uk-box", hsPub, hsRatchet(t), offProofGate(t), nil)(ctx, h.s, rec, "v7"); c.OK || c.Reason != t7Failed {
		t.Fatalf("T7 under another op = %+v", c)
	}
	if _, err := os.Stat(filepath.Join(opt3, "versions")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a release was staged for a stale op: %v", err)
	}
}

// probeStore's SDK path, locally: against a minimal S3 endpoint it lists one
// key, then puts, gets and deletes .carlos-move-probe/<id>, in that order,
// path-style at the delivered endpoint, and a refusal fails it. (The real
// server is TestT4StorageProbeMinio's, in CI.)
func TestProbeStoreSpeaksS3(t *testing.T) {
	var seen []string
	objects := map[string][]byte{}
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if !strings.Contains(r.Header.Get("Authorization"), "Credential=good/") {
			w.WriteHeader(http.StatusForbidden)
			io.WriteString(w, `<?xml version="1.0"?><Error><Code>AccessDenied</Code><Message>no</Message></Error>`)
			return
		}
		key := strings.TrimPrefix(r.URL.Path, "/bkt")
		switch {
		case r.Method == http.MethodGet && r.URL.Query().Get("list-type") == "2":
			seen = append(seen, "list")
			io.WriteString(w, `<?xml version="1.0"?><ListBucketResult><Name>bkt</Name><KeyCount>0</KeyCount><IsTruncated>false</IsTruncated></ListBucketResult>`)
		case r.Method == http.MethodPut:
			seen = append(seen, "put "+key)
			b, _ := io.ReadAll(r.Body)
			objects[key] = b
		case r.Method == http.MethodGet:
			seen = append(seen, "get "+key)
			w.Write(objects[key])
		case r.Method == http.MethodDelete:
			seen = append(seen, "delete "+key)
			delete(objects, key)
			w.WriteHeader(http.StatusNoContent)
		default:
			w.WriteHeader(http.StatusMethodNotAllowed)
		}
	}))
	t.Cleanup(srv.Close)
	env := map[string]string{"STORE_BUCKET": "bkt", "STORE_ENDPOINT": srv.URL, "STORE_ACCESS_KEY": "good", "STORE_SECRET_KEY": "s"}
	if prefix, err := probeStores(context.Background(), env, p1ID); err != nil {
		t.Fatalf("%s: %v", prefix, err)
	}
	k := "/.carlos-move-probe/" + p1ID
	if got, want := strings.Join(seen, ","), "list,put "+k+",get "+k+",delete "+k; got != want {
		t.Fatalf("requests %q, want %q", got, want)
	}
	if len(objects) != 0 {
		t.Fatal("the probe object was left behind")
	}
	env["STORE_ACCESS_KEY"] = "bad"
	if prefix, err := probeStores(context.Background(), env, p1ID); err == nil || prefix != "STORE" {
		t.Fatalf("wrong credentials: %q %v", prefix, err)
	}
}

// T11 prices the database's logical size, read from the newest staged LTX
// header (Commit pages of PageSize), never the compressed bytes staged
// (Task 14 review finding 1). The source's local database file counts when
// it is larger; a newest file that is not LTX fails T6 and T11 closed.
func TestT11PricesTheLogicalDatabaseSize(t *testing.T) {
	t.Parallel()
	h := newPhase1Harness(t)
	h.start(t, "manual")
	newest := ltxForTest(t, 4096, 1<<18, 3, 3) // a 1 GiB database, one page in this file
	if len(newest) > 64<<10 {
		t.Fatalf("the fixture is %d bytes: not compressed enough to prove anything", len(newest))
	}
	key := "instances/" + p1Host + "/" + h.inc + "/g-cur/0000/0000000000000003-0000000000000003.ltx"
	h.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "newest", Size: int64(len(newest))}, string(newest)})
	h.pass(t)
	h.targetPass(t, h.targetCfg())
	r := h.targetReport(t)
	if r.DBMiB != 1024 {
		t.Fatalf("T11 priced %v MiB, want the 1024 MiB the newest header names (staged bytes are about %d KiB)", r.DBMiB, len(newest)>>10)
	}

	// A larger local database file on the source wins.
	db, err := os.Create(provision.DefaultDBPath(h.cfg.DataDir, p1Host))
	if err != nil {
		t.Fatal(err)
	}
	if err := db.Truncate(2 << 30); err != nil { // sparse
		t.Fatal(err)
	}
	db.Close()
	h.pass(t)
	if sr := h.sourceReport(t); sr.DBBytes != 2<<30 {
		t.Fatalf("source DBBytes %d", sr.DBBytes)
	}
	h.targetPass(t, h.targetCfg())
	if r := h.targetReport(t); r.DBMiB != 2048 {
		t.Fatalf("T11 priced %v MiB with a 2 GiB source file", r.DBMiB)
	}

	// The newest staged file is not LTX: the size is unknown, so T6 and
	// T11 fail closed.
	h2 := newPhase1Harness(t)
	h2.start(t, "manual")
	h2.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "junk", Size: 4}, "junk"})
	h2.pass(t)
	h2.targetPass(t, h2.targetCfg())
	f := failing(h2.targetReport(t))
	if f[moves.CheckBudget] != t11Unmeasured || f[moves.CheckDisk] != t11Unmeasured {
		t.Fatalf("an unreadable newest header: %v", f)
	}

	// A header that parses but names no pages is no measurement either.
	zero := append([]byte(nil), newest...)
	binary.BigEndian.PutUint32(zero[12:], 0) // Commit
	h3 := newPhase1Harness(t)
	h3.start(t, "manual")
	h3.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "zero", Size: int64(len(zero))}, string(zero)})
	h3.pass(t)
	h3.targetPass(t, h3.targetCfg())
	if f := failing(h3.targetReport(t)); f[moves.CheckBudget] != t11Unmeasured {
		t.Fatalf("a header naming zero pages: %v", f)
	}
}

// T5's verdicts (with the carried .env) are kept only while their move is
// in flight into this box.
func TestArchiveVerdictsAreEvictedWhenTheMoveEnds(t *testing.T) {
	ctx := context.Background()
	h := newPhase1Harness(t)
	h.start(t, "manual")
	h.pass(t)
	h.targetPass(t, h.targetCfg())
	if len(h.worker2.archives) != 1 {
		t.Fatalf("%d verdicts cached", len(h.worker2.archives))
	}
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil {
		t.Fatal(err)
	}
	// The source's pass begins the abort; the record stays aborting while
	// the target region parks the lane (Plan 5 Task 1). The move is over for
	// the target's caches already.
	h.pass(t)
	if m := h.read(t, p1Host).Move; m == nil || m.State != provision.MoveAborting {
		t.Fatalf("the record during the park wait: %+v", m)
	}
	h.targetPass(t, h.targetCfg())
	if len(h.worker2.archives) != 0 {
		t.Fatalf("a verdict outlived its move: %v", h.worker2.archives)
	}
}