treeprobe

type Aggregator ¶

type Aggregator struct {
	// contains filtered or unexported fields
}

Aggregator collects Decoded events into a per-cell map + tracks level-and-run totals. Safe for single-goroutine use only — the harness feeds it from one parser loop.

func NewAggregator() *Aggregator

func (a *Aggregator) CacheHitRate() float64

func (a *Aggregator) Cells() []*Cell

func (a *Aggregator) Levels() []*LevelDone

func (a *Aggregator) Observe(d *Decoded)

func (a *Aggregator) RunDone() *RunDone

func (a *Aggregator) ServerError() *ServerError

func (a *Aggregator) StatusUpdateCount() int

type Cell ¶

type Cell struct {
	Key        CellKey
	Discovered *Discovered
	Result     *PingResult
}

Cell is one (level, parent, remote) entry's collected state. Result is the canonical metric carrier from the server; Discovered is set if a `discovered` event preceded the result (always true in well-formed streams).

type CellKey ¶

type CellKey struct {
	Level    int32
	ParentPK string
	RemotePK string
}

CellKey identifies one (level, parent_pk, remote_pk) tuple. Used as a map key on the aggregator + as the CSV row identity.

type Decoded ¶

type Decoded struct {
	TS   string
	Type EventType

	Discovered   *Discovered
	PingResult   *PingResult
	LevelDone    *LevelDone
	RunDone      *RunDone
	StatusUpdate *StatusUpdate
	ServerError  *ServerError
}

Decoded is the parser's emit shape. Exactly one of the *Payload fields is non-nil per Decoded, matching Envelope.Type. The TS field is the RFC3339Nano timestamp string from the envelope — callers convert to time.Time on demand.

type Discovered ¶

type Discovered struct {
	TpID     string `json:"tp_id"`
	TpType   string `json:"tp_type"`
	RemotePK string `json:"remote_pk"`
	ParentPK string `json:"parent_pk"`
	Level    int32  `json:"level"`
}

Discovered fires when the BFS first observes a new (transport, remote_pk) pair, before any ping is attempted.

type Envelope ¶

type Envelope struct {
	TS   string          `json:"ts"`
	Type EventType       `json:"type"`
	Data json.RawMessage `json:"data"`
}

Envelope is the outer NDJSON line shape. Data is held as json.RawMessage so the consumer can defer decoding to the type-specific struct once it has Type in hand.

type EventType ¶

type EventType string

EventType is the discriminator string in the envelope's "type" field. Mirrors the classifyPayload switch in cmd/skywire-cli/commands/visor/ping/tree_stream.go.

const (
	TypeDiscovered   EventType = "discovered"
	TypePingResult   EventType = "ping_result"
	TypeLevelDone    EventType = "level_done"
	TypeRunDone      EventType = "run_done"
	TypeStatusUpdate EventType = "status_update"
	TypeServerError  EventType = "server_error"
)

type Int64String ¶

type Int64String int64

Int64String is the JSON shape protobuf uses for int64 fields: always a JSON string. Wrapping it in a named type lets the parser surface the convention to consumers + lets us add convenience methods for nanosecond → time.Duration conversion without polluting every struct field.

func (i Int64String) MarshalJSON() ([]byte, error)

func (i *Int64String) UnmarshalJSON(b []byte) error

type LevelDone ¶

type LevelDone struct {
	Level         int32 `json:"level"`
	Attempted     int32 `json:"attempted"`
	Succeeded     int32 `json:"succeeded"`
	Failed        int32 `json:"failed"`
	SkippedCached int32 `json:"skipped_cached"`
}

LevelDone fires when the BFS finishes all peers at a given level. SkippedCached is non-zero when transport_summary fast-path short-circuited some peers (Beta's PR #2733). Lets the harness compute use_transport_latency hit-rate per level.

type Parser ¶

type Parser struct {
	// contains filtered or unexported fields
}

Parser wraps a bufio.Scanner over an NDJSON stream from `cli visor ping tree-stream`. Use NewParser to construct, then Next() in a loop until io.EOF.

func NewParser(r io.Reader) *Parser

func (p *Parser) Next() (*Decoded, error)

type PingResult ¶

type PingResult struct {
	TpID     string `json:"tp_id"`
	TpType   string `json:"tp_type"`
	RemotePK string `json:"remote_pk"`
	ParentPK string `json:"parent_pk"`
	Level    int32  `json:"level"`
	// Proto field name is "canceled" (en-US per misspell linter,
	// renamed from initial pre-review spelling during 2026-05-20
	// design review).
	Canceled bool `json:"canceled"`

	SampleCount  int32       `json:"sample_count"`
	SetupLatency Int64String `json:"setup_latency_ns"`
	PingAvgNs    Int64String `json:"ping_avg_ns"`
	PingP50Ns    Int64String `json:"ping_p50_ns"`
	PingP99Ns    Int64String `json:"ping_p99_ns"`
	// JitterNs is population stddev of ping samples:
	// sqrt(sum((x-mean)^2) / N). Spec'd 2026-05-20 00:54Z.
	JitterNs Int64String `json:"jitter_ns"`

	Failed   bool   `json:"failed"`
	SetupErr string `json:"setup_err,omitempty"`
	PingErr  string `json:"ping_err,omitempty"`
	CalcErr  string `json:"calc_err,omitempty"`

	// LatencySource is one of:
	//   "live_ping"          — fresh measurement
	//   "transport_summary"  — smoothed RTT from TransportSummary.LatencyMS
	//                          (PR #2733, level-1 only)
	//   "skipped"            — cached fast-path explicitly disabled
	LatencySource string     `json:"latency_source"`
	Route         []RouteHop `json:"route,omitempty"`
}

PingResult is the canonical measurement event. One per peer in the tree, regardless of whether the latency was sampled live or pulled from the cached TransportSummary. LatencySource discriminates the provenance.

Wire-NOT-deterministic ordering: results arrive interleaved by completion, not in BFS traversal order. The harness sorts/groups by (Level, ParentPK, RemotePK) on receive.

type RouteHop ¶

type RouteHop struct {
	TpID     string `json:"tp_id"`
	TpType   string `json:"tp_type"`
	RemotePK string `json:"remote_pk"`
}

RouteHop is one entry in the route taken from probe origin to the target peer. PingResult.Route is empty for level-1 entries.

type RunDone ¶

type RunDone struct {
	TotalDiscovered    int32       `json:"total_discovered,omitempty"`
	TotalPinged        int32       `json:"total_pinged,omitempty"`
	TotalSucceeded     int32       `json:"total_succeeded,omitempty"`
	TotalFailed        int32       `json:"total_failed,omitempty"`
	TotalSkippedCached int32       `json:"total_skipped_cached,omitempty"`
	WallTimeNs         Int64String `json:"wall_time_ns"`
	PeakInFlight       int32       `json:"peak_in_flight,omitempty"`
	TerminationReason  string      `json:"termination_reason,omitempty"`
}

RunDone fires once at the end of a successful (or canceled) run. Field names match PingTreeRunDone in pkg/visor/rpcgrpc/ping.pb.go; most fields are int32 counters, WallTimeNs is the protobuf int64 → JSON string convention.

TerminationReason is one of:

"max_level"      — hit cfg.MaxLevel
"no_neighbors"   — BFS exhausted before MaxLevel
"hops_target"    — reached cfg.Hops target level
"context_cancel" — client disconnect / server shutdown
"error"          — see ServerError preceding this event

type ServerError ¶

type ServerError struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

ServerError is emitted when the BFS handler hits a fatal error it can't recover from. Always the last event before the stream terminates.

type StatusUpdate ¶

type StatusUpdate struct {
	Phase    string `json:"phase"`
	InFlight int32  `json:"in_flight"`
	Pending  int32  `json:"pending"`
}

StatusUpdate fires periodically (cadence server-chosen) to surface live in-flight + queue state. Dropped first under back- pressure on the bounded event channel.