v221 ⑥ probe traffic: binary GB convention + one-shot calibration
Unit unification: GB entries converted at 1e9 while every byte formatter is binary (1024^3), so a 3000 GB limit rendered 2.73 TB instead of 2.93 TB. calBytes and the form's limit conversion now use 1024^3. Stored traffic_limits (bytes, typed x1e9) are rescaled x1.073741824 exactly once at store open, guarded by the probe_setting mark gb_1024_migrated_v1; calibration values are stored in GB (no rescale — only their conversion changed) and price/sort/reset-day are unitless. One-shot calibration: the form's cal fields are deltas to apply now. probe.MergeCalDelta folds a non-zero submitted value into the stored standing correction and keeps the standing value on a zero/absent submission; putMeta merges against the stored row (SaveMeta/UpsertMeta keep plain replace — the notify sweep relies on it). GET no longer echoes the standing correction into the form, and the frontend keeps the cal inputs as local state that starts empty and clears on successful save, so a later unrelated save resubmits nothing. Tests: calBytes(3000)=3000x2^30; migration 3.0e12 -> 3000x2^30 with no second rescale on reopen; one-shot flow (save delta, save unrelated with zero cal -> calibrated traffic unmoved, second delta accumulates once). Existing TestTrafficCalibrationAppliedAndFloored expectation updated to the binary figure. No vitest covers ProbeMetaSection (disclosed in the report).
This commit is contained in:
Muse
@@ -34,6 +34,9 @@ type Meta struct {
|
||||
// corrections in GB (V2.0.1), either sign: added to the counted
|
||||
// uplink (tx) / downlink (rx) figures wherever the ledger is
|
||||
// read — panel statistics, the public page and traffic alerts.
|
||||
// The stored values are the STANDING accumulated correction;
|
||||
// the edit form submits one-shot deltas that MergeCalDelta folds
|
||||
// into them (V2.2.1 ⑥b).
|
||||
TrafficCalUpGB float64
|
||||
TrafficCalDownGB float64
|
||||
// StatIface pins the one interface whose counters the agent
|
||||
@@ -138,6 +141,30 @@ func (s *Store) UpsertMeta(m Meta) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// MergeCalDelta folds a probe-meta form submission into the stored
|
||||
// row under the one-shot calibration semantics (V2.2.1 ⑥b): the
|
||||
// form's calibration fields are a correction TO APPLY NOW, not the
|
||||
// new standing value. A non-zero submitted value adds onto the stored
|
||||
// standing correction; a zero field (an absent field is the same
|
||||
// zero — the wire type cannot distinguish them, and both mean
|
||||
// "nothing to apply") keeps the stored value untouched. Every other
|
||||
// field takes the submitted value, as UpsertMeta's full replace
|
||||
// does. Paired with the form clearing its calibration inputs after a
|
||||
// successful save, a later unrelated save can never re-apply an old
|
||||
// correction.
|
||||
func MergeCalDelta(existing, submitted Meta) Meta {
|
||||
out := submitted
|
||||
out.TrafficCalUpGB = existing.TrafficCalUpGB
|
||||
if submitted.TrafficCalUpGB != 0 {
|
||||
out.TrafficCalUpGB += submitted.TrafficCalUpGB
|
||||
}
|
||||
out.TrafficCalDownGB = existing.TrafficCalDownGB
|
||||
if submitted.TrafficCalDownGB != 0 {
|
||||
out.TrafficCalDownGB += submitted.TrafficCalDownGB
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func boolToInt(b bool) int64 {
|
||||
if b {
|
||||
return 1
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
package probe
|
||||
|
||||
import (
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// V2.2.1 ⑥: GB entries convert on the binary convention (1024^3)
|
||||
// every formatter uses, stored decimal-era limits are rescaled once,
|
||||
// and calibration submissions are one-shot deltas folded into the
|
||||
// standing correction — never re-applied by a later unrelated save.
|
||||
|
||||
func TestCalBytesBinaryConvention(t *testing.T) {
|
||||
if got := calBytes(3000); got != 3000*(1<<30) {
|
||||
t.Fatalf("calBytes(3000) = %d, want %d (3000 GiB)", got, 3000*(1<<30))
|
||||
}
|
||||
if got := calBytes(0); got != 0 {
|
||||
t.Fatalf("calBytes(0) = %d, want 0", got)
|
||||
}
|
||||
if got := calBytes(-1.5); got != -1610612736 {
|
||||
t.Fatalf("calBytes(-1.5) = %d, want -1610612736", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGBConventionMigrationRescalesLimitsOnce(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "probe.db")
|
||||
st, err := Open(path)
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
// Seed a legacy row as a pre-V2.2.1 database holds it — 3000 GB
|
||||
// typed at the decimal convention = 3.0e12 stored bytes — and
|
||||
// strike the migration mark so the next open sees an unmigrated
|
||||
// database.
|
||||
if _, err := st.db.Exec(`INSERT INTO probe_meta (agent_id, traffic_limit) VALUES (9, 3000000000000)`); err != nil {
|
||||
t.Fatalf("seed legacy limit: %v", err)
|
||||
}
|
||||
if _, err := st.db.Exec(`DELETE FROM probe_setting WHERE key = ?`, gbConventionKey); err != nil {
|
||||
t.Fatalf("strike mark: %v", err)
|
||||
}
|
||||
if err := st.Close(); err != nil {
|
||||
t.Fatalf("close: %v", err)
|
||||
}
|
||||
|
||||
st2, err := Open(path)
|
||||
if err != nil {
|
||||
t.Fatalf("reopen (migration run): %v", err)
|
||||
}
|
||||
m, found, err := st2.Meta(9)
|
||||
if err != nil || !found {
|
||||
t.Fatalf("meta after migration: found=%v err=%v", found, err)
|
||||
}
|
||||
// 3.0e12 x 1024^3/1e9 = 3000 x 2^30 exactly (~3.221e12).
|
||||
if want := int64(3000) * (1 << 30); m.TrafficLimit != want {
|
||||
t.Fatalf("migrated limit = %d, want %d", m.TrafficLimit, want)
|
||||
}
|
||||
if err := st2.Close(); err != nil {
|
||||
t.Fatalf("close 2: %v", err)
|
||||
}
|
||||
|
||||
st3, err := Open(path)
|
||||
if err != nil {
|
||||
t.Fatalf("reopen 2: %v", err)
|
||||
}
|
||||
defer st3.Close()
|
||||
m, found, err = st3.Meta(9)
|
||||
if err != nil || !found {
|
||||
t.Fatalf("meta after second open: found=%v err=%v", found, err)
|
||||
}
|
||||
if want := int64(3000) * (1 << 30); m.TrafficLimit != want {
|
||||
t.Fatalf("limit after second open = %d, want %d (no double rescale)", m.TrafficLimit, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOneShotCalibrationSaveFlow(t *testing.T) {
|
||||
st := newTestStore(t)
|
||||
now := time.Now()
|
||||
book(t, st, 5, "boot-a", 1000, 1000, now, nil, 0) // baseline: books 0
|
||||
book(t, st, 5, "boot-a", 3000, 2000, now, nil, 0) // books rx 2000, tx 1000
|
||||
|
||||
// First save carries a +1 GB uplink correction delta.
|
||||
submitted := DefaultMeta(5)
|
||||
submitted.TrafficCalUpGB = 1
|
||||
merged := MergeCalDelta(DefaultMeta(5), submitted)
|
||||
if merged.TrafficCalUpGB != 1 {
|
||||
t.Fatalf("standing after first fold = %v, want 1", merged.TrafficCalUpGB)
|
||||
}
|
||||
if err := st.UpsertMeta(merged); err != nil {
|
||||
t.Fatalf("upsert 1: %v", err)
|
||||
}
|
||||
after1, ok, err := st.TrafficFor(5)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("traffic after save 1: ok=%v err=%v", ok, err)
|
||||
}
|
||||
if want := int64(1000) + (1 << 30); after1.TotalTx != want {
|
||||
t.Fatalf("TotalTx after save 1 = %d, want %d", after1.TotalTx, want)
|
||||
}
|
||||
if after1.TotalRx != 2000 {
|
||||
t.Fatalf("TotalRx after save 1 = %d, want 2000 (uncalibrated)", after1.TotalRx)
|
||||
}
|
||||
|
||||
// Second save changes another field; the (cleared) form submits
|
||||
// zero calibration — the calibrated figure must not move again.
|
||||
stored, found, err := st.Meta(5)
|
||||
if err != nil || !found {
|
||||
t.Fatalf("stored meta: found=%v err=%v", found, err)
|
||||
}
|
||||
sub2 := DefaultMeta(5)
|
||||
sub2.GroupName = "g2"
|
||||
merged2 := MergeCalDelta(stored, sub2)
|
||||
if merged2.TrafficCalUpGB != 1 {
|
||||
t.Fatalf("standing after zero-delta save = %v, want 1 (unchanged)", merged2.TrafficCalUpGB)
|
||||
}
|
||||
if err := st.UpsertMeta(merged2); err != nil {
|
||||
t.Fatalf("upsert 2: %v", err)
|
||||
}
|
||||
after2, ok, err := st.TrafficFor(5)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("traffic after save 2: ok=%v err=%v", ok, err)
|
||||
}
|
||||
if after2.TotalTx != after1.TotalTx {
|
||||
t.Fatalf("TotalTx moved on an unrelated save: %d -> %d (double-applied calibration)", after1.TotalTx, after2.TotalTx)
|
||||
}
|
||||
|
||||
// A further non-zero delta accumulates exactly once more.
|
||||
sub3 := DefaultMeta(5)
|
||||
sub3.TrafficCalUpGB = 0.5
|
||||
merged3 := MergeCalDelta(merged2, sub3)
|
||||
if merged3.TrafficCalUpGB != 1.5 {
|
||||
t.Fatalf("standing after second delta = %v, want 1.5", merged3.TrafficCalUpGB)
|
||||
}
|
||||
if err := st.UpsertMeta(merged3); err != nil {
|
||||
t.Fatalf("upsert 3: %v", err)
|
||||
}
|
||||
after3, _, err := st.TrafficFor(5)
|
||||
if err != nil {
|
||||
t.Fatalf("traffic after save 3: %v", err)
|
||||
}
|
||||
if want := int64(1000) + calBytes(1.5); after3.TotalTx != want {
|
||||
t.Fatalf("TotalTx after save 3 = %d, want %d", after3.TotalTx, want)
|
||||
}
|
||||
}
|
||||
@@ -91,8 +91,10 @@ func TestTrafficCalibrationAppliedAndFloored(t *testing.T) {
|
||||
t.Fatalf("AllTraffic: %v", err)
|
||||
}
|
||||
tr := all[5]
|
||||
if tr.TotalRx != 3_500_000_000 || tr.MonthRx != 3_500_000_000 || tr.DayRx != 3_500_000_000 {
|
||||
t.Errorf("rx figures after +0.5 GB calibration = %d/%d/%d, want 3500000000 each", tr.TotalRx, tr.MonthRx, tr.DayRx)
|
||||
// V2.2.1 ⑥: calibration converts on the binary convention —
|
||||
// +0.5 GB = 0.5 x 2^30 = 536870912 B (was 5e8 at the old 1e9).
|
||||
if tr.TotalRx != 3_536_870_912 || tr.MonthRx != 3_536_870_912 || tr.DayRx != 3_536_870_912 {
|
||||
t.Errorf("rx figures after +0.5 GB calibration = %d/%d/%d, want 3536870912 each", tr.TotalRx, tr.MonthRx, tr.DayRx)
|
||||
}
|
||||
if tr.TotalTx != 0 || tr.MonthTx != 0 || tr.DayTx != 0 {
|
||||
t.Errorf("tx figures after -2 GB calibration = %d/%d/%d, want 0 (floored)", tr.TotalTx, tr.MonthTx, tr.DayTx)
|
||||
|
||||
@@ -213,9 +213,52 @@ func Open(path string) (*Store, error) {
|
||||
db.Close()
|
||||
return nil, fmt.Errorf("probe schema migration: %w", err)
|
||||
}
|
||||
if err := migrateGBConvention(db); err != nil {
|
||||
db.Close()
|
||||
return nil, fmt.Errorf("probe gb migration: %w", err)
|
||||
}
|
||||
return &Store{db: db, nodes: make(map[int64]*nodeState)}, nil
|
||||
}
|
||||
|
||||
// gbConventionKey is the probe_setting mark recording that the
|
||||
// V2.2.1 ⑥ traffic-limit rescale has been applied to this database.
|
||||
const gbConventionKey = "gb_1024_migrated_v1"
|
||||
|
||||
// migrateGBConvention rescales stored probe_meta traffic limits from
|
||||
// the old decimal entry convention to the unified binary one, once.
|
||||
// Limits are typed in GB and were stored as GB × 1e9 bytes, while
|
||||
// every byte formatter (panel, public page) divides by 1024^3 — a
|
||||
// 3000 GB limit displayed as 2.73 TB instead of 2.93 TB. The GB the
|
||||
// operator typed is the intent, so stored bytes are scaled by
|
||||
// 1024^3/1e9 = 1.073741824 and now read back correctly under the
|
||||
// binary conversion. Only traffic_limit is byte-denominated:
|
||||
// calibration values are stored in GB (their byte conversion,
|
||||
// calBytes, carries the fix) and price/sort/reset-day are unitless —
|
||||
// neither is touched here. The probe_setting mark makes the rescale
|
||||
// run exactly once per database.
|
||||
func migrateGBConvention(db *sql.DB) error {
|
||||
var v string
|
||||
err := db.QueryRow(`SELECT value FROM probe_setting WHERE key = ?`, gbConventionKey).Scan(&v)
|
||||
if err == nil {
|
||||
return nil // already applied
|
||||
}
|
||||
if err != sql.ErrNoRows {
|
||||
return err
|
||||
}
|
||||
tx, err := db.Begin()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
if _, err := tx.Exec(`UPDATE probe_meta SET traffic_limit = CAST(ROUND(traffic_limit * 1.073741824) AS INTEGER) WHERE traffic_limit > 0`); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := tx.Exec(`INSERT OR REPLACE INTO probe_setting (key, value) VALUES (?, '1')`, gbConventionKey); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
// metaColumnMigrations are the probe_meta columns added after the
|
||||
// table's first release; CREATE TABLE IF NOT EXISTS never retrofits
|
||||
// them onto an existing database, so they are added here, once, by
|
||||
|
||||
@@ -332,9 +332,13 @@ func (s *Store) TrafficFor(agentID int64) (Traffic, bool, error) {
|
||||
// AllTraffic returns every agent's ledger in one query, gated as in
|
||||
// TrafficFor — the bulk read port for B3's node list.
|
||||
// calBytes converts a GB calibration value to bytes on the panel's
|
||||
// 1 GB = 1e9 B convention (the same one the traffic limit uses).
|
||||
// binary convention, 1 GB = 1024^3 B — the convention every byte
|
||||
// formatter in the panel and on the public page uses, and the one
|
||||
// the traffic limit's GB entry converts with (V2.2.1 ⑥; the decimal
|
||||
// 1e9 stood here before, which made a 3000 GB limit read as 2.73 TB
|
||||
// instead of 2.93 TB wherever bytes were formatted).
|
||||
func calBytes(gb float64) int64 {
|
||||
return int64(math.Round(gb * 1e9))
|
||||
return int64(math.Round(gb * (1 << 30)))
|
||||
}
|
||||
|
||||
// applyCalibration shifts a ledger row by the operator's per-node
|
||||
|
||||
Reference in New Issue
Block a user