amadan

carlos / platform Public

host: tests for the participants' reports, the stage book, the sealer's lift and the status block

TestLaneReportUsesTheExistingProbe, TestTargetReadyReportsEveryFactWithKnown,
TestSourcePrecheckFacts, TestStrayFileRefusesBeforeStop,
TestEmptyStrayDirsRemovedAtStartOnly, TestQueuedRoundRunsOnlyLanePrecheckAndReady,
TestParticipantsReReportOnANewPrecheckStamp, TestStageBookAcrossRounds,
TestObservedMirrorEmitsEveryDecision (the box half, through NewS3Mirror and
stageDeleteGate), TestSealerPublishesTombstoneLiftedBeforeCleanup,
TestStatusHostsBlockThroughTheRealPath, TestHostFactsCarryTheLifetimeAndAges
and TestVersionedArmedForOffline, over the seal tests' S3 world.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 77da46fa5240614fdf2f28b946863411d5a738ed parent 74ad1e0
1 file changed, +1127 −0
  • internal/host/movesoffline_reports_test.go +1127 −0
diff --git a/internal/host/movesoffline_reports_test.go b/internal/host/movesoffline_reports_test.go
new file mode 100644
index 000000000..ebd8cdd60
--- /dev/null
+++ b/internal/host/movesoffline_reports_test.go
@@ -0,0 +1,1127 @@
+//go:build linux
+
+package host
+
+import (
+ "context"
+ "crypto/md5"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "reflect"
+ "sort"
+ "strings"
+ "sync"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/layout"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+// The boxes' side of a self-serve move (self-serve moves plan 2 Task 6),
+// tested over the seal tests' world: the S3 wire protocol behind
+// lease.NewS3Mirror and lease.NewVersionedS3 (leasetest.NewS3Server), real
+// litestream replicas, real registries and activators, the production
+// edges-file writer, and offlineRound, the moves pass's dispatch. ie-a (the
+// home) is the source participant on instance i-ie, uk-box the target on
+// i-uk.
+
+const (
+ ssSourceInstance = "i-ie"
+ ssTargetInstance = "i-uk"
+)
+
+// ssParticipants are the move's pinned boxes.
+func ssParticipants() provision.Participants {
+ return provision.Participants{
+ Source: []provision.Participant{{BoxID: "ie-a", Label: "ie-a", InstanceID: ssSourceInstance, Verified: true}},
+ Target: provision.Participant{BoxID: "uk-box", Label: "uk-box", InstanceID: ssTargetInstance, Verified: true},
+ }
+}
+
+// ssIdentity gives b its instance id, as armMovesPass sets it before the
+// worker and its keep view are built, and answers the stage path probe.
+func ssIdentity(b *offBox, instance string) {
+ b.w.cfg.InstanceID = instance
+ b.w.cfg.Offline = newOfflineView(b.w.cfg, nil)
+ armed := true
+ b.w.cfg.StagePathArmed = func() *bool { return &armed }
+}
+
+// newSSWorld is a seal world whose two boxes carry their pinned identity.
+func newSSWorld(t *testing.T) *sealWorld {
+ sw := newSealWorld(t)
+ ssIdentity(sw.ie, ssSourceInstance)
+ ssIdentity(sw.uk, ssTargetInstance)
+ sw.uk.w.cfg.ChannelVersion = func(context.Context, release.Store, registry.Route) (string, error) { return sealVersion, nil }
+ sw.uk.w.cfg.OwnPrefix = sealPrefix
+ return sw
+}
+
+// queue is the self-serve queue's record write: offline-queued with the
+// participants pinned, staging into data-uk.
+func (sw *sealWorld) queue() {
+ sw.t.Helper()
+ inc := sw.record().Incarnation
+ sw.move(sw.proposed(func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineQueue(offID, "", "uk", "owner@example.com", provision.StageTarget{Bucket: sealUK, Region: sealUKRegion}, ssParticipants(),
+ provision.SelfServeMove{ID: "ss_1", Incarnation: inc}, at, provision.Receipt{})
+ }))
+}
+
+// toStaging is the job's queued-to-staging write.
+func (sw *sealWorld) toStaging() {
+ sw.t.Helper()
+ sw.advance(time.Second)
+ sw.move(sw.proposed(func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineQueuedToStaging(offID, at, provision.Receipt{})
+ }))
+}
+
+// precheckAgain is the job's precheck request: a newer precheck_at.
+func (sw *sealWorld) precheckAgain() {
+ sw.t.Helper()
+ sw.advance(time.Second)
+ sw.move(sw.proposed(func(at time.Time) func(*provision.Record) error {
+ return provision.OfflinePrecheck(offID, at, provision.Receipt{})
+ }))
+}
+
+// targetRound is the target box's tick without rewriting its published
+// edges file (newSealWorld's carries the data bucket the move stages into).
+func (sw *sealWorld) targetRound(b *offBox) error {
+ sw.t.Helper()
+ b.reconcile()
+ ctx := context.Background()
+ refreshOfflineView(ctx, sw.s, b.w.cfg.Offline)
+ return b.w.offlineRound(ctx, sw.s, sw.record())
+}
+
+func (sw *sealWorld) mustTargetRound(b *offBox) {
+ sw.t.Helper()
+ if err := sw.targetRound(b); err != nil {
+ sw.t.Fatal(err)
+ }
+}
+
+// validFor fails unless r validates for the record's move now.
+func (sw *sealWorld) validFor(r moves.OfflineReport) {
+ sw.t.Helper()
+ if err := moves.ValidateOfflineReportFor(r, sw.record().Move, sw.clock()); err != nil {
+ sw.t.Fatalf("%s/%s does not validate for the move: %v", r.Role, r.Phase, err)
+ }
+}
+
+// --- the lane, the precheck and the ready report --------------------------
+
+func TestLaneReportUsesTheExistingProbe(t *testing.T) {
+ sw := newSSWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ sw.queue()
+ // The probe armOffline armed is laneProbe over NewVersionedS3; count its
+ // calls around it.
+ calls := 0
+ probe := ie.w.cfg.LaneProbe
+ if probe == nil || ie.w.cfg.Versioned == nil {
+ t.Fatal("armOffline armed no lane probe")
+ }
+ ie.w.cfg.LaneProbe = func(ctx context.Context, bucket, region, id string) error {
+ calls++
+ return probe(ctx, bucket, region, id)
+ }
+ sw.mustStagePass(ie)
+ r := sw.report(moves.OfflineRoleLane, "ie-a")
+ sw.validFor(r)
+ want := moves.LaneReport{Bucket: sealUK, Region: sealUKRegion, Key: moves.LaneKey(offID), OK: true}
+ if r.Phase != moves.OfflinePhaseLane || r.Lane == nil || *r.Lane != want || r.BoxID != "ie-a" || r.InstanceID != ssSourceInstance {
+ t.Fatalf("lane report %+v lane %+v", r, r.Lane)
+ }
+ if calls != 1 {
+ t.Fatalf("%d probes", calls)
+ }
+ if _, ok := sw.fake.Current(sealUK, moves.LaneKey(offID)); ok {
+ t.Fatal("the probe object is left")
+ }
+ vers, err := sw.fake.ListVersions(context.Background(), sealUK, moves.LaneKey(offID))
+ if err != nil || len(vers) != 0 {
+ t.Fatalf("the probe left versions %+v (%v): the existing probe deletes its version", vers, err)
+ }
+ // Once per move: another pass probes nothing.
+ sw.advance(time.Minute)
+ sw.mustStagePass(ie)
+ if calls != 1 || !sw.report(moves.OfflineRoleLane, "ie-a").At.Equal(r.At) {
+ t.Fatalf("the lane was probed again (%d)", calls)
+ }
+ // A probe that fails is reported, with its error.
+ sw2 := newSSWorld(t)
+ sw2.instance(sw2.ie)
+ sw2.queue()
+ sw2.ie.w.cfg.LaneProbe = func(context.Context, string, string, string) error { return errors.New("put: AccessDenied") }
+ sw2.mustStagePass(sw2.ie)
+ if r := sw2.report(moves.OfflineRoleLane, "ie-a"); r.Lane == nil || r.Lane.OK || r.Lane.Error != "put: AccessDenied" {
+ t.Fatalf("failed lane %+v", r.Lane)
+ }
+}
+
+func TestTargetReadyReportsEveryFactWithKnown(t *testing.T) {
+ sw := newSSWorld(t)
+ uk := sw.uk
+ sw.queue()
+ sw.mustTargetRound(uk)
+ r := sw.report(moves.OfflineRoleTarget, "uk-box")
+ sw.validFor(r)
+ if r.Phase != moves.OfflinePhaseReady || r.Ready == nil || !r.Request.Equal(*sw.record().Move.PrecheckAt) {
+ t.Fatalf("ready report %+v", r)
+ }
+ rd := *r.Ready
+ clean := map[string]moves.Fact{
+ "route_absent": rd.RouteAbsent, "timers_absent": rd.TimersAbsent, "live_link": rd.LiveLink, "programs_file": rd.ProgramsFile,
+ "park_dir": rd.ParkDir, "disk_free": rd.DiskFree, "lock": rd.Lock, "stage_path_armed": rd.StagePathArmed,
+ }
+ for name, f := range clean {
+ if !f.Known || !f.OK {
+ t.Errorf("%s on a clean target: %+v", name, f)
+ }
+ }
+ wantPaths := []string{provision.InstanceDir(uk.dd, offHost), layout.RawInstanceDir(uk.dd, offHost), filepath.Join(uk.dd, offHost),
+ filepath.Join(uk.dd, offHost+".env"), filepath.Join(uk.dd, offHost+".config.env"), filepath.Join(uk.dd, offHost+".db")}
+ if len(rd.DataPaths) != len(wantPaths) {
+ t.Fatalf("data paths %v", rd.DataPaths)
+ }
+ for _, p := range wantPaths {
+ if f := rd.DataPaths[p]; !f.Known || !f.OK {
+ t.Errorf("%s on a clean target: %+v", p, f)
+ }
+ }
+ if !rd.MovesPrefix.Known || len(rd.MovesPrefix.Objects) != 0 || !rd.InstancesPrefix.Known || len(rd.InstancesPrefix.Objects) != 0 {
+ t.Fatalf("prefixes %+v %+v", rd.MovesPrefix, rd.InstancesPrefix)
+ }
+ if rd.ChannelBuild != sealVersion || rd.DiskFreeBytes <= 0 || rd.Capabilities == nil {
+ t.Fatalf("channel %q disk %d capabilities %v", rd.ChannelBuild, rd.DiskFreeBytes, rd.Capabilities)
+ }
+
+ // Each fact turns, and says so, on a target that is not clean. A new
+ // precheck stamp asks for the report again.
+ ctx := context.Background()
+ if err := uk.reg.UpsertTimer(registry.Timer{Host: offHost, Name: "n", Kind: "cron", Spec: "0 3 * * *", Path: "/n", NextAt: 1}); err != nil {
+ t.Fatal(err)
+ }
+ for _, p := range wantPaths {
+ writeFile(t, p, "left behind")
+ }
+ if err := os.MkdirAll(filepath.Join(uk.dd, ".offline-moves", offHost, offID, "target-before"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ writeFile(t, filepath.Join(uk.dd, "programs", offHost+".json"), `{"schema":1,"programs":["example.com/shop"]}`)
+ if err := os.MkdirAll(filepath.Dir(release.LiveLink(uk.opt, offHost)), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(filepath.Join(uk.opt, "versions", "v41", sw.acct, "app"), release.LiveLink(uk.opt, offHost)); err != nil {
+ t.Fatal(err)
+ }
+ sw.fake.Put(sealUK, activator.MovesPrefix(offHost)+"0000/foreign.ltx", []byte("foreign"))
+ sw.fake.Put(sealUK, sealPrefix+"/"+offHost+"/0000/old.ltx", []byte("old"))
+ sw.fake.Put(sealUK, sealPrefix+"/"+offHost+"/lock.json", []byte(`{"owner":"moved:`+offID2+`"}`))
+ // A delete marker is not a current object.
+ sw.fake.Put(sealUK, activator.MovesPrefix(offHost)+"0000/gone.ltx", []byte("gone"))
+ sw.fake.BareDelete(sealUK, activator.MovesPrefix(offHost)+"0000/gone.ltx")
+ uk.w.cfg.DiskFree = func(string) (uint64, uint64, error) { return 100 << 30, 1 << 30, nil }
+ off := false
+ uk.w.cfg.StagePathArmed = func() *bool { return &off }
+ sw.precheckAgain()
+ sw.mustTargetRound(uk)
+ r = sw.report(moves.OfflineRoleTarget, "uk-box")
+ sw.validFor(r)
+ rd = *r.Ready
+ turned := map[string]moves.Fact{
+ "timers_absent": rd.TimersAbsent, "live_link": rd.LiveLink, "programs_file": rd.ProgramsFile,
+ "park_dir": rd.ParkDir, "disk_free": rd.DiskFree, "lock": rd.Lock, "stage_path_armed": rd.StagePathArmed,
+ }
+ for name, f := range turned {
+ if !f.Known || f.OK {
+ t.Errorf("%s on a dirty target: %+v", name, f)
+ }
+ }
+ for _, p := range wantPaths {
+ if f := rd.DataPaths[p]; !f.Known || f.OK {
+ t.Errorf("%s present: %+v", p, f)
+ }
+ }
+ if rd.DiskFreeBytes != 1<<30 || !strings.Contains(rd.LiveLink.Detail, "v41") || !strings.Contains(rd.ProgramsFile.Detail, "example.com/shop") ||
+ !strings.Contains(rd.Lock.Detail, sealPrefix+"/"+offHost+"/lock.json") {
+ t.Fatalf("details: disk %d live %q programs %q lock %q", rd.DiskFreeBytes, rd.LiveLink.Detail, rd.ProgramsFile.Detail, rd.Lock.Detail)
+ }
+ mp, ip := rd.MovesPrefix.Objects, rd.InstancesPrefix.Objects
+ if len(mp) != 1 || mp["0000/foreign.ltx"].ETag == "" || mp["0000/foreign.ltx"].VersionID == "" || mp["0000/foreign.ltx"].LastModified.IsZero() {
+ t.Fatalf("moves prefix %+v", mp)
+ }
+ if len(ip) != 1 || ip["0000/old.ltx"].Size != 3 {
+ t.Fatalf("instances prefix %+v (lock.json is the lock fact, not an object)", ip)
+ }
+ // A route for the host on the target, and a live link naming the
+ // channel's own build, which is clean.
+ if err := os.Remove(release.LiveLink(uk.opt, offHost)); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(release.VersionPath(uk.opt, sw.acct, sealVersion, "app"), release.LiveLink(uk.opt, offHost)); err != nil {
+ t.Fatal(err)
+ }
+ if err := uk.reg.Upsert(registry.Route{Host: offHost, Kind: "instance", Socket: "/run/x.sock"}); err != nil {
+ t.Fatal(err)
+ }
+ sw.precheckAgain()
+ sw.mustTargetRound(uk)
+ rd = *sw.report(moves.OfflineRoleTarget, "uk-box").Ready
+ if !rd.LiveLink.Known || !rd.LiveLink.OK || !rd.RouteAbsent.Known || rd.RouteAbsent.OK {
+ t.Fatalf("live link %+v route %+v", rd.LiveLink, rd.RouteAbsent)
+ }
+ // A box that cannot read a fact says it does not know.
+ uk.w.cfg.DiskFree = func(string) (uint64, uint64, error) { return 0, 0, errors.New("statfs: EIO") }
+ uk.w.cfg.StagePathArmed = func() *bool { return nil }
+ uk.w.cfg.ChannelVersion = func(context.Context, release.Store, registry.Route) (string, error) { return "", errors.New("no key") }
+ sw.precheckAgain()
+ sw.mustTargetRound(uk)
+ rd = *sw.report(moves.OfflineRoleTarget, "uk-box").Ready
+ if rd.DiskFree.Known || rd.StagePathArmed.Known || rd.LiveLink.Known || rd.ChannelBuild != "" {
+ t.Fatalf("unknown facts reported known: disk %+v stage %+v live %+v", rd.DiskFree, rd.StagePathArmed, rd.LiveLink)
+ }
+ _ = ctx
+}
+
+func TestSourcePrecheckFacts(t *testing.T) {
+ sw := newSSWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ sw.queue()
+ sw.mustStagePass(ie)
+ r := sw.report(moves.OfflineRoleSource, "ie-a")
+ sw.validFor(r)
+ if r.Phase != moves.OfflinePhasePrecheck || r.Precheck == nil || !r.Request.Equal(*sw.record().Move.PrecheckAt) {
+ t.Fatalf("precheck report %+v", r)
+ }
+ p := *r.Precheck
+ _, bytes, entries, err := archive.PlainDigest(filepath.Join(ie.dd, offHost), filepath.Join(ie.dd, offHost+".env"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if p.Layout != moves.LayoutFlat || r.Layout != moves.LayoutFlat || p.FilesBytes != bytes || p.FilesEntries != entries ||
+ !p.ArchiveEligible.Known || !p.ArchiveEligible.OK || !p.Strays.Known || len(p.Strays.Entries) != 0 ||
+ p.Build != sealVersion || !p.StagePathArmed.Known || !p.StagePathArmed.OK {
+ t.Fatalf("precheck %+v (archive %d bytes %d entries)", p, bytes, entries)
+ }
+ // An entry the archive refuses (a FIFO) is an ineligible archive with
+ // the archive's own reason, and the files are still counted.
+ if err := syscall.Mkfifo(filepath.Join(ie.dd, offHost, "pipe"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ sw.precheckAgain()
+ sw.mustStagePass(ie)
+ p = *sw.report(moves.OfflineRoleSource, "ie-a").Precheck
+ if !p.ArchiveEligible.Known || p.ArchiveEligible.OK || !strings.Contains(p.ArchiveEligible.Detail, "pipe") || p.FilesBytes != bytes || p.FilesEntries != entries {
+ t.Fatalf("an ineligible archive: %+v", p)
+ }
+ if err := os.Remove(filepath.Join(ie.dd, offHost, "pipe")); err != nil {
+ t.Fatal(err)
+ }
+ // A nested source reports its layout, and the stray entries beside its
+ // files directory with their sizes.
+ sw.nest(ie)
+ raw := layout.RawInstanceDir(ie.dd, offHost)
+ writeFile(t, filepath.Join(raw, "uploads-cache", "x"), "12345")
+ sw.precheckAgain()
+ sw.mustStagePass(ie)
+ r = sw.report(moves.OfflineRoleSource, "ie-a")
+ p = *r.Precheck
+ if p.Layout != moves.LayoutNested || r.Layout != moves.LayoutNested {
+ t.Fatalf("nested layout %q %q", p.Layout, r.Layout)
+ }
+ if len(p.Strays.Entries) != 1 || p.Strays.Entries[0] != (moves.StrayEntry{Path: filepath.Join(raw, "uploads-cache"), Kind: moves.StrayDir, Bytes: 5, Entries: 1, Known: true}) {
+ t.Fatalf("strays %+v", p.Strays)
+ }
+}
+
+func TestStrayFileRefusesBeforeStop(t *testing.T) {
+ sw := newSSWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ stray := filepath.Join(ie.dd, offHost+"-blobs")
+ writeFile(t, stray, "not a directory")
+ sw.queue()
+ sw.mustStagePass(ie)
+ p := sw.report(moves.OfflineRoleSource, "ie-a").Precheck
+ if p == nil || len(p.Strays.Entries) != 1 || p.Strays.Entries[0].Path != stray || p.Strays.Entries[0].Kind != moves.StrayFile || p.Strays.Entries[0].Bytes != int64(len("not a directory")) {
+ t.Fatalf("strays %+v", p)
+ }
+ // The fact is reported before anything stops: the row serves, unfenced.
+ if rt := ie.row(offHost); rt == nil || rt.HoldOp != "" || ie.svc.Fenced(offHost) {
+ t.Fatalf("the precheck stopped the instance: %+v", rt)
+ }
+ // And the start never removes a file: only an empty directory.
+ sw.toStaging()
+ sw.mustStagePass(ie)
+ sw.advance(time.Second)
+ sw.move(sw.proposed(func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineStartStaged(offID, "owner@example.com", at, provision.Receipt{})
+ }))
+ sw.mustStagePass(ie)
+ if !lexists(stray) {
+ t.Fatal("the start removed a stray file")
+ }
+ if r := sw.report(moves.OfflineRoleSource, "ie-a"); r.Phase != moves.OfflinePhaseStopped || r.Precheck != nil {
+ t.Fatalf("stopped %+v precheck %+v", r, r.Precheck)
+ }
+}
+
+func TestEmptyStrayDirsRemovedAtStartOnly(t *testing.T) {
+ sw := newSSWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ empty := filepath.Join(ie.dd, offHost+"-cache")
+ full := filepath.Join(ie.dd, offHost+"-blobs")
+ if err := os.MkdirAll(empty, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ writeFile(t, filepath.Join(full, "a"), "kept")
+ sw.queue()
+ sw.mustStagePass(ie)
+ strays := sw.report(moves.OfflineRoleSource, "ie-a").Precheck.Strays
+ if len(strays.Entries) != 2 || strays.Entries[0].Path != full || strays.Entries[1].Path != empty || strays.Entries[1].Entries != 0 || strays.Entries[0].Entries != 1 {
+ t.Fatalf("strays %+v", strays)
+ }
+ sw.toStaging()
+ sw.precheckAgain()
+ sw.mustStagePass(ie)
+ if !isDir(empty) || !isDir(full) {
+ t.Fatal("a stray directory was removed before the start")
+ }
+ sw.advance(time.Second)
+ sw.move(sw.proposed(func(at time.Time) func(*provision.Record) error {
+ return provision.OfflineStartStaged(offID, "owner@example.com", at, provision.Receipt{})
+ }))
+ sw.mustStagePass(ie)
+ r := sw.report(moves.OfflineRoleSource, "ie-a")
+ if r.Phase != moves.OfflinePhaseStopped || !r.Held || !r.Fenced || !r.Quiesced || r.Procs != 0 {
+ t.Fatalf("fixture: the stop does not hold: %+v", r)
+ }
+ if lexists(empty) || !isDir(full) {
+ t.Fatalf("after the start: empty dir present %v, full dir present %v", lexists(empty), isDir(full))
+ }
+ if r.Precheck == nil || !reflect.DeepEqual(r.Precheck.StraysRemoved, []string{empty}) || len(r.Precheck.StraysLeft) != 1 ||
+ r.Precheck.StraysLeft[0].Path != full || r.Precheck.StraysLeft[0].Detail != "not empty" {
+ t.Fatalf("stopped report's strays %+v", r.Precheck)
+ }
+ sw.validFor(r)
+ // A later pass keeps what the start removed.
+ sw.advance(time.Second)
+ sw.mustStagePass(ie)
+ if r := sw.report(moves.OfflineRoleSource, "ie-a"); r.Precheck == nil || !reflect.DeepEqual(r.Precheck.StraysRemoved, []string{empty}) {
+ t.Fatalf("a later pass lost the removal: %+v", r.Precheck)
+ }
+ // An operator's move removes nothing at its start.
+ sw2 := newSealWorld(t)
+ sw2.instance(sw2.ie)
+ opEmpty := filepath.Join(sw2.ie.dd, offHost+"-cache")
+ if err := os.MkdirAll(opEmpty, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ sw2.start()
+ sw2.mustPass(sw2.ie)
+ if !isDir(opEmpty) || sw2.report(moves.OfflineRoleSource, "ie-a").Precheck != nil {
+ t.Fatal("an operator move's start removed a stray directory")
+ }
+}
+
+// --- the dispatch ----------------------------------------------------------
+
+func TestQueuedRoundRunsOnlyLanePrecheckAndReady(t *testing.T) {
+ sw := newSSWorld(t)
+ ie, uk := sw.ie, sw.uk
+ ieB := sw.sourceBox("ie-b") // a second box of the source region, not pinned
+ ssIdentity(ieB, "i-ie-b")
+ sw.instance(ie)
+ ieB.tick()
+ sw.queue()
+ spy := sw.spyCopies()
+ sw.mustStagePass(ieB)
+ sw.mustStagePass(ie)
+ sw.mustTargetRound(uk)
+ if keys := spy.take(); len(keys) != 0 {
+ t.Fatalf("a queued move copied %v", keys)
+ }
+ if n := len(sw.list(sealUK, activator.MovesPrefix(offHost))); n != 0 {
+ t.Fatalf("%d objects staged while queued", n)
+ }
+ // The three reports, and nothing else.
+ keys, err := sw.s.List(context.Background(), moves.OfflineReportPrefix(offHost, offID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ sort.Strings(keys)
+ want := []string{
+ moves.OfflineReportKey(offHost, offID, moves.OfflineRoleLane, "ie-a"),
+ moves.OfflineReportKey(offHost, offID, moves.OfflineRoleSource, "ie-a"),
+ moves.OfflineReportKey(offHost, offID, moves.OfflineRoleTarget, "uk-box"),
+ }
+ sort.Strings(want)
+ if !reflect.DeepEqual(keys, want) {
+ t.Fatalf("reports %v, want %v", keys, want)
+ }
+ for _, role := range []string{moves.OfflineRoleLane, moves.OfflineRoleSource} {
+ sw.validFor(sw.report(role, "ie-a"))
+ }
+ sw.validFor(sw.report(moves.OfflineRoleTarget, "uk-box"))
+ // Nothing held, nothing fenced, nothing kept for a box the move does
+ // not name.
+ if rt := ie.row(offHost); rt == nil || rt.HoldOp != "" || ie.svc.Fenced(offHost) {
+ t.Fatalf("queued held or fenced the source: %+v", rt)
+ }
+ if ieB.w.cfg.Offline.keep(offHost) {
+ t.Fatal("a box the move does not name keeps its row")
+ }
+ // A replaced source (same box id, another instance) has no role: it
+ // writes nothing, and logs once.
+ ie.w.cfg.InstanceID = "i-replaced"
+ sw.precheckAgain()
+ before, _, _ := sw.s.Get(context.Background(), moves.OfflineReportKey(offHost, offID, moves.OfflineRoleSource, "ie-a"))
+ sw.mustStagePass(ie)
+ after, _, _ := sw.s.Get(context.Background(), moves.OfflineReportKey(offHost, offID, moves.OfflineRoleSource, "ie-a"))
+ if string(before) != string(after) {
+ t.Fatal("a replaced box reported")
+ }
+ if _, logged := ie.w.replaced.Load(offID); !logged {
+ t.Fatal("a replaced box did not log")
+ }
+ // The hold follows the participants too: a start holds the pinned
+ // source and not the other box of its region.
+ ie.w.cfg.InstanceID = ssSourceInstance
+ hold := offlineHold(ie.w.cfg)
+ holdB := offlineHold(ieB.w.cfg)
+ rec := sw.record()
+ rec.Move.State = provision.OfflineStopping
+ if _, _, ok := hold(rec); !ok {
+ t.Fatal("the pinned source does not hold at the start")
+ }
+ if _, _, ok := holdB(rec); ok {
+ t.Fatal("another box of the source region holds a pinned move")
+ }
+}
+
+func TestParticipantsReReportOnANewPrecheckStamp(t *testing.T) {
+ sw := newSSWorld(t)
+ ie, uk := sw.ie, sw.uk
+ sw.instance(ie)
+ sw.queue()
+ sw.mustStagePass(ie)
+ sw.mustTargetRound(uk)
+ src, tgt := sw.report(moves.OfflineRoleSource, "ie-a"), sw.report(moves.OfflineRoleTarget, "uk-box")
+ lane := sw.report(moves.OfflineRoleLane, "ie-a")
+ // The same stamp: nothing is written again.
+ sw.advance(time.Minute)
+ sw.mustStagePass(ie)
+ sw.mustTargetRound(uk)
+ if !sw.report(moves.OfflineRoleSource, "ie-a").At.Equal(src.At) || !sw.report(moves.OfflineRoleTarget, "uk-box").At.Equal(tgt.At) {
+ t.Fatal("a pass with the same precheck stamp reported again")
+ }
+ // Staging: the stager stages; the precheck and ready reports stand.
+ sw.toStaging()
+ sw.mustStagePass(ie)
+ sw.mustTargetRound(uk)
+ if st := sw.report(moves.OfflineRoleStager, "ie-a"); st.Phase != moves.OfflinePhaseStaged || !st.StagedKnown {
+ t.Fatalf("stager %+v", st)
+ }
+ if !sw.report(moves.OfflineRoleSource, "ie-a").At.Equal(src.At) || !sw.report(moves.OfflineRoleTarget, "uk-box").At.Equal(tgt.At) {
+ t.Fatal("staging rewrote a report its stamp already answers")
+ }
+ // A newer stamp: both report again for it; the lane stays.
+ sw.precheckAgain()
+ stamp := *sw.record().Move.PrecheckAt
+ sw.mustStagePass(ie)
+ sw.mustTargetRound(uk)
+ for _, r := range []moves.OfflineReport{sw.report(moves.OfflineRoleSource, "ie-a"), sw.report(moves.OfflineRoleTarget, "uk-box")} {
+ if !r.Request.Equal(stamp) || !r.Request.After(src.Request) {
+ t.Fatalf("%s answers %v, want %v", r.Role, r.Request, stamp)
+ }
+ sw.validFor(r)
+ }
+ if !sw.report(moves.OfflineRoleLane, "ie-a").At.Equal(lane.At) {
+ t.Fatal("the lane was probed again")
+ }
+}
+
+// --- the staged book -------------------------------------------------------
+
+// hookMirror is the production mirror with hooks around its copies.
+type hookMirror struct {
+ lease.MirrorStore
+ mu sync.Mutex
+ beforeCopy func(srcKey string)
+ afterCopy func(dstKey string)
+ failCopy func(srcKey string) error
+}
+
+func (h *hookMirror) Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) {
+ h.mu.Lock()
+ before, after, fail := h.beforeCopy, h.afterCopy, h.failCopy
+ h.mu.Unlock()
+ if fail != nil {
+ if err := fail(srcKey); err != nil {
+ return "", err
+ }
+ }
+ if before != nil {
+ before(srcKey)
+ }
+ etag, err := h.MirrorStore.Copy(ctx, srcBucket, srcKey, srcETag, dstBucket, dstKey)
+ if err == nil && after != nil {
+ after(dstKey)
+ }
+ return etag, err
+}
+
+func (h *hookMirror) set(f func(*hookMirror)) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ f(h)
+}
+
+// hooked arms b's mirror as the production NewS3Mirror wrapped in hooks,
+// on a fresh worker (an agent restart), keeping its identity.
+func (sw *sealWorld) hooked(b *offBox) *hookMirror {
+ sw.t.Helper()
+ inst := b.w.cfg.InstanceID
+ b.restart()
+ sw.arm(b)
+ ssIdentity(b, inst)
+ h := &hookMirror{}
+ b.w.cfg.Mirror = func(ctx context.Context, dst string) (lease.MirrorStore, error) {
+ m, err := lease.NewS3Mirror(ctx, sealIERegion, dst)
+ if err != nil {
+ return nil, err
+ }
+ h.MirrorStore = m
+ return h, nil
+ }
+ return h
+}
+
+// book is the stager's last report's book.
+func (sw *sealWorld) book() map[string]moves.StagedObject {
+ sw.t.Helper()
+ r := sw.report(moves.OfflineRoleStager, "ie-a")
+ if !r.StagedKnown || r.StagedBucket != sealUK || r.StagedRegion != sealUKRegion {
+ sw.t.Fatalf("no book: %+v", r)
+ }
+ return r.Staged
+}
+
+// stagedFacts is the staging prefix as the target lists it.
+func (sw *sealWorld) stagedFacts() moves.PrefixFacts {
+ sw.t.Helper()
+ vs, err := lease.NewVersionedS3(context.Background(), map[string]string{sealUK: sealUKRegion})
+ if err != nil {
+ sw.t.Fatal(err)
+ }
+ f, _ := prefixFacts(context.Background(), vs, sealUK, activator.MovesPrefix(offHost))
+ return f
+}
+
+func (sw *sealWorld) requireOwned(barrier bool) {
+ sw.t.Helper()
+ r := sw.report(moves.OfflineRoleStager, "ie-a")
+ m := sw.record().Move
+ if err := moves.StagedMatches(r, m.StageBucket, m.StageRegion, *m.StagedAt, sw.stagedFacts(), barrier); err != nil {
+ sw.t.Fatalf("the staged prefix is not the book: %v\nbook %+v", err, r.Staged)
+ }
+}
+
+func TestStageBookAcrossRounds(t *testing.T) {
+ sw := newSSWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ repl := sealPrefix + "/" + offHost + "/"
+ staged := activator.MovesPrefix(offHost)
+ sw.fake.Put(sealIE, repl+"extra/single", []byte("single part"))
+ sw.fake.PutMultipart(sealIE, repl+"extra/multi", []byte("a multipart upload"))
+ sw.fake.Put(sealIE, repl+"extra/doomed", []byte("deleted later"))
+ sw.queue()
+ sw.toStaging()
+ h := sw.hooked(ie)
+ sw.mustStagePass(ie)
+ book := sw.book()
+ if len(book) != len(sw.replicaRel()) {
+ t.Fatalf("book %d entries, replica %d", len(book), len(sw.replicaRel()))
+ }
+ sw.requireOwned(true)
+ single, multi := book["extra/single"], book["extra/multi"]
+ if single.ETag == "" || multi.ETag == "" || strings.HasSuffix(multi.ETag, "-1") {
+ t.Fatalf("book entries %+v %+v", single, multi)
+ }
+ r1 := sw.report(moves.OfflineRoleStager, "ie-a")
+ if r1.Rounds != 1 || r1.RoundCopiedBytes == 0 || r1.RoundDeferredObjects != 0 || !r1.StagedRequest.Equal(r1.StageRequest) {
+ t.Fatalf("round 1 report %+v", r1)
+ }
+ sw.validFor(r1)
+
+ // Round 2 copies nothing: same-ETag skips and the receipt-vouched
+ // multipart skip are in the book with the destination's ETags. A key
+ // whose copy is deferred is in the book, deferred, with nothing copied.
+ sw.fake.Put(sealIE, repl+"extra/busy", []byte("busy v1"))
+ h.set(func(h *hookMirror) {
+ h.beforeCopy = func(k string) {
+ if k == repl+"extra/busy" {
+ sw.fake.Put(sealIE, repl+"extra/busy", []byte("busy v2"))
+ }
+ }
+ })
+ sw.restage()
+ sw.mustStagePass(ie)
+ book = sw.book()
+ if book["extra/single"] != single || book["extra/multi"] != multi {
+ t.Fatalf("skips changed the book: %+v %+v", book["extra/single"], book["extra/multi"])
+ }
+ if b := book["extra/busy"]; !b.Deferred || b.ETag != "" {
+ t.Fatalf("the deferred key %+v", b)
+ }
+ if r := sw.report(moves.OfflineRoleStager, "ie-a"); r.RoundDeferredObjects != 1 || r.Rounds != 2 {
+ t.Fatalf("round 2 %+v", r)
+ }
+ sw.requireOwned(false)
+ // Round 3: the deferred key is copied; a deleted key is gone.
+ h.set(func(h *hookMirror) { h.beforeCopy = nil })
+ sw.fake.BareDelete(sealIE, repl+"extra/doomed")
+ sw.restage()
+ sw.mustStagePass(ie)
+ book = sw.book()
+ if b := book["extra/busy"]; b.Deferred || b.ETag == "" {
+ t.Fatalf("the deferred key was not copied: %+v", b)
+ }
+ if _, ok := book["extra/doomed"]; ok {
+ t.Fatal("a deleted key is in the book")
+ }
+ sw.requireOwned(true)
+
+ // A key deleted by a round that crashed before its report: the report
+ // still lists it, the restarted stager seeds from the report, and the
+ // next complete round drops it.
+ sw.fake.BareDelete(sealIE, repl+"extra/single")
+ ie.w.cfg.stageFault = func(string) error { return errors.New("crash before the report") }
+ sw.restage()
+ if err := sw.stagePass(ie); err == nil {
+ t.Fatal("no crash")
+ }
+ if _, ok := sw.book()["extra/single"]; !ok {
+ t.Fatal("fixture: the crashed round reported")
+ }
+ if _, ok := sw.fake.Current(sealUK, staged+"extra/single"); ok {
+ t.Fatal("fixture: the crashed round did not delete")
+ }
+ h = sw.hooked(ie) // the restart
+ sw.restage()
+ sw.mustStagePass(ie)
+ if _, ok := sw.book()["extra/single"]; ok {
+ t.Fatal("a key a crashed round deleted is still in the book")
+ }
+ sw.requireOwned(true)
+
+ // A deferred key never copied whose source vanished is dropped.
+ sw.fake.Put(sealIE, repl+"extra/flaky", []byte("flaky v1"))
+ h.set(func(h *hookMirror) {
+ h.beforeCopy = func(k string) {
+ if k == repl+"extra/flaky" {
+ sw.fake.Put(sealIE, repl+"extra/flaky", []byte("flaky v2"))
+ }
+ }
+ })
+ sw.restage()
+ sw.mustStagePass(ie)
+ if b := sw.book()["extra/flaky"]; !b.Deferred || b.ETag != "" {
+ t.Fatalf("fixture: flaky %+v", b)
+ }
+ h.set(func(h *hookMirror) { h.beforeCopy = nil })
+ sw.fake.BareDelete(sealIE, repl+"extra/flaky")
+ sw.restage()
+ sw.mustStagePass(ie)
+ if _, ok := sw.book()["extra/flaky"]; ok {
+ t.Fatal("a deferred key whose source vanished is still in the book")
+ }
+
+ // An overwrite of the destination between the copy and the report:
+ // the book keeps the copy's ETag, so the ownership check catches it.
+ sw.fake.Put(sealIE, repl+"extra/overwritten", []byte("ours"))
+ h.set(func(h *hookMirror) {
+ h.afterCopy = func(k string) {
+ if k == staged+"extra/overwritten" {
+ sw.fake.Put(sealUK, k, []byte("someone else's"))
+ }
+ }
+ })
+ sw.restage()
+ sw.mustStagePass(ie)
+ ours := sw.book()["extra/overwritten"]
+ if theirs, _ := sw.fake.Current(sealUK, staged+"extra/overwritten"); string(theirs) != "someone else's" || ours.ETag != md5hexOf("ours") {
+ t.Fatalf("book %+v, destination %q", ours, theirs)
+ }
+ r := sw.report(moves.OfflineRoleStager, "ie-a")
+ m := sw.record().Move
+ if err := moves.StagedMatches(r, m.StageBucket, m.StageRegion, *m.StagedAt, sw.stagedFacts(), false); !errors.Is(err, moves.ErrStagedMismatch) {
+ t.Fatalf("an overwritten destination matched the book: %v", err)
+ }
+ h.set(func(h *hookMirror) { h.afterCopy = nil })
+ // The next round copies ours again (the ETags differ) and converges.
+ sw.restage()
+ sw.mustStagePass(ie)
+ sw.requireOwned(true)
+
+ // A crash before the report: the next round re-copies what it must and
+ // the book converges on the staged prefix.
+ sw.fake.Put(sealIE, repl+"extra/late", []byte("late"))
+ ie.w.cfg.stageFault = func(string) error { return errors.New("crash before the report") }
+ sw.restage()
+ if err := sw.stagePass(ie); err == nil {
+ t.Fatal("no crash")
+ }
+ sw.hooked(ie)
+ sw.restage()
+ sw.mustStagePass(ie)
+ if _, ok := sw.book()["extra/late"]; !ok {
+ t.Fatal("the book did not converge after a crash")
+ }
+ sw.requireOwned(true)
+
+ // The book is seeded from the last report: a restarted stager whose
+ // round fails before it decides anything (the replica cannot be
+ // listed) reports exactly the book it was seeded with.
+ failList := func() {
+ sw.fake.FailOn = func(op, key string) error {
+ if op == "list" && strings.HasPrefix(key, repl) {
+ return errors.New("injected list failure")
+ }
+ return nil
+ }
+ }
+ seeded := sw.book()
+ if len(seeded) == 0 {
+ t.Fatal("fixture: an empty book")
+ }
+ sw.hooked(ie)
+ failList()
+ sw.restage()
+ if err := sw.stagePass(ie); err == nil {
+ t.Fatal("fixture: the round did not fail")
+ }
+ sw.fake.FailOn = nil
+ if r := sw.report(moves.OfflineRoleStager, "ie-a"); r.Phase != moves.OfflinePhaseFailed || !reflect.DeepEqual(r.Staged, seeded) {
+ t.Fatalf("a restarted stager's failed round reports %+v, want the seeded book", r.Staged)
+ }
+ // A stager report of another destination seeds nothing: a book whose
+ // bucket is not the record's is not this move's.
+ key := moves.OfflineReportKey(offHost, offID, moves.OfflineRoleStager, "ie-a")
+ bad := sw.report(moves.OfflineRoleStager, "ie-a")
+ bad.StagedBucket = "data-fr"
+ bad.Staged["ghost"] = moves.StagedObject{Size: 1, ETag: "x"}
+ body, _ := json.Marshal(bad)
+ if err := moves.PutReplace(context.Background(), sw.s, key, body); err != nil {
+ t.Fatal(err)
+ }
+ sw.hooked(ie)
+ failList()
+ sw.restage()
+ if err := sw.stagePass(ie); err == nil {
+ t.Fatal("fixture: the round did not fail")
+ }
+ sw.fake.FailOn = nil
+ if b := sw.book(); len(b) != 0 {
+ t.Fatalf("a report of another destination seeded the book: %+v", b)
+ }
+}
+
+// md5hexOf is a single-part object's ETag.
+func md5hexOf(s string) string {
+ sum := md5.Sum([]byte(s))
+ return hex.EncodeToString(sum[:])
+}
+
+// TestObservedMirrorEmitsEveryDecision (the box half): MirrorPrefixKeep
+// through the production NewS3Mirror wrapper and stageDeleteGate leaves
+// exactly what the observed round leaves, receipts included, for two
+// identical replicas.
+func TestObservedMirrorEmitsEveryDecision(t *testing.T) {
+ sw := newSealWorld(t)
+ ctx := context.Background()
+ for _, h := range []string{"a.example.com", "b.example.com"} {
+ sw.fake.Put(sealIE, "instances/"+h+"/single", []byte("single"))
+ sw.fake.PutMultipart(sealIE, "instances/"+h+"/multi", []byte("multipart body"))
+ sw.fake.Put(sealIE, "instances/"+h+"/lock.json", []byte("lease"))
+ sw.fake.Put(sealUK, "moves/"+h+"/gone", []byte("gone"))
+ }
+ m, err := lease.NewS3Mirror(ctx, sealIERegion, sealUKRegion)
+ if err != nil {
+ t.Fatal(err)
+ }
+ gate := func() *stageDeleteGate {
+ return &stageDeleteGate{MirrorStore: m, check: func(context.Context) error { return nil }}
+ }
+ var events []lease.CopyEvent
+ for round := 0; round < 2; round++ {
+ ra, err := lease.MirrorPrefixKeep(ctx, gate(), sealIE, "instances/a.example.com", sealUK, "moves/a.example.com/", lease.Receipts{Bucket: sealIE, Key: "move-receipts/a.example.com.json"}, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ events = nil
+ rb, err := lease.MirrorPrefixKeepObserved(ctx, gate(), sealIE, "instances/b.example.com", sealUK, "moves/b.example.com/", lease.Receipts{Bucket: sealIE, Key: "move-receipts/b.example.com.json"}, nil,
+ func(e lease.CopyEvent) { events = append(events, e) })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(ra, rb) {
+ t.Fatalf("round %d: plain %+v observed %+v", round, ra, rb)
+ }
+ }
+ rel := func(h string) map[string]string {
+ out := map[string]string{}
+ for k, o := range sw.list(sealUK, "moves/"+h+"/") {
+ out[strings.TrimPrefix(k, "moves/"+h+"/")] = o.ETag
+ }
+ return out
+ }
+ if a, b := rel("a.example.com"), rel("b.example.com"); !reflect.DeepEqual(a, b) {
+ t.Fatalf("destinations differ: %v %v", a, b)
+ }
+ ra, _ := sw.fake.Current(sealIE, "move-receipts/a.example.com.json")
+ rb, _ := sw.fake.Current(sealIE, "move-receipts/b.example.com.json")
+ if strings.ReplaceAll(string(ra), "a.example.com", "x") != strings.ReplaceAll(string(rb), "b.example.com", "x") {
+ t.Fatalf("receipts differ:\n%s\n%s", ra, rb)
+ }
+ // The second round decided both keys without copying: the receipt
+ // vouched for the multipart copy.
+ got := map[string]bool{}
+ for _, e := range events {
+ if e.Unchanged {
+ got[e.Rel] = true
+ }
+ }
+ if !got["single"] || !got["multi"] || !events[len(events)-1].Complete {
+ t.Fatalf("second round events %+v", events)
+ }
+}
+
+// --- the sealer's revert ---------------------------------------------------
+
+func TestSealerPublishesTombstoneLiftedBeforeCleanup(t *testing.T) {
+ sw := newSealWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ sw.startAndSeal("ie-a", ie)
+ sw.mustPass(ie)
+ if seal, _, ok := sw.sealRecord(); !ok || seal.FinalCopiedBytes == 0 {
+ t.Fatalf("the seal records no final copy: %+v", seal)
+ }
+ sw.abort()
+ offlineSealFaults.Store(ie.w, func(point string) error {
+ if point == "revert-lift" {
+ return errors.New("crash after the lift")
+ }
+ return nil
+ })
+ t.Cleanup(func() { offlineSealFaults.Delete(ie.w) })
+ if err := sw.pass(ie); err == nil {
+ t.Fatal("no crash")
+ }
+ r := sw.report(moves.OfflineRoleSealer, "ie-a")
+ if r.Phase != moves.OfflinePhaseReverting || !r.TombstoneLifted || r.TombstoneLiftedAt.IsZero() {
+ t.Fatalf("after the lift: %+v", r)
+ }
+ if owner, expires, _ := sw.lock(); !lease.IsLiftedTombstone(owner, expires) {
+ t.Fatal("fixture: the tombstone is not lifted")
+ }
+ if n := len(sw.list(sealUK, activator.MovesPrefix(offHost))); n == 0 {
+ t.Fatal("the cleanup ran before the reverting report")
+ }
+ first := r.TombstoneLiftedAt
+ // A retry keeps the first lift time.
+ sw.advance(time.Minute)
+ if err := sw.pass(ie); err == nil {
+ t.Fatal("no crash")
+ }
+ if r := sw.report(moves.OfflineRoleSealer, "ie-a"); !r.TombstoneLiftedAt.Equal(first) || !r.At.After(first) {
+ t.Fatalf("the retry moved the lift time: %+v", r)
+ }
+ offlineSealFaults.Delete(ie.w)
+ sw.advance(time.Minute)
+ sw.mustPass(ie)
+ r = sw.report(moves.OfflineRoleSealer, "ie-a")
+ if r.Phase != moves.OfflinePhaseReverted || !r.TombstoneLifted || !r.TombstoneLiftedAt.Equal(first) {
+ t.Fatalf("reverted %+v", r)
+ }
+ if n := len(sw.list(sealUK, activator.MovesPrefix(offHost))); n != 0 {
+ t.Fatalf("%d staged objects left", n)
+ }
+}
+
+// --- the status block ------------------------------------------------------
+
+func TestStatusHostsBlockThroughTheRealPath(t *testing.T) {
+ sw := newSSWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ ctx := context.Background()
+ routes, err := ie.reg.List()
+ if err != nil {
+ t.Fatal(err)
+ }
+ now := time.Now()
+ p := status.Probes{MachineID: "machine-ie", InstanceID: ssSourceInstance, Release: "v1"}
+ getenv := func(k string) string {
+ if k == moveLanesEnv {
+ return "data-uk, data-fr data-uk"
+ }
+ return ""
+ }
+ selfServeStatusProbes(ctx, &p, ie.w, "ie-a", getenv, routes, now)
+ if err := statusTick(ctx, sw.s, ie.reg, p, true, "ie-a"); err != nil {
+ t.Fatal(err)
+ }
+ boxes, complete, err := status.LoadBoxesComplete(ctx, sw.s)
+ if err != nil || !complete {
+ t.Fatalf("boxes %v %v", complete, err)
+ }
+ var mine *status.BoxMetrics
+ for _, b := range boxes {
+ if b.Label == "ie-a" {
+ mine = b.Metrics
+ }
+ }
+ if mine == nil {
+ t.Fatalf("no report for ie-a: %+v", boxes)
+ }
+ h, ok := mine.Hosts[offHost]
+ if mine.BoxID != "ie-a" || mine.InstanceID != ssSourceInstance || !reflect.DeepEqual(mine.Lanes, []string{"data-fr", "data-uk"}) || !ok || !h.Known {
+ t.Fatalf("metrics box %q instance %q lanes %v host %+v", mine.BoxID, mine.InstanceID, mine.Lanes, h)
+ }
+ if !moves.ParticipantStillPresent(ssParticipants().Source[0], mine) {
+ t.Fatal("the pinned source is not present by its own report")
+ }
+ fleet, _, err := sw.s.Get(ctx, "status/fleet.json")
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, leak := range []string{offHost, ssSourceInstance, "data-uk", `"box_id"`, `"hosts"`} {
+ if strings.Contains(string(fleet), leak) {
+ t.Fatalf("fleet.json carries %s", leak)
+ }
+ }
+}
+
+func TestHostFactsCarryTheLifetimeAndAges(t *testing.T) {
+ sw := newSSWorld(t)
+ ie := sw.ie
+ sw.instance(ie)
+ ctx := context.Background()
+ rt := ie.row(offHost)
+ row := *rt
+ row.Incarnation = 7
+ routes := []registry.Route{row, {Host: "static.example.com", Kind: "static"}, {Host: "alias.example.com", Platform: true}}
+ t0 := time.Date(2026, 10, 22, 9, 0, 0, 0, time.UTC)
+ facts := ie.w.hostFacts(ctx, routes, t0)
+ if len(facts) != 1 {
+ t.Fatalf("facts for %v", facts)
+ }
+ f := facts[offHost]
+ var replicaBytes int64
+ for _, o := range sw.replicaRel() {
+ replicaBytes += o.Size
+ }
+ _, bytes, entries, err := archive.PlainDigest(filepath.Join(ie.dd, offHost), filepath.Join(ie.dd, offHost+".env"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := status.HostFacts{CollectedAt: t0.Unix(), Known: true, Incarnation: 7, ReplicaBytes: replicaBytes, ReplicaObjects: len(sw.replicaRel()),
+ ReplicaCollectedAt: t0.Unix(), FilesBytes: bytes, FilesEntries: entries}
+ if !reflect.DeepEqual(f, want) {
+ t.Fatalf("facts\n got %+v\nwant %+v", f, want)
+ }
+ // A listing that fails keeps the last one, with its own age, and the
+ // facts are not known.
+ sw.fake.FailOn = func(op, key string) error {
+ if op == "list" {
+ return errors.New("injected list failure")
+ }
+ return nil
+ }
+ t.Cleanup(func() { sw.fake.FailOn = nil })
+ t1 := t0.Add(5 * time.Minute)
+ f = ie.w.hostFacts(ctx, routes, t1)[offHost]
+ if f.Known || f.CollectedAt != t1.Unix() || f.ReplicaCollectedAt != t0.Unix() || f.ReplicaBytes != replicaBytes || !strings.Contains(f.Err, "replica") {
+ t.Fatalf("after a failed listing %+v", f)
+ }
+ sw.fake.FailOn = nil
+ // Strays are listed; a box that cannot take part publishes nothing.
+ writeFile(t, filepath.Join(ie.dd, offHost+"-blobs"), "x")
+ if f := ie.w.hostFacts(ctx, routes, t1)[offHost]; !reflect.DeepEqual(f.Strays, []string{filepath.Join(ie.dd, offHost+"-blobs")}) || !f.Known {
+ t.Fatalf("strays %+v", f)
+ }
+ ie.w.cfg.OfflineCapable = false
+ if ie.w.hostFacts(ctx, routes, t1) != nil {
+ t.Fatal("an incapable box published host facts")
+ }
+
+ // measureFiles: the archive's rules, over the cap and on to twice it.
+ dir := t.TempDir()
+ files, env := filepath.Join(dir, "files"), filepath.Join(dir, "x.env")
+ writeFile(t, filepath.Join(files, "a"), "0123456789")
+ writeFile(t, filepath.Join(files, archive.MarkerName), "gen")
+ writeFile(t, env, "ENV=1")
+ if err := os.Symlink("a", filepath.Join(files, "link")); err != nil {
+ t.Fatal(err)
+ }
+ if b, n, over, trunc, err := measureFiles(ctx, files, env, time.Minute, 100); err != nil || b != 15 || n != 2 || over || trunc {
+ t.Fatalf("under the cap: %d %d %v %v %v", b, n, over, trunc, err)
+ }
+ if _, _, over, trunc, _ := measureFiles(ctx, files, env, time.Minute, 12); !over || trunc {
+ t.Fatalf("over the cap: over %v truncated %v", over, trunc)
+ }
+ if _, _, over, trunc, _ := measureFiles(ctx, files, env, time.Minute, 7); !over || !trunc {
+ t.Fatalf("at twice the cap: over %v truncated %v", over, trunc)
+ }
+ if _, _, _, trunc, _ := measureFiles(ctx, files, "", 0, 100); !trunc {
+ t.Fatal("a spent budget is not truncated")
+ }
+ if b, n, _, _, err := measureFiles(ctx, filepath.Join(dir, "absent"), "", time.Minute, 100); err != nil || b != 0 || n != 0 {
+ t.Fatalf("a missing files directory: %d %d %v", b, n, err)
+ }
+}
+
+func TestVersionedArmedForOffline(t *testing.T) {
+ getenv := func(k string) string {
+ if k == "CARLOS_HIBERNATE_BUCKET" {
+ return "data"
+ }
+ return ""
+ }
+ cfg := moveCfg{Box: "uk-a", DataDir: t.TempDir()}
+ armOffline(&cfg, cfg.DataDir, nil, nil, nil, hostwrite.Config{}, getenv)
+ if cfg.Versioned == nil || cfg.LaneProbe == nil || cfg.DiskFree == nil || cfg.StagePathArmed == nil || cfg.BoxID != "uk-a" || !cfg.OfflineCapable {
+ t.Fatalf("armOffline with Phase 1 off: versioned %v lane %v disk %v stage %v box id %q", cfg.Versioned != nil, cfg.LaneProbe != nil, cfg.DiskFree != nil, cfg.StagePathArmed != nil, cfg.BoxID)
+ }
+ // What armPhase1 set is kept.
+ called := false
+ mine := func(context.Context, string, string, string) error { called = true; return nil }
+ cfg = moveCfg{Box: "uk-a", BoxID: "uk-a-id", LaneProbe: mine}
+ armOffline(&cfg, t.TempDir(), nil, nil, nil, hostwrite.Config{}, getenv)
+ if err := cfg.LaneProbe(context.Background(), "", "", ""); err != nil || !called || cfg.BoxID != "uk-a-id" {
+ t.Fatal("armOffline replaced a lane probe or a box id it was given")
+ }
+}