From 3b38e9008df220680f76bb051ab677543a0f2321 Mon Sep 17 00:00:00 2001 From: Moses Narrow <36607567+0pcom@users.noreply.github.com> Date: Wed, 26 Aug 2026 14:36:01 -0500 Subject: [PATCH] feat(proxy): split status.skysocks goodput up/down, per-direction share bars, fix direct-leg rtt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit status.skysocks route tree + 'proxy tree': - Goodput split into per-direction rates (goodput_up_bps / goodput_down_bps, sampled separately in snapshotLegs; goodput_bps kept as the sum) shown next to the up/down byte totals instead of one combined column. Also fixes the CLI 'proxy tree' path, which dropped goodput entirely. - Per-direction bandwidth-SHARE bars (████░, fixed width) restored — this route's fraction of the route-group aggregate up-/down-goodput. The original share bars lived in the pre-#4137 mux table; re-added in the shared RouteTree so CLI and HTML stay unified. - Direct-leg latency: hop-0 now shows the live E2E route rtt for a direct leg instead of tp.GetLatency()'s RAW last 60s-cadence sample (SetLatency overwrites Avg with no EWMA), which made the transport rtt disagree with the smoothed route rtt for up to a minute. tp.GetLatency() is left untouched for route-finder / tp ls / policy scoring; multihop legs keep the near-edge transport RTT on hop 0. --- cmd/skywire-cli/commands/proxy/tree.go | 4 + pkg/proxystatus/provider.go | 12 ++- pkg/proxystatus/routetree.go | 94 ++++++++++++++++------ pkg/router/route_group.go | 15 +++- pkg/router/route_mux.go | 106 +++++++++++++++---------- pkg/visor/api_routing.go | 4 + pkg/visor/embedded_proxystatus.go | 2 + pkg/visor/rpc.go | 18 +++-- 8 files changed, 182 insertions(+), 73 deletions(-) diff --git a/cmd/skywire-cli/commands/proxy/tree.go b/cmd/skywire-cli/commands/proxy/tree.go index 0ff444c4cf..b2b9e473c4 100644 --- a/cmd/skywire-cli/commands/proxy/tree.go +++ b/cmd/skywire-cli/commands/proxy/tree.go @@ -158,6 +158,8 @@ type treeLegInfo struct { Direct bool `json:"direct"` SentBytes uint64 `json:"sent_bytes"` RecvBytes uint64 `json:"recv_bytes"` + GoodputUpBps float64 `json:"goodput_up_bps,omitempty"` + GoodputDownBps float64 `json:"goodput_down_bps,omitempty"` Retransmits uint64 `json:"retransmits"` Alive bool `json:"alive"` Standby bool `json:"standby"` @@ -192,6 +194,8 @@ func (rg treeRouteGroup) toSnapshot() proxystatus.Snapshot { Direct: l.Direct, SentBytes: l.SentBytes, RecvBytes: l.RecvBytes, + GoodputUpBps: l.GoodputUpBps, + GoodputDownBps: l.GoodputDownBps, Retransmits: l.Retransmits, Alive: l.Alive, Standby: l.Standby, diff --git a/pkg/proxystatus/provider.go b/pkg/proxystatus/provider.go index 02dda2b770..b48e02a71d 100644 --- a/pkg/proxystatus/provider.go +++ b/pkg/proxystatus/provider.go @@ -67,10 +67,14 @@ type Leg struct { // GoodputBps is this leg's recent goodput — the EWMA of (sent+recv) bytes // per second over the page's refresh window (~1s), i.e. the RATE the leg is // moving now, distinct from the cumulative SentBytes/RecvBytes counters. 0 - // until a second sample lands. - GoodputBps float64 - Alive bool - Standby bool + // until a second sample lands. GoodputUpBps/GoodputDownBps split it by + // direction (send-rate / recv-rate), shown beside the ↑/↓ byte totals and + // driving the per-route share bars; GoodputBps is their sum. + GoodputBps float64 + GoodputUpBps float64 + GoodputDownBps float64 + Alive bool + Standby bool // Hops is the leg's full forward route (every hop to the destination), // full PKs, per-hop transport type + latency where known. Hops []Hop diff --git a/pkg/proxystatus/routetree.go b/pkg/proxystatus/routetree.go index ceb8cbf07d..ad0a7dfe23 100644 --- a/pkg/proxystatus/routetree.go +++ b/pkg/proxystatus/routetree.go @@ -10,21 +10,25 @@ // summary branches it reads as a normal downward tree with the source PK at top // in the same PK column as the hops: // -// -// │ -// R[0] ● 143ms 1.6K↑ 1.5K↓ ┬── [stcpr] 143ms -// R[1] ● 47ms 0.5K↑ 0.4K↓ ┴── [sudph] 47ms -// └── [stcpr] 88ms +// +// │ +// R[0] ● 143ms 1.6K↑ 8.1K/s ███░ 1.5K↓ 12K/s ██░░ ┬── [stcpr] 143ms +// R[1] ● 47ms 0.5K↑ 2.0K/s █░░░ 0.4K↓ 3K/s █░░░ ┴── [sudph] 47ms +// └── [stcpr] 88ms // // - Root = this visor (the source PK, taken from any leg's first hop). // - Each ACTIVE route is a top-level right child; DEAD legs are pruned. // - The LEFT block of a route is its per-route summary: R[n], a state glyph // (● active / ○ standby — the caller colors it, no state words), the -// end-to-end ROUTE rtt, and bandwidth X↑ Y↓. +// end-to-end ROUTE rtt, then per direction the cumulative bytes (X↑ / Y↓), +// the live goodput RATE (send-rate / recv-rate) and a SHARE bar — this +// route's fraction of the group's aggregate up / down goodput. // - The RIGHT branch of a route is its hop chain; each hop node's label is the // peer PK (never truncated) and its trailing columns are [], -// , — the per-hop TRANSPORT rtt (distinct from the -// route rtt on the left). +// , — the per-hop TRANSPORT rtt. For a DIRECT (1-hop) +// leg this equals the left ROUTE rtt (same physical link, same live +// measurement); on a multihop leg the near-edge transport rtt and the +// whole-path route rtt differ. package proxystatus import ( @@ -74,9 +78,20 @@ func RouteTree(snap Snapshot) *bitree.Node { return legs[i].Index < legs[j].Index }) + // Aggregate per-direction goodput across the surviving (alive) legs — the + // denominator each route's share bar is drawn against (this route's fraction + // of the whole group's up / down goodput). Summed here rather than plumbed + // from AggGoodput*Bps so the shared adapter stays self-contained on the leg + // list both surfaces already carry. + var aggUp, aggDown float64 + for _, l := range legs { + aggUp += l.GoodputUpBps + aggDown += l.GoodputDownBps + } + w := legSumWidths(legs) for _, l := range legs { - root.Right = append(root.Right, routeToNode(l, w)) + root.Right = append(root.Right, routeToNode(l, w, aggUp, aggDown)) } return root } @@ -101,11 +116,16 @@ const ( rttColWidth = 6 // bwColWidth fits "0B↑" through "999.9K↑" / "15.3M↑" (compactBytes + arrow). bwColWidth = 7 - // gpColWidth fits the per-leg goodput rate cell, "—" through "999.9K/s" / + // gpColWidth fits a per-direction goodput rate cell, "—" through "999.9K/s" / // "15.3M/s" (compactBytes + "/s"). Same pin-the-width reasoning as bwColWidth: // the rate changes on every live push, so a fixed column keeps the tree that - // follows it column-stable. + // follows it column-stable. Used for BOTH the up-rate and the down-rate. gpColWidth = 9 + // shareBarWidth is the cell count of each per-direction bandwidth-SHARE bar + // (████░ style) — this route's fraction of the route-group's aggregate + // up-goodput / down-goodput. Constant width by construction, so it never + // reflows the tree as the shares shift on a live push. + shareBarWidth = 5 ) // legSumWidths measures the widest R[n] identity across the legs (that count is @@ -191,13 +211,14 @@ func hopClassMap(snap Snapshot) map[string]string { // a template row that lines up with the columns of the tree beneath. Only the // page uses it; `proxy tree` renders headerless. func TreeHeader() (left, label string, cols []string) { - return "R[n] · state · route-rtt · bw ↑↓ · goodput/s", "peer-pk", []string{"[type]", "tp-id", "tp-rtt"} + return "R[n] · state · route-rtt · ↑bytes rate share · ↓bytes rate share", "peer-pk", []string{"[type]", "tp-id", "tp-rtt"} } // routeToNode turns one leg into a hop-chain right-branch carrying its left -// summary on the head (spine) row. -func routeToNode(l Leg, w sumWidths) *bitree.Node { - left := &bitree.Node{Label: legSummary(l, w)} +// summary on the head (spine) row. aggUp/aggDown are the route-group's +// aggregate up/down goodput, the denominators for this leg's share bars. +func routeToNode(l Leg, w sumWidths, aggUp, aggDown float64) *bitree.Node { + left := &bitree.Node{Label: legSummary(l, w, aggUp, aggDown)} if len(l.Hops) == 0 { // No recorded path: a single leaf at the remote PK. @@ -224,12 +245,14 @@ func hopToNode(h Hop) *bitree.Node { } // legSummary is the left annotation for a route: identity, state glyph, -// end-to-end route rtt, and bandwidth. No state word — the glyph (colored by -// the surface's StyleCell) carries active vs standby. Each field is padded to a -// common width (w) so the fields line up in fixed columns across all routes: the -// R[n] identity left-justified, the numeric route-rtt and ↑/↓ bandwidth -// right-justified. -func legSummary(l Leg, w sumWidths) string { +// end-to-end route rtt, and per-direction bandwidth. No state word — the glyph +// (colored by the surface's StyleCell) carries active vs standby. Each field is +// padded to a common width (w) so the fields line up in fixed columns across all +// routes. The bandwidth is shown per direction: the cumulative ↑ byte total then +// this route's live send-RATE and its share of the group's up-goodput, then the +// ↓ total, live recv-RATE and its share of the group's down-goodput. aggUp/aggDown +// are the group's aggregate up/down goodput (the share-bar denominators). +func legSummary(l Leg, w sumWidths, aggUp, aggDown float64) string { g := GlyphActive if l.Standby { g = GlyphStandby @@ -237,9 +260,14 @@ func legSummary(l Leg, w sumWidths) string { idx := padRightRunes(fmt.Sprintf("R[%d]", l.Index), w.idx) rtt := padLeftRunes(routeRTTCompact(l.RouteLatencyMS), w.rtt) up := padLeftRunes(compactBytes(l.SentBytes)+"↑", w.up) + upRate := padLeftRunes(compactRate(l.GoodputUpBps), w.gp) + upBar := shareBar(l.GoodputUpBps, aggUp, shareBarWidth) down := padLeftRunes(compactBytes(l.RecvBytes)+"↓", w.down) - gp := padLeftRunes(compactRate(l.GoodputBps), w.gp) - return idx + " " + g + " " + rtt + " " + up + " " + down + " " + gp + downRate := padLeftRunes(compactRate(l.GoodputDownBps), w.gp) + downBar := shareBar(l.GoodputDownBps, aggDown, shareBarWidth) + return idx + " " + g + " " + rtt + + " " + up + " " + upRate + " " + upBar + + " " + down + " " + downRate + " " + downBar } // compactRate formats a goodput rate (bytes/sec) for the fixed-width leg cell: @@ -253,6 +281,26 @@ func compactRate(bps float64) string { return compactBytes(uint64(bps)) + "/s" } +// shareBar renders a fixed-width unicode meter (████░) of val's fraction of +// total — this route's share of the group's aggregate up- or down-goodput. The +// filled cell count rounds val/total*width; an all-empty bar (░░░░░) means no +// share (total is zero, or this leg is idle in that direction). Constant width by +// construction so it never reflows the columns to its right on a live push. Same +// glyph block as the tree's box-drawing, so the system-monospace stack tiles it. +func shareBar(val, total float64, width int) string { + filled := 0 + if total > 0 && val > 0 { + filled = int(val/total*float64(width) + 0.5) + if filled > width { + filled = width + } + if filled == 0 { + filled = 1 // a nonzero share always shows at least one cell + } + } + return strings.Repeat("█", filled) + strings.Repeat("░", width-filled) +} + // padRightRunes/padLeftRunes pad s to n display columns (rune count), on the // right (left-justify) and left (right-justify) respectively. func padRightRunes(s string, n int) string { diff --git a/pkg/router/route_group.go b/pkg/router/route_group.go index cb3b256f46..efc950d75b 100644 --- a/pkg/router/route_group.go +++ b/pkg/router/route_group.go @@ -485,7 +485,20 @@ func (rg *RouteGroup) MuxStats() MuxInfo { } } leg.Hops[0].LatencyMS = leg.LatencyMS - if len(leg.Hops) == 2 && leg.RouteLatencyMS > leg.LatencyMS { + // For a DIRECT (1-hop) leg the whole route IS this single + // transport hop, so its live end-to-end route latency and the + // hop's transport RTT are the same physical measurement. Prefer + // the E2E value (RouteLatencyMS) — it is the EWMA-smoothed + // leg-liveness pong sampled every legLivenessInterval (30s), + // whereas leg.LatencyMS is tp.GetLatency(): the RAW last sample + // of the 60s transport-ping loop (SetLatency overwrites Avg, it + // is not smoothed), so a single spike sticks for up to a minute + // and the tree's left route-rtt and right transport-rtt disagree. + // Mirrors snapshotLegs' E2E-preferred latency. Multihop legs keep + // the near-edge transport RTT on hop 0. + if leg.Direct && leg.RouteLatencyMS > 0 { + leg.Hops[0].LatencyMS = leg.RouteLatencyMS + } else if len(leg.Hops) == 2 && leg.RouteLatencyMS > leg.LatencyMS { leg.Hops[1].LatencyMS = leg.RouteLatencyMS - leg.LatencyMS } } diff --git a/pkg/router/route_mux.go b/pkg/router/route_mux.go index f04c11bd8a..2989732b08 100644 --- a/pkg/router/route_mux.go +++ b/pkg/router/route_mux.go @@ -38,12 +38,17 @@ type LegStats struct { // the signal a routing policy needs to shed lossy intermediates and // the scheduler needs to deweight them. Retransmits uint64 - // GoodputBps is this leg's recent goodput — the EWMA of (sent+recv) - // bytes per second over the telemetry refresh window (~1s for the - // status page). Distinct from the cumulative SentBytes/RecvBytes - // counters above: it is the RATE a throughput display (and a - // capacity-weighting policy) wants, not the lifetime total. 0 until a - // second sample lands. Sampled in snapshotLegs. + // GoodputUpBps / GoodputDownBps are this leg's recent goodput split by + // direction — the EWMA of the SENT (up) and RECV (down) byte deltas per + // second over the telemetry refresh window (~1s for the status page). + // Distinct from the cumulative SentBytes/RecvBytes counters above: they + // are the RATE each direction is moving now, not the lifetime total. 0 + // until a second sample lands. Sampled in snapshotLegs. + GoodputUpBps float64 + GoodputDownBps float64 + // GoodputBps is the combined (up+down) recent goodput, retained as the + // sum of GoodputUpBps+GoodputDownBps for back-compat with callers that + // want a single figure. GoodputBps float64 } @@ -70,15 +75,19 @@ type legCounters struct { lastTotalBytes uint64 // Goodput-rate sampling (bytes/sec EWMA over the observer's refresh // window), maintained by snapshotLegs — NOT the data path and NOT the - // capacity rebuild. lastRateBytes/lastRateNano snapshot sent+recv and the - // wall clock at the previous sample; the delta over the elapsed window, - // EWMA-smoothed, is goodputBps. Kept separate from lastTotalBytes (which - // the capacity rebuild resets on its own cadence) so the two samplers - // never disturb each other. Touched under legMu (snapshotLegs upgrades to - // a write lock for the sample), so no atomic needed. - lastRateBytes uint64 - lastRateNano int64 - goodputBps float64 + // capacity rebuild. lastRateSentBytes/lastRateRecvBytes/lastRateNano + // snapshot the sent counter, recv counter and wall clock at the previous + // sample; each direction's delta over the elapsed window, EWMA-smoothed, + // is goodputUpBps (sent/sec) / goodputDownBps (recv/sec). Kept separate + // from lastTotalBytes (which the capacity rebuild resets on its own + // cadence) so the two samplers never disturb each other. Touched under + // legMu (snapshotLegs upgrades to a write lock for the sample), so no + // atomic needed. + lastRateSentBytes uint64 + lastRateRecvBytes uint64 + lastRateNano int64 + goodputUpBps float64 + goodputDownBps float64 } // routeMux encapsulates route multiplexing state and logic. @@ -563,29 +572,33 @@ func (m *routeMux) snapshotLegs() []LegStats { for i, c := range m.legs { sent := atomic.LoadUint64(&c.sentBytes) recv := atomic.LoadUint64(&c.recvBytes) - m.sampleGoodput(c, sent+recv, now) + m.sampleGoodput(c, sent, recv, now) out[i] = LegStats{ - Index: i, - SentBytes: sent, - SentPackets: atomic.LoadUint64(&c.sentPackets), - RecvBytes: recv, - RecvPackets: atomic.LoadUint64(&c.recvPackets), - Retransmits: atomic.LoadUint64(&c.retransmits), - GoodputBps: c.goodputBps, + Index: i, + SentBytes: sent, + SentPackets: atomic.LoadUint64(&c.sentPackets), + RecvBytes: recv, + RecvPackets: atomic.LoadUint64(&c.recvPackets), + Retransmits: atomic.LoadUint64(&c.retransmits), + GoodputUpBps: c.goodputUpBps, + GoodputDownBps: c.goodputDownBps, + GoodputBps: c.goodputUpBps + c.goodputDownBps, } } m.legMu.Unlock() return out } -// sampleGoodput updates leg c's goodput EWMA from the total (sent+recv) byte -// count observed at wall-clock now (UnixNano). Caller holds legMu for writing. -// The first observation only seeds the baseline (no rate emitted). To keep the -// metric stable when several observers interleave, samples closer together than -// goodputMinSampleNano are skipped and the stored rate is left unchanged. -func (m *routeMux) sampleGoodput(c *legCounters, total uint64, now int64) { +// sampleGoodput updates leg c's per-direction goodput EWMAs from the sent and +// recv byte counters observed at wall-clock now (UnixNano). Caller holds legMu +// for writing. The first observation only seeds the baseline (no rate emitted). +// To keep the metric stable when several observers interleave, samples closer +// together than goodputMinSampleNano are skipped and the stored rates are left +// unchanged. +func (m *routeMux) sampleGoodput(c *legCounters, sent, recv uint64, now int64) { if c.lastRateNano == 0 { - c.lastRateBytes = total + c.lastRateSentBytes = sent + c.lastRateRecvBytes = recv c.lastRateNano = now return } @@ -593,18 +606,31 @@ func (m *routeMux) sampleGoodput(c *legCounters, total uint64, now int64) { if elapsed < goodputMinSampleNano { return } - var delta uint64 - if total >= c.lastRateBytes { - delta = total - c.lastRateBytes + secs := float64(elapsed) / float64(time.Second) + c.goodputUpBps = ewmaRate(c.goodputUpBps, byteDelta(sent, c.lastRateSentBytes), secs) + c.goodputDownBps = ewmaRate(c.goodputDownBps, byteDelta(recv, c.lastRateRecvBytes), secs) + c.lastRateSentBytes = sent + c.lastRateRecvBytes = recv + c.lastRateNano = now +} + +// byteDelta is cur-prev, clamped at 0 so a counter reset (route rebuild) yields +// no negative rate. +func byteDelta(cur, prev uint64) uint64 { + if cur >= prev { + return cur - prev } - sample := float64(delta) / (float64(elapsed) / float64(time.Second)) - if c.goodputBps == 0 { - c.goodputBps = sample - } else { - c.goodputBps = goodputEWMAAlpha*sample + (1-goodputEWMAAlpha)*c.goodputBps + return 0 +} + +// ewmaRate folds a byte delta over secs seconds into the running bytes/sec EWMA +// (goodputEWMAAlpha weights the newest sample). A zero prior seeds directly. +func ewmaRate(prev float64, delta uint64, secs float64) float64 { + sample := float64(delta) / secs + if prev == 0 { + return sample } - c.lastRateBytes = total - c.lastRateNano = now + return goodputEWMAAlpha*sample + (1-goodputEWMAAlpha)*prev } // wrapPayload creates a sequenced data packet and optionally stores it for retransmission. diff --git a/pkg/visor/api_routing.go b/pkg/visor/api_routing.go index 099c2fe52a..c42a3e7241 100644 --- a/pkg/visor/api_routing.go +++ b/pkg/visor/api_routing.go @@ -150,6 +150,8 @@ func muxRouteGroupInfoFrom(infos []router.MuxInfo) []MuxRouteGroupInfo { entry.AggSentBytes += leg.SentBytes entry.AggRecvBytes += leg.RecvBytes entry.AggGoodputBps += leg.GoodputBps + entry.AggGoodputUpBps += leg.GoodputUpBps + entry.AggGoodputDownBps += leg.GoodputDownBps entry.Legs = append(entry.Legs, MuxLegInfo{ Index: leg.Index, TransportID: leg.TransportID, @@ -165,6 +167,8 @@ func muxRouteGroupInfoFrom(infos []router.MuxInfo) []MuxRouteGroupInfo { RecvPackets: leg.RecvPackets, Retransmits: leg.Retransmits, GoodputBps: leg.GoodputBps, + GoodputUpBps: leg.GoodputUpBps, + GoodputDownBps: leg.GoodputDownBps, Alive: leg.Alive, Standby: leg.Standby, }) diff --git a/pkg/visor/embedded_proxystatus.go b/pkg/visor/embedded_proxystatus.go index 4bf7f38c20..4a2e212a27 100644 --- a/pkg/visor/embedded_proxystatus.go +++ b/pkg/visor/embedded_proxystatus.go @@ -134,6 +134,8 @@ func (p *visorStatusProvider) StatusSnapshot(surface proxystatus.Surface) (proxy RecvBytes: leg.RecvBytes, Retransmits: leg.Retransmits, GoodputBps: leg.GoodputBps, + GoodputUpBps: leg.GoodputUpBps, + GoodputDownBps: leg.GoodputDownBps, Alive: leg.Alive, Standby: leg.Standby, Hops: proxyHopsFrom(leg.Hops), diff --git a/pkg/visor/rpc.go b/pkg/visor/rpc.go index edf186493b..2614e62055 100644 --- a/pkg/visor/rpc.go +++ b/pkg/visor/rpc.go @@ -310,8 +310,12 @@ type MuxRouteGroupInfo struct { // AggGoodputBps is the route-group's recent goodput — the sum of the // per-leg goodput RATES (bytes/sec), i.e. what the whole group is moving // right now, distinct from the cumulative AggSentBytes/AggRecvBytes totals. - AggGoodputBps float64 `json:"agg_goodput_bps,omitempty"` - Legs []MuxLegInfo `json:"legs"` + // AggGoodputUpBps/AggGoodputDownBps split it by direction (sum of the + // per-leg send-rates / recv-rates); AggGoodputBps is their sum. + AggGoodputBps float64 `json:"agg_goodput_bps,omitempty"` + AggGoodputUpBps float64 `json:"agg_goodput_up_bps,omitempty"` + AggGoodputDownBps float64 `json:"agg_goodput_down_bps,omitempty"` + Legs []MuxLegInfo `json:"legs"` } // MuxLegInfo is one route in a mux'd group. @@ -340,9 +344,13 @@ type MuxLegInfo struct { // per second over the telemetry refresh window (~1s for the status page). // The RATE the leg is currently moving, as opposed to the cumulative // SentBytes/RecvBytes totals. 0 until a second sample lands. - GoodputBps float64 `json:"goodput_bps,omitempty"` - Alive bool `json:"alive"` - Standby bool `json:"standby"` + // GoodputUpBps/GoodputDownBps split it by direction (send-rate / recv-rate); + // GoodputBps is their sum. + GoodputBps float64 `json:"goodput_bps,omitempty"` + GoodputUpBps float64 `json:"goodput_up_bps,omitempty"` + GoodputDownBps float64 `json:"goodput_down_bps,omitempty"` + Alive bool `json:"alive"` + Standby bool `json:"standby"` // Hops is the leg's full forward route (every hop to the destination), // with full PKs and per-hop transport type + latency where known. Hops []MuxHopInfo `json:"hops,omitempty"`