amadan

carlos / platform Public

console: the reports gate judges the source's status facts before the disk decision uses the replica size

The facts are judged as the self-serve door judges them
(selfServeSourceFacts): this lifetime and fresh, then over the cap (the
door's sentence, final even on a cut walk), then missing, unknown or cut
short ("could not determine the replica's size on <label>"); facts too
old or of another lifetime are waited for. A missing or unknown fact no
longer reads as a replica of size 0. TestReportsJudgeTheSourceFacts
(RED before: all three passed into staging).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 2ced088e5bf9946f36974c8e54e181b74713a156 parent a905f7c
3 files changed, +110 −12
  • internal/console/selfserve_checks.go +1 −0
  • internal/console/selfserve_job.go +31 −12
  • internal/e2e/selfserve/job_phases_test.go +78 −0
diff --git a/internal/console/selfserve_checks.go b/internal/console/selfserve_checks.go
index 82bea2d04..154ece8c3 100644
--- a/internal/console/selfserve_checks.go
+++ b/internal/console/selfserve_checks.go
@@ -33,6 +33,7 @@ const (
factSourceFacts = "the source's facts"
factTargetFacts = "the target's facts"
factStagedCopy = "the staged copy"
+ factReplica = "the replica's size"
selfServeReadyMiB = 16 << 20
)
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index 6006ac08e..58fc90009 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -25,6 +25,7 @@ import (
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/status"
)
// The waits' allowances and bounds (spec §6.2, §7.2).
@@ -580,7 +581,7 @@ func (s *Server) selfServeExpiryReason(ctx context.Context, row *SelfServeMove,
if mv := selfServeOurs(rec, row); mv != nil {
if ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv); err == nil {
if fleet, ok := s.readOfflineFleet(ctx); ok {
- if v := s.selfServeReportsVerdict(row, mv, ev, fleet); v.waiting != "" {
+ if v := s.selfServeReportsVerdict(row, mv, ev, fleet, now); v.waiting != "" {
return v.waiting
}
}
@@ -892,16 +893,30 @@ type ssVerdict struct {
fail, waiting string
}
-// selfServeReplicaBytes is the replica's size as the source box last
-// measured it for the host.
-func selfServeReplicaBytes(row *SelfServeMove, f offlineFleet) int64 {
- if len(row.Participants.Source) == 0 {
- return 0
+// selfServeJudgedFacts is the source's status facts for the host, judged
+// as the self-serve door judges them (selfServeSourceFacts): a report for
+// this lifetime collected recently, then over the cap (final, even on a cut
+// walk), then a walk that did not finish or was cut short. Over the cap
+// fails the move with the door's sentence; facts that are missing or not
+// known fail it as a replica size that could not be determined; facts too
+// old or of another lifetime are waited for (the box reports again), the
+// reports' deadline bounding the wait. Only judged facts give the replica
+// size a decision compares with.
+func selfServeJudgedFacts(row *SelfServeMove, f offlineFleet, now time.Time) (status.HostFacts, ssVerdict) {
+ label := ""
+ if len(row.Participants.Source) > 0 {
+ label = row.Participants.Source[0].Label
}
- if m := f.metrics[row.Participants.Source[0].Label]; m != nil {
- return m.Hosts[row.Host].ReplicaBytes
+ facts, ref := selfServeSourceFacts(f.metrics[label], row.Host, row.Incarnation, now)
+ switch {
+ case ref == nil:
+ return facts, ssVerdict{pass: true}
+ case ref.msg == msgSelfServeTooLarge:
+ return facts, ssVerdict{fail: ref.msg}
+ case ref.msg == msgSelfServeSourceStale:
+ return facts, ssVerdict{waiting: couldNotDetermine(factReplica, label)}
}
- return 0
+ return facts, ssVerdict{fail: couldNotDetermine(factReplica, label)}
}
// selfServeReportsVerdict is the reports phase's decision: the lane from
@@ -909,10 +924,14 @@ func selfServeReplicaBytes(row *SelfServeMove, f offlineFleet) int64 {
// from its pinned participant and answering the record's precheck stamp
// (Request == PrecheckAt), read before anything later overwrites their
// keys; then the facts.
-func (s *Server) selfServeReportsVerdict(row *SelfServeMove, mv *provision.MoveState, ev offlineEvidence, f offlineFleet) ssVerdict {
+func (s *Server) selfServeReportsVerdict(row *SelfServeMove, mv *provision.MoveState, ev offlineEvidence, f offlineFleet, now time.Time) ssVerdict {
if len(row.Participants.Source) == 0 || mv.PrecheckAt == nil {
return ssVerdict{fail: msgSelfServeCouldNotStart}
}
+ facts, v := selfServeJudgedFacts(row, f, now)
+ if !v.pass {
+ return v
+ }
host, src, tgt, at := row.Host, row.Participants.Source[0], row.Participants.Target, *mv.PrecheckAt
lane, ok := ev.get(moves.OfflineRoleLane, src.Label)
if !ok || lane.Lane == nil {
@@ -934,7 +953,7 @@ func (s *Server) selfServeReportsVerdict(row *SelfServeMove, mv *provision.MoveS
if !ok || rd.Phase != moves.OfflinePhaseReady || !rd.Request.Equal(at) {
return ssVerdict{waiting: couldNotDetermine(factTargetFacts, tgt.Label)}
}
- if why := targetReadyDecision(host, tgt.Label, rd.Ready, selfServeReplicaBytes(row, f), true); why != "" {
+ if why := targetReadyDecision(host, tgt.Label, rd.Ready, facts.ReplicaBytes, true); why != "" {
return ssVerdict{fail: why}
}
return ssVerdict{pass: true}
@@ -963,7 +982,7 @@ func (s *Server) selfServeReports(ctx context.Context, t *leaseToken, row *SelfS
if err != nil {
return err
}
- v := s.selfServeReportsVerdict(row, mv, ev, fleet)
+ v := s.selfServeReportsVerdict(row, mv, ev, fleet, now)
switch {
case v.fail != "":
return s.selfServeAbort(row, v.fail, now)
diff --git a/internal/e2e/selfserve/job_phases_test.go b/internal/e2e/selfserve/job_phases_test.go
index 93bcae47b..e1de7c20d 100644
--- a/internal/e2e/selfserve/job_phases_test.go
+++ b/internal/e2e/selfserve/job_phases_test.go
@@ -499,3 +499,81 @@ func TestCutHandoffSurvivesACrash(t *testing.T) {
}
})
}
+
+// editSourceFacts rewrites the source box's status block for the host, as
+// the box's status tick would have published it.
+func (w *jobWorld) editSourceFacts(edit func(hosts map[string]status.HostFacts)) {
+ w.t.Helper()
+ boxes, _, err := status.LoadBoxesComplete(w.ctx, w.raw)
+ if err != nil {
+ w.t.Fatal(err)
+ }
+ for _, b := range boxes {
+ if b.Label != host.E2ESourceBox || b.Metrics == nil {
+ continue
+ }
+ if b.Metrics.Hosts == nil {
+ b.Metrics.Hosts = map[string]status.HostFacts{}
+ }
+ edit(b.Metrics.Hosts)
+ b.GeneratedAt = time.Now().UTC()
+ if err := status.WriteBox(w.ctx, w.raw, b); err != nil {
+ w.t.Fatal(err)
+ }
+ return
+ }
+ w.t.Fatal("no status block for the source")
+}
+
+// TestReportsJudgeTheSourceFacts: the reports gate judges the source's
+// status facts for the host before any decision uses them (the replica's
+// size is what the target's free disk is compared with): facts over the
+// cap abort "this account's files are too large to move this way", even on
+// a cut walk; missing or unknown facts abort "could not determine the
+// replica's size on ie-box". None of them passes the disk check as size 0.
+func TestReportsJudgeTheSourceFacts(t *testing.T) {
+ for _, c := range []struct {
+ name, reason string
+ edit func(hosts map[string]status.HostFacts)
+ }{
+ {"missing", "could not determine the replica's size on ie-box", func(h map[string]status.HostFacts) { delete(h, host.E2EHost) }},
+ {"unknown", "could not determine the replica's size on ie-box", func(h map[string]status.HostFacts) {
+ f := h[host.E2EHost]
+ f.Known = false
+ h[host.E2EHost] = f
+ }},
+ {"over the cap on a cut walk", "this account's files are too large to move this way", func(h map[string]status.HostFacts) {
+ f := h[host.E2EHost]
+ f.OverCap, f.Truncated, f.Known = true, true, false
+ h[host.E2EHost] = f
+ }},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ w := newJobWorld(t)
+ row := w.toReports()
+ w.editSourceFacts(c.edit)
+ w.consoleUntil(15, "finished", func() bool { r := w.row(row.ID); return r.Outcome != "" && !r.ReleasePending })
+ r := w.row(row.ID)
+ if r.Outcome != "aborted" || r.OutcomeReason != c.reason || w.hits["stage:write"] != 0 {
+ t.Fatalf("finished %s, stage writes %d; want %q", describe(r), w.hits["stage:write"], c.reason)
+ }
+ })
+ }
+ t.Run("of another lifetime, waited for", func(t *testing.T) {
+ w := newJobWorld(t)
+ row := w.toReports()
+ w.editSourceFacts(func(h map[string]status.HostFacts) {
+ f := h[host.E2EHost]
+ f.Incarnation++
+ h[host.E2EHost] = f
+ })
+ for i := 0; i < 4; i++ {
+ w.tick()
+ }
+ if r := w.row(row.ID); r.Phase != "reports" || r.Outcome != "" || w.hits["stage:write"] != 0 {
+ t.Fatalf("facts of another lifetime decided the move: %s", describe(r))
+ }
+ w.up()
+ w.phase(row.ID, "staging", 20) // the box reports again for this lifetime
+ })
+}