amadan

carlos / platform Public

offline return: the disk rule sizes the data the return brings (final review F3)

returnIncomingBytes judges the current source's status facts for the host
as the forward's gates do (this lifetime, fresh, known, not cut short);
return mode's disk rule needs twice the larger of that and the leftovers
listed in the target bucket. Facts the console cannot judge refuse with
"could not determine the replica's size on <box>", never as zero.

Witness: TestReturnDiskUsesCurrentSourceSize (cleaned leftovers, growth
since the forward, leftovers the larger, four unknown-size cases, a small
replica accepted; each at the stage door and a direct start). The return
fixture publishes the current source's facts.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude bb900ba2c0d4774d0888b2b95411801223e8cdd0 parent 47567b3
2 files changed, +130 −9
  • internal/console/offlinereturn.go +44 −7
  • internal/console/offlinereturn_test.go +86 −2
diff --git a/internal/console/offlinereturn.go b/internal/console/offlinereturn.go
index 5dcf11ad9..f4c6f19e0 100644
--- a/internal/console/offlinereturn.go
+++ b/internal/console/offlinereturn.go
@@ -253,7 +253,11 @@ func (s *Server) readReturnReport(ctx context.Context, host, returnOf string, o
// empty, for a nested forward; absent for a flat one;
// - park directories: absent or the forward's own source/ and sealer/;
// - everything else: the ordinary rules.
-func returnReadyDecision(host, label string, o *moves.OpenRecord, r *moves.TargetReadyReport, want targetWant) string {
+//
+// incoming is the replica the return brings, as the current source's judged
+// facts size it (returnIncomingBytes); the disk rule compares the free disk
+// with twice the larger of that and the leftovers listed (final review F3).
+func returnReadyDecision(host, label string, o *moves.OpenRecord, r *moves.TargetReadyReport, want targetWant, incoming int64) string {
if r == nil || r.Leftovers == nil || r.Leftovers.ForwardID != o.MoveID {
return couldNotDetermine(factTargetFacts, label)
}
@@ -324,7 +328,7 @@ func returnReadyDecision(host, label string, o *moves.OpenRecord, r *moves.Targe
switch {
case !r.ParkDir.Known:
return couldNotDetermine(factParkDir, label)
- case !r.DiskFree.Known || r.DiskFreeBytes < 2*returnReplicaBytes(r):
+ case !r.DiskFree.Known || r.DiskFreeBytes < 2*max(incoming, returnReplicaBytes(r)):
return ssDiskLow(host)
case !r.MovesPrefix.Known:
return couldNotDetermine(factMovesPrefix, label)
@@ -351,9 +355,11 @@ func returnReadyDecision(host, label string, o *moves.OpenRecord, r *moves.Targe
return ""
}
-// returnReplicaBytes is the replica a return carries, as the report lists
-// the target bucket's instances prefix (the forward's leftover replica,
-// which the return's newer copy replaces): the free disk must be twice it.
+// returnReplicaBytes is the forward's leftover replica, as the report lists
+// the target bucket's instances prefix (which the return's newer copy
+// replaces). It is a lower bound only: the leftovers may be cleaned away or
+// the application grown since, so the disk rule takes the larger of it and
+// the current source's size (returnIncomingBytes).
func returnReplicaBytes(r *moves.TargetReadyReport) int64 {
var n int64
for _, o := range r.InstancesPrefix.Objects {
@@ -362,6 +368,32 @@ func returnReplicaBytes(r *moves.TargetReadyReport) int64 {
return n
}
+// returnIncomingBytes is the replica a return brings: the current source's
+// replica size from its status facts for host, judged as the forward's
+// gates judge them (this lifetime, collected recently, a walk that finished
+// and was not cut short). Facts the console cannot judge refuse as the
+// replica's size not determined, never as zero (final review F3).
+func returnIncomingBytes(sel offlineSelection, host string, inc uint64, now time.Time) (int64, *offlineRefusal) {
+ label := ""
+ if len(sel.Source) > 0 {
+ label = sel.Source[0].Label
+ }
+ unknown := &offlineRefusal{status: http.StatusConflict, msg: couldNotDetermine(factReplica, label)}
+ m := sel.Fleet.metrics[label]
+ if m == nil {
+ return 0, unknown
+ }
+ f, ok := m.Hosts[host]
+ fresh := func(at int64) bool {
+ age := now.Sub(time.Unix(at, 0))
+ return at > 0 && age <= selfServeFactsFresh && age >= -selfServeFactsSkew
+ }
+ if !ok || f.Incarnation != inc || !fresh(f.CollectedAt) || !fresh(f.ReplicaCollectedAt) || !f.Known || f.Truncated || f.ReplicaBytes < 0 {
+ return 0, unknown
+ }
+ return f.ReplicaBytes, nil
+}
+
// ssProgramsEntry is return mode's programs file with an entry for the host
// (the ordinary rule: OK only when the host has none; pending copy review).
func ssProgramsEntry(host string) string {
@@ -394,11 +426,16 @@ func (s *Server) offlineReturnChecks(ctx context.Context, d offlineDoor, sel off
if ref != nil {
return nil, ref
}
- rep, ref := s.readReturnReport(ctx, d.host, returnOf, o, sel, time.Now())
+ now := time.Now()
+ rep, ref := s.readReturnReport(ctx, d.host, returnOf, o, sel, now)
+ if ref != nil {
+ return nil, ref
+ }
+ incoming, ref := returnIncomingBytes(sel, d.host, d.rec.Incarnation, now)
if ref != nil {
return nil, ref
}
- if why := returnReadyDecision(d.host, sel.Target.Label, o, rep.Ready, s.selfServeTargetWant()); why != "" {
+ if why := returnReadyDecision(d.host, sel.Target.Label, o, rep.Ready, s.selfServeTargetWant(), incoming); why != "" {
return nil, &offlineRefusal{status: http.StatusConflict, msg: why}
}
return o, nil
diff --git a/internal/console/offlinereturn_test.go b/internal/console/offlinereturn_test.go
index 0015f9415..edc078963 100644
--- a/internal/console/offlinereturn_test.go
+++ b/internal/console/offlinereturn_test.go
@@ -9,6 +9,7 @@ import (
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/status"
)
// The operator's return handshake at the console (self-serve moves plan 2
@@ -42,9 +43,27 @@ func newReturnWorld(t *testing.T, edit func(o *moves.OpenRecord)) *returnWorld {
}
body, _ := json.Marshal(w.open)
offlinePut(t, w.fx.rel, moves.OpenRecordKey(offlineHost, returnFwd), body)
+ // The current source (ie-box) reports the instance's replica, known
+ // and fresh, as its status block does.
+ w.sourceFacts(nil)
return w
}
+// sourceFacts writes ie-box's status block with the instance's facts: a
+// known, fresh 9-byte replica of this lifetime, edited.
+func (w *returnWorld) sourceFacts(edit func(f *status.HostFacts)) {
+ w.t.Helper()
+ now := time.Now()
+ f := status.HostFacts{CollectedAt: now.Unix(), Known: true, Incarnation: w.record().Incarnation, ReplicaBytes: 9, ReplicaObjects: 1, ReplicaCollectedAt: now.Unix()}
+ if edit != nil {
+ edit(&f)
+ }
+ sb := status.Box{Label: "ie-box", GeneratedAt: now, Metrics: &status.BoxMetrics{Hosts: map[string]status.HostFacts{offlineHost: f}}}
+ if err := status.WriteBox(context.Background(), w.fx.rel, sb); err != nil {
+ w.t.Fatal(err)
+ }
+}
+
// request asks for the return and answers the request key.
func (w *returnWorld) request() string {
w.t.Helper()
@@ -310,11 +329,11 @@ func TestOperatorReturnHandshakeAtTheConsole(t *testing.T) {
r := w.ready()
r.SecretsKeyFP = "another"
want := targetWant{caps: []string{offlineMoveCapability, offlineStageCapability}}
- if got := returnReadyDecision(offlineHost, "uk-box", &w.open, r, want); got != "" {
+ if got := returnReadyDecision(offlineHost, "uk-box", &w.open, r, want, 9); got != "" {
t.Fatalf("no fingerprint to compare: %q", got)
}
want.secretsFP = testKeyFP
- if got := returnReadyDecision(offlineHost, "uk-box", &w.open, r, want); got != "the target box's secrets key does not match" {
+ if got := returnReadyDecision(offlineHost, "uk-box", &w.open, r, want, 9); got != "the target box's secrets key does not match" {
t.Fatalf("another key: %q", got)
}
})
@@ -361,3 +380,68 @@ func TestReturnNeedsTheForwardCompleted(t *testing.T) {
})
}
}
+
+// TestReturnDiskUsesCurrentSourceSize (final review F3): return mode's disk
+// rule sizes the data the return brings from the current source's judged
+// replica size (fresh, of this lifetime, known), the larger of that and the
+// leftovers listed in the target bucket. Leftovers cleaned away, or an
+// application grown since the forward, cannot let a short disk through; a
+// current size the console cannot judge refuses as could not determine,
+// never as zero. At the stage door and at a direct start.
+func TestReturnDiskUsesCurrentSourceSize(t *testing.T) {
+ low := "the target box has too little free disk for " + offlineHost
+ unknown := "could not determine the replica's size on ie-box"
+ for name, c := range map[string]struct {
+ facts func(f *status.HostFacts)
+ ready func(w *returnWorld, r *moves.TargetReadyReport)
+ want string
+ }{
+ "leftovers cleaned, a large current replica": {
+ facts: func(f *status.HostFacts) { f.ReplicaBytes = 6 << 30 },
+ ready: func(w *returnWorld, r *moves.TargetReadyReport) { r.InstancesPrefix.Objects = nil },
+ want: low},
+ "grown since the forward": {
+ facts: func(f *status.HostFacts) { f.ReplicaBytes = 6 << 30 },
+ ready: func(w *returnWorld, r *moves.TargetReadyReport) {
+ r.InstancesPrefix.Objects["0000/a.ltx"] = moves.PrefixObject{Size: 1 << 30, LastModified: w.open.OpenedAt.Add(-time.Minute)}
+ },
+ want: low},
+ "the leftovers still the larger": {
+ ready: func(w *returnWorld, r *moves.TargetReadyReport) {
+ r.InstancesPrefix.Objects["0000/a.ltx"] = moves.PrefixObject{Size: 6 << 30, LastModified: w.open.OpenedAt.Add(-time.Minute)}
+ },
+ want: low},
+ "the current size not known": {facts: func(f *status.HostFacts) { f.Known = false }, want: unknown},
+ "the current size cut short": {facts: func(f *status.HostFacts) { f.Truncated = true }, want: unknown},
+ "the current size of another lifetime": {facts: func(f *status.HostFacts) { f.Incarnation++ }, want: unknown},
+ "the current size stale": {facts: func(f *status.HostFacts) {
+ old := time.Now().Add(-time.Hour).Unix()
+ f.CollectedAt, f.ReplicaCollectedAt = old, old
+ }, want: unknown},
+ "a small current replica": {facts: func(f *status.HostFacts) { f.ReplicaBytes = 1 << 30 }},
+ } {
+ for _, door := range []string{"/stage", ""} {
+ t.Run(name+" at "+map[string]string{"/stage": "stage", "": "start"}[door], func(t *testing.T) {
+ w := newReturnWorld(t, nil)
+ key := w.request()
+ r := w.ready()
+ if c.ready != nil {
+ c.ready(w, r)
+ }
+ w.report("uk-box", "uk-box", "", key, time.Now(), r)
+ w.sourceFacts(c.facts)
+ body := `{"region":"uk","return_of":"` + returnFwd + `"}`
+ if c.want == "" {
+ if code, out := w.req(http.MethodPost, door, body); code != http.StatusAccepted {
+ t.Fatalf("%s = %d %s", door, code, out)
+ }
+ return
+ }
+ w.refused(http.MethodPost, door, body, http.StatusConflict, c.want)
+ if rec := w.record(); rec.Move != nil {
+ t.Fatalf("a refused return wrote %+v", rec.Move)
+ }
+ })
+ }
+ }
+}