amadan

carlos / platform Public

console: a rated timing that finished inside its fixed part, or within a tick of it, is left out of the rate's median (the default stands until five informative rows exist; the 100 ms clamp is gone); a land or promote row needs its landed bytes; the window's floor is a bound parameter

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 4a41b86f8533764673afe0e1dd12c6d7648f996e parent 4997a30
2 files changed, +120 −35
  • internal/console/selfserve_estimate.go +49 −28
  • internal/console/selfserve_estimate_test.go +71 −7
diff --git a/internal/console/selfserve_estimate.go b/internal/console/selfserve_estimate.go
index d05799c8e..5403cabc9 100644
--- a/internal/console/selfserve_estimate.go
+++ b/internal/console/selfserve_estimate.go
@@ -19,11 +19,11 @@ import (
// Copy)).
//
// Each term starts at its default and, once the region pair has five
-// finished moves' timings for it, is refined from the median of the last
+// finished moves' informative timings for it, is refined from the median of the last
// twenty (Task 9): every term's fixed part from the rows under 1 MiB (the
// median of their seconds), and a rate term's (copy, land, promote) rate
-// from the rows of at least 1 MiB, measured past that fixed part
-// (estimateRules). The timings are the marks the console observed (console
+// from the rows of at least 1 MiB that took at least a tick past that
+// fixed part (estimateRules). The timings are the marks the console observed (console
// observation times, insert-only, never overwritten) and the five rows a
// finished move derives from them; a move that aborted, or lacks a mark,
// contributes no row for that phase.
@@ -72,10 +72,12 @@ const (
// rate; a row under it measures its term's fixed part only.
const estimateRateFloor = mib
-// estimateRateEpsilon bounds a rate sample's time past the fixed part
-// from below: a row whose seconds are all overhead (a small final copy)
-// says its bytes moved in at most this long, never in no time.
-const estimateRateEpsilon = 100 * time.Millisecond
+// estimateResolution is how precisely a timing is known: the marks are
+// the pass's observations, one tick (selfServeTickInterval's 5 s) apart.
+// A rated row whose time past its term's fixed part is under it carries
+// no rate (its bytes might have moved in no time or in a whole tick), so
+// it is left out of the rate's median.
+const estimateResolution = 5 * time.Second
// estimateRules are the terms' rules for the region pair from -> to: the
// defaults, each refined from the pair's timings when it has enough. For
@@ -83,9 +85,14 @@ const estimateRateEpsilon = 100 * time.Millisecond
// median of their seconds once there are estimateMinRows of the last
// estimateWindow, else the default. For a rate term (copy, land, promote)
// the rows at or over the floor measure its rate past that fixed part:
-// the median of bytes / max(seconds - fixed, estimateRateEpsilon), once
-// there are enough, else the default. So a phase's overhead is counted
-// once, in the fixed part, and never again in the rate. A timing that
+// the median of bytes / (seconds - fixed) over the rows whose time past
+// the fixed part is at least estimateResolution, once estimateMinRows of
+// them exist among the last estimateWindow, else the default. A row that
+// finished inside its fixed part, or within one tick of it, says nothing
+// about the lane's speed and is left out, so a lane of quick moves keeps
+// the default rate rather than one set by the rows' sizes. So a phase's
+// overhead is counted once, in the fixed part, and never again in the
+// rate. A timing that
// cannot be read keeps the default (logged): the estimate is never refused
// for it.
func (s *Server) estimateRules(from, to string) struct{ Stop, Copy, Land, Open, Promote term } {
@@ -115,15 +122,13 @@ func (s *Server) estimateRules(from, to string) struct{ Stop, Copy, Land, Open,
if t.BytesPerSecond == 0 {
return // a fixed term
}
- if rows := window(phase, true); rows != nil {
- xs := make([]float64, len(rows))
- for i, m := range rows {
- past := m.Seconds - t.Fixed.Seconds()
- if past < estimateRateEpsilon.Seconds() {
- past = estimateRateEpsilon.Seconds()
- }
- xs[i] = float64(m.Bytes) / past
+ var xs []float64
+ for _, m := range window(phase, true) {
+ if past := m.Seconds - t.Fixed.Seconds(); past >= estimateResolution.Seconds() {
+ xs = append(xs, float64(m.Bytes)/past)
}
+ }
+ if len(xs) >= estimateMinRows {
t.BytesPerSecond = median(xs)
}
}
@@ -279,15 +284,18 @@ func (s *Store) MoveTimings(from, to, phase string, limit int) ([]MoveTiming, er
// most limit.
func (s *Store) moveTimingWindow(from, to, phase string, rated bool, limit int) ([]MoveTiming, error) {
if rated {
- return s.moveTimings(fmt.Sprintf(`AND bytes >= %d AND seconds > 0`, int64(estimateRateFloor)), from, to, phase, limit)
+ return s.moveTimings(`AND bytes >= ? AND seconds > 0`, from, to, phase, limit, int64(estimateRateFloor))
}
- return s.moveTimings(fmt.Sprintf(`AND bytes < %d AND seconds >= 0`, int64(estimateRateFloor)), from, to, phase, limit)
+ return s.moveTimings(`AND bytes < ? AND seconds >= 0`, from, to, phase, limit, int64(estimateRateFloor))
}
-func (s *Store) moveTimings(cond, from, to, phase string, limit int) ([]MoveTiming, error) {
+// moveTimings runs the window's query: cond is one of the constant
+// conditions above, its values in args, bound.
+func (s *Store) moveTimings(cond, from, to, phase string, limit int, args ...any) ([]MoveTiming, error) {
+ q := append(append([]any{from, to, phase}, args...), limit)
rows, err := s.db.Query(`SELECT id, phase, from_region, to_region, bytes, seconds, at FROM move_timings
WHERE from_region = ? AND to_region = ? AND phase = ? `+cond+`
-ORDER BY at DESC, id DESC LIMIT ?`, from, to, phase, limit)
+ORDER BY at DESC, id DESC LIMIT ?`, q...)
if err != nil {
return nil, err
}
@@ -308,12 +316,22 @@ ORDER BY at DESC, id DESC LIMIT ?`, from, to, phase, limit)
// selfServeTimingPhases are the five rows a finished move derives: each
// phase from one mark to another, with the bytes of a third (none for a
// fixed term).
-var selfServeTimingPhases = []struct{ phase, from, to, bytes string }{
- {"stop", "offline_at", "stopped_at", ""},
- {"copy", "seal_at", "sealed_at", "sealed_at"},
- {"land", "flip_at", "landed_at", "landed_at"},
- {"open", "open_at", "opened_at", ""},
- {"promote", "open_at", "first_ok_at", "landed_replica"},
+//
+// needBytes: the bytes mark must carry bytes, or the phase gives no row.
+// A landed mark without bytes is one whose report read failed (the
+// landing and the replica always carry some); its seconds are a whole
+// transfer and would otherwise fall under the rate floor and pass for the
+// land or promote term's fixed part. A final copy that carried nothing is
+// a true fixed-part sample, so copy needs none.
+var selfServeTimingPhases = []struct {
+ phase, from, to, bytes string
+ needBytes bool
+}{
+ {"stop", "offline_at", "stopped_at", "", false},
+ {"copy", "seal_at", "sealed_at", "sealed_at", false},
+ {"land", "flip_at", "landed_at", "landed_at", true},
+ {"open", "open_at", "opened_at", "", false},
+ {"promote", "open_at", "first_ok_at", "landed_replica", true},
}
// selfServeMoveTimings are the rows row's marks give: none unless the move
@@ -334,6 +352,9 @@ func selfServeMoveTimings(row *SelfServeMove, marks map[string]timingMark) []Mov
if p.bytes != "" {
m.Bytes = marks[p.bytes].Bytes
}
+ if p.needBytes && m.Bytes <= 0 {
+ continue
+ }
out = append(out, m)
}
return out
diff --git a/internal/console/selfserve_estimate_test.go b/internal/console/selfserve_estimate_test.go
index 134d48448..787fbc607 100644
--- a/internal/console/selfserve_estimate_test.go
+++ b/internal/console/selfserve_estimate_test.go
@@ -164,8 +164,8 @@ func TestEstimateBeforeAndAfterStaging(t *testing.T) {
s := &Server{store: st}
const replica, files = 3 << 30, 4 * mib
for i := 0; i < 5; i++ {
- // 100 MiB past the default 10 s fixed part in 2 s: 50 MiB/s.
- ssTiming(t, st, fmt.Sprintf("mv_lane%02d", i), "copy", 100*mib, 12, ssT0.Add(time.Duration(i)*time.Minute))
+ // 500 MiB past the default 10 s fixed part in 10 s: 50 MiB/s.
+ ssTiming(t, st, fmt.Sprintf("mv_lane%02d", i), "copy", 500*mib, 20, ssT0.Add(time.Duration(i)*time.Minute))
}
// Before a move is requested the final copy is the whole replica (the
// door's estimate).
@@ -204,9 +204,11 @@ func TestEstimateBeforeAndAfterStaging(t *testing.T) {
}
// TestEstimateRatesExcludeTheFixedPart: a rate is measured past the
-// term's fixed part, bytes / max(seconds - fixed, 100 ms), over rows of at
-// least 1 MiB; so small final copies, whose seconds are almost all
-// overhead, never predict a large replica as hours.
+// term's fixed part, bytes / (seconds - fixed), over rows of at least 1 MiB
+// that took at least a tick (5 s) past it; a row that finished inside its
+// fixed part carries no rate and is left out. So small final copies never
+// predict a large replica as hours, and quick large landings never shrink
+// a large instance's landing deadline to its floor.
func TestEstimateRatesExcludeTheFixedPart(t *testing.T) {
const replica = 3 << 30
pure := 10*time.Second + ssRate(replica, 25*mib) // the default copy term: about two minutes
@@ -221,6 +223,42 @@ func TestEstimateRatesExcludeTheFixedPart(t *testing.T) {
if got.Copy > 2*pure || got.Copy < pure/2 {
t.Fatalf("a 3 GiB copy estimates %v from five small final copies, want within 2x of %v", got.Copy, pure)
}
+ // Their seconds are all overhead: they carry no rate, so the
+ // default rate stands.
+ if r := s.estimateRules("ie", "uk"); r.Copy.BytesPerSecond != estimateDefaults.Copy.BytesPerSecond {
+ t.Fatalf("copy rate %v from rows inside the fixed part, want the default %v", r.Copy.BytesPerSecond, estimateDefaults.Copy.BytesPerSecond)
+ }
+ })
+
+ t.Run("large rows inside the fixed part neither raise the rate nor shorten the land deadline", func(t *testing.T) {
+ st := ssStore(t)
+ s := &Server{store: st}
+ // Five landings of 50 MiB to 2 GiB that finished in 10 to 33 s:
+ // within the default 30 s fixed part, or less than one tick (5 s)
+ // past it, so none says how fast the lane is.
+ for i, row := range []struct {
+ bytes int64
+ sec float64
+ }{{50 * mib, 10}, {200 * mib, 18}, {500 * mib, 25}, {1 << 30, 29}, {2 << 30, 33}} {
+ ssTiming(t, st, fmt.Sprintf("mv_fastland%02d", i), "land", row.bytes, row.sec, ssT0.Add(time.Duration(i)*time.Minute))
+ }
+ if r := s.estimateRules("ie", "uk"); r.Land.BytesPerSecond != estimateDefaults.Land.BytesPerSecond || r.Land.Fixed != estimateDefaults.Land.Fixed {
+ t.Errorf("land rule %+v, want the default %+v", r.Land, estimateDefaults.Land)
+ }
+ // A 30 GiB instance: the landing's deadline is three times the
+ // default's estimate, about 40 minutes, not the 10-minute floor.
+ const files, big = 2 << 30, 28 << 30
+ terms, _ := s.estimate("ie", "uk", 0, files, big)
+ raw, err := json.Marshal(terms)
+ if err != nil {
+ t.Fatal(err)
+ }
+ row := &SelfServeMove{EstimateTerms: string(raw)}
+ now := ssT0
+ want := now.Add(3 * (30*time.Second + ssRate(files+big, 40*mib)))
+ if got := selfServePhaseDeadline(row, ssLanding, now); got.Sub(want).Abs() > time.Second {
+ t.Fatalf("the landing's deadline is %v after now, want %v", got.Sub(now), want.Sub(now))
+ }
})
t.Run("a measured fixed part and rate are recovered", func(t *testing.T) {
@@ -229,7 +267,7 @@ func TestEstimateRatesExcludeTheFixedPart(t *testing.T) {
for i := 0; i < 5; i++ {
at := ssT0.Add(time.Duration(i) * time.Minute)
ssTiming(t, st, fmt.Sprintf("mv_fixed%02d", i), "copy", mib/4, 6, at) // the overhead alone
- ssTiming(t, st, fmt.Sprintf("mv_rated%02d", i), "copy", 200*mib, 6+200.0/40, at) // 6 s + 200 MiB at 40 MiB/s
+ ssTiming(t, st, fmt.Sprintf("mv_rated%02d", i), "copy", 400*mib, 6+400.0/40, at) // 6 s + 400 MiB at 40 MiB/s
}
r := s.estimateRules("ie", "uk")
if r.Copy.Fixed != 6*time.Second || r.Copy.BytesPerSecond != 40*mib {
@@ -370,9 +408,35 @@ func TestAbortedMovesContributeNoRates(t *testing.T) {
}
}
+ // A landed mark without bytes (its report read failed) derives no land
+ // or promote row: its seconds are a full transfer, not a fixed part.
+ // A final copy that carried nothing still derives its copy row.
+ ssAdmit(t, st, ssMove("mv_nobytesnobytesnobytesnobyt", "d."+ssHost, ssT0, "ie-1", "uk-1"))
+ for i, m := range []string{"offline_at", "stopped_at", "seal_at", "sealed_at", "flip_at", "landed_at", "landed_replica", "open_at", "opened_at", "first_ok_at"} {
+ if err := st.MarkSelfServeTiming("mv_nobytesnobytesnobytesnobyt", m, ssT0.Add(time.Duration(i*i)*time.Second), 0); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := st.FinishSelfServeMove("mv_nobytesnobytesnobytesnobyt", provision.OfflineOutcomeDone, "", "", ssT0.Add(time.Hour)); err != nil {
+ t.Fatal(err)
+ }
+ if err := s.selfServeTimingsOnFinish(ctx, ssLoad(t, st, "mv_nobytesnobytesnobytesnobyt")); err != nil {
+ t.Fatal(err)
+ }
+ for phase, want := range map[string]bool{"stop": true, "copy": true, "land": false, "open": true, "promote": false} {
+ got, _ := st.MoveTimings("ie", "uk", phase, 20)
+ has := false
+ for _, m := range got {
+ has = has || m.ID == "mv_nobytesnobytesnobytesnobyt"
+ }
+ if has != want {
+ t.Errorf("%s: a row from a move without landed bytes = %v, want %v", phase, has, want)
+ }
+ }
+
// A done move without a mark contributes no row for its phase.
finish("mv_nofirstoknofirstoknofirsto", "c."+ssHost, provision.OfflineOutcomeDone, "first_ok_at", "seal_at")
- for phase, n := range map[string]int{"promote": 1, "copy": 1, "stop": 2, "land": 2, "open": 2} {
+ for phase, n := range map[string]int{"promote": 1, "copy": 2, "stop": 3, "land": 2, "open": 3} {
if got, _ := st.MoveTimings("ie", "uk", phase, 20); len(got) != n {
t.Errorf("%s: %d rows, want %d", phase, len(got), n)
}