Help us shape the future of Go! Take the 2026 Go Developer Survey

api

package
v1.3.104 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 31, 2026 License: AGPL-3.0 Imports: 48 Imported by: 0

Documentation

Overview

Package api pkg/deployment/tpd/api/api.go c4-net-discovery

Package api pkg/deployment/tpd/api/ar_ips.go c4-net-discovery

Public IPs from the address resolver's bindings feed, so the per-visor bandwidth day can leave out transports between two visors on one IP. A visor marks such transports itself, but only where it learns the peer's address: a webrtc conn has none to compare, and visors from before the mark send none. The resolver sees every bound visor's public IP, whatever transport carries the bytes.

Package api pkg/deployment/tpd/api/charts.go c4-net-discovery

Package api pkg/deployment/tpd/api/cxo_metrics_publisher.go c4-net-discovery

CXO publisher for the network-wide transport-metrics aggregate. The publisher writes ONE GZIPPED JSON LEAF PER CALENDAR DAY 

metrics/day/<YYYY-MM-DD>

and a day window (1, 7, 30) is assembled reader-side from the N newest leaves. See pkg/deployment/tpd/store/cxo_metrics_layout.go for the pivot/merge pair and for why Live and Latency live only in the newest day's leaf.

This replaced three OVERLAPPING window leaves (metrics/days/1, /7, /30) recomputed on separate tickers. Two things were wrong with that:

  • Content-addressing bought nothing. One leaf per window meant a single changed byte made the whole multi-megabyte object new, so the 30-day window went back over the wire in full every 30 minutes even though 29 of its 30 days could not have changed. Per-day leaves make a settled day hash identically forever, so only the current day actually moves.

  • The recompute was redundant. Every 30 minutes the old scheme asked the store for 30×1 + 6×7 + 1×30 = 102 transport-days; this one asks for 30×1 + 1×30 = 60, because the short windows are no longer separate queries at all.

Two properties of CXO still shape how a body is written:

  • CXO stores and propagates object bytes verbatim — it does not compress. Bodies are gzipped here, as every sibling TPD publisher already does. Readers use cxoutils.Gunzip, which passes a raw body through unchanged.

  • An object over skyobject MaxObjectSize (16 MB) is refused at Put time, and one refused Put kills the whole feed: no Root is ever published and every subscriber sits in "timeout waiting for Root" forever. A single production day is ~16 MB of JSON that gzips to ~4 MB, so a day fits comfortably — but the part-splitting safety net is kept, at

    metrics/day/<YYYY-MM-DD>/part/<NNNN>

    because "it fits today" is not a guarantee about a larger network tomorrow.

Package api pkg/deployment/tpd/api/cxo_perkey_publisher.go c4-net-discovery

CXO publisher for the per-key transport rollup: every visor's transports counted by type, the GET /all-transports/per-key-stats body, at 

perkey/stats

It is the slower tier cxo_stats_publisher.go leaves this body out of. At ~38 KB gzipped it is too big for the stats feed a dashboard holds open, and the whole point is not to be the all-transports snapshot (~2.6 MB) a reader would otherwise download to count per key itself — which is what the network view did on every refresh.

The body carries the same completeness stamp as the stats feed, judged on the transport total, and the same holdover: after a TPD restart the registry refills from near zero, and a per-key table read mid-refill shows most visors with too few transports and nothing to say so.

Package api pkg/deployment/tpd/api/cxo_register.go c4-net-discovery

CXO-driven register / deregister entry points. The visor publishes transports/<uuid>/entry and transports/<uuid>/tombstone leaves on its TreeStore feed; pkg/deployment/tpd/cxoaggregator dispatches those leaves to these methods on the API.

The auth check (reporter must be an edge of the entry / existing entry) is duplicated here even though CXO already authenticates the publishing visor by Root signature: the Root sig proves the visor PK that signed it, but the leaf itself can carry any payload, so TPD has to reject entries that name an unrelated PK as an edge. This is the parity check for the SW-Sig httpauth used by the v2/v3 HTTP register endpoints, where the "auth PK is an edge" filter is applied during registerTransportV3 (see endpoints.go).

Package api pkg/deployment/tpd/api/cxo_routing_publisher.go c4-net-discovery

CXO routing feed: what visors route on. It carries the transports that exist now (see routingEntries) with the latency and throughput routes are weighed by, as a snapshot sharded one leaf per visor plus the deltas since it (see package routingfeed for the layout): 

routing/<pk>          // gzipped JSON []allTransportsWireEntry
routing/delta/<seq>   // gzipped JSON routingfeed.Delta

A transport lives in the shard of its lower-keyed edge (entries keep their edges sorted), so each appears once. The all-transports feed republished its whole ~19 MB snapshot every minute and, the transport set churning, every subscriber fetched all of it every time. Here an unchanged shard re-encodes to the object subscribers already hold, so a tick ships only the shards that changed. Figures are rounded to two significant digits so probe jitter does not rewrite a shard.

Package api pkg/deployment/tpd/api/cxo_stats_publisher.go c4-net-discovery

CXO publisher for TPD's NETWORK-AGGREGATE stats — the sub-kilobyte reductions that the bulk feeds are otherwise downloaded to compute. 

stats/network   the GET /all-transports/stats shape (138 B live)
stats/versions  the GET /version fleet histogram (300 B live)
stats/daily     the GET /metric daily aggregate (~2.7 KB live)

Why a separate feed rather than another path on an existing one: size is cadence. The metrics and all-transports feeds are staggered to minutes because recomputing them is expensive; the first two bodies are a few hundred bytes and cost a map walk over caches the HTTP handlers already read, so they republish every statsPublishInterval. A chart that wants "transports on the network, every 15 seconds" costs ~400 B per sample here instead of the 2.4 MB all-transports snapshot or the 16.8 MB metrics window it would otherwise reduce locally.

stats/daily is the exception on cadence and the reason statsDailyInterval exists. Its BODY is small — the whole 30-day series with a per-type breakdown in under three kilobytes — but producing it is a 30-day store query, not a map walk, so it runs on its own slow timer inside the same loop rather than on the 12 s tick. It is here rather than on the metrics feed because a consumer that wants the reduction must not be made to subscribe to the ~120 MB of per-transport day leaves the reduction was computed from; that is the entire point of this feed.

The per-key rollup (GET /all-transports/per-key-stats, ~38 KB gzipped) is deliberately NOT here. It is two orders of magnitude larger than these bodies and has its own slower feed (cxo_perkey_publisher.go); putting it on this one would drag the whole feed onto the large-feed first-sync budget and defeat the point of a feed a dashboard can hold continuously.

PARTIAL-AGGREGATE HAZARD (skycoin/skywire#4513). TPD's transport aggregate is READABLE WHILE IT REFILLS after a restart: the count climbs monotonically from near zero to its settled value, every type diluted in proportion, with nothing in the body to say it is partial. TPD restarts on every deploy, so a naive publisher at this cadence would stamp a permanent sawtooth into every time series built on it. This publisher therefore judges each sample against a trailing peak (see completenessTracker) and:

  • stamps every body with observed_at, complete, confidence and the trailing peak it was judged against, so a consumer can filter;
  • HOLDS the last complete sample on the feed rather than overwriting it with one that looks partial, for up to statsIncompleteHoldover.

The holdover bound rather than an indefinite hold is deliberate: a genuine network-wide drop below the trailing peak (a dmsg server outage, say) is indistinguishable from a refill at sample time, and a feed that silently freezes on real news is worse than one that publishes the news marked complete=false. After the bound the sample goes out with its honest verdict attached.

Package api pkg/deployment/tpd/api/cxo_uptime_publisher.go c4-net-discovery

CXO publisher for the network-wide visor-uptime aggregate. Mirrors MetricsCXOPublisher (cxo_metrics_publisher.go) but produces the `GET /uptimes?v=v3` shape — `[]VisorSummary` with the per-day timeline strings — so subscribers (the hvui Network Uptime tab, reached through a visor's on-demand subscriber) get the exact online/offline intervals the integrated tracker recorded.

Same content-addressing benefit as the metrics publisher: only the visors whose Daily/Timeline buckets actually changed since the last tick produce a Root delta, so the wire footprint scales with churn rather than fleet size.

Package api pkg/deployment/tpd/api/cxo_visorbw_publisher.go c4-net-discovery

CXO per-visor bandwidth feed: what the reward system pays pool 2 from. 

visorbw/day/<YYYY-MM-DD>   gzipped JSON store.VisorBWDay

One leaf per SETTLED day, for the last visorBWWindowDays days. Each is reduced once from the per-transport records the metrics publisher saved when the day settled, so it is final: the bytes of a past day do not change. Transports a visor marked as reaching a peer on its own network are left out (see pkg/deployment/tpd/store/visorbw.go).

Package api pkg/deployment/tpd/api/dmsg_roles.go

Package api pkg/deployment/tpd/api/edge_response_cache.go c4-net-discovery

Package api pkg/deployment/tpd/api/endpoints.go c4-net-discovery

Package api pkg/deployment/tpd/api/entries_json.go c4-net-discovery

Package api pkg/deployment/tpd/api/metrics.go c4-net-discovery

Package api pkg/deployment/tpd/api/ratelimit.go c4-net-discovery

Package api pkg/deployment/tpd/api/reconcile_throttle.go c4-net-discovery

Package api pkg/deployment/tpd/api/response_cache.go c4-net-discovery

Package api pkg/deployment/tpd/api/routing_entries.go c4-net-discovery

Package api pkg/deployment/tpd/api/routing_smoothing.go c4-net-discovery

Package api pkg/deployment/tpd/api/visor_table.go c4-net-discovery

Index

Constants

View Source 
const (
	// SnapshotAtHeader carries the RFC3339Nano instant the serving snapshot
	// was filled. Two bodies with the same value are two views of one set.
	SnapshotAtHeader = "X-Tpd-Snapshot-At"
	// SnapshotTransportsHeader carries how many transports that snapshot
	// holds, so the per-key body can be checked against its OWN aggregate
	// rather than against whatever the aggregate reads at some other moment.
	SnapshotTransportsHeader = "X-Tpd-Snapshot-Transports"
)

Snapshot headers. Absent on a cold-cache response, which is the honest answer: that body was counted straight out of the store and shares no snapshot with anything.

View Source 
const (
	// StatsPathNetwork carries the network-wide transport aggregate
	// (the GET /all-transports/stats shape plus a completeness stamp).
	StatsPathNetwork = "stats/network"
	// StatsPathVersions carries the fleet version histogram (the
	// GET /version shape plus a completeness stamp).
	StatsPathVersions = "stats/versions"
	// StatsPathDaily carries the network-wide daily aggregate (the
	// GET /metric shape plus a completeness stamp) — per-day
	// bandwidth, latency and by-type breakdown over statsDailyDays.
	StatsPathDaily = "stats/daily"
)

Published paths. Exported so visor-side subscribers don't have to duplicate the strings.

View Source 
const (
	// ConfidenceSettled means the sample reached statsCompleteRatio of
	// the trailing peak — safe to chart as an absolute count.
	ConfidenceSettled = "settled"
	// ConfidenceRefilling means the sample fell materially below the
	// trailing peak. Under #4513 that is the signature of reading the
	// aggregate mid-rebuild; treat the value as a lower bound.
	ConfidenceRefilling = "refilling"
	// ConfidenceWarmup means the publisher has not observed long
	// enough to have a trustworthy peak to judge against.
	ConfidenceWarmup = "warmup"
)

Confidence values stamped on a published body.

View Source 
const (
	DefaultRoutingLatencyWindow    = 6
	DefaultRoutingSmoothHysteresis = 0.25
)

Routing smoothing defaults. A 10-minute feed with a 6-publish window takes the minimum latency of the last hour.

View Source 
const DefaultRoutingPublishEvery = 10 * time.Minute

DefaultRoutingPublishEvery is how often the routing shards are recomputed when the config leaves it empty. Telemetry is applied every 15 minutes, so publishing faster resends whole shards for noise to every subscriber.

View Source 
const DefaultRoutingSnapshotEvery = 6

DefaultRoutingSnapshotEvery is how many publishes go by between snapshots when the config leaves it empty: an hour at the default interval.

View Source 
const PerKeyPath = "perkey/stats"

PerKeyPath is the one leaf the per-key feed writes.

View Source 
const RoutingPathPrefix = routingfeed.Prefix

RoutingPathPrefix is the routing feed's sub-tree; shard paths are RoutingPathPrefix + <pk hex>.

Variables

View Source 
var (
	// ErrEmptyPubKey indicates that provided public key is empty.
	ErrEmptyPubKey = errors.New("public key cannot be empty")
	// ErrInvalidPubKey indicates that provided public key is invalid.
	ErrInvalidPubKey = errors.New("public key is invalid")
	// ErrEmptyTransportID indicates that provided transport ID is empty.
	ErrEmptyTransportID = errors.New("transport ID cannot be empty")
	// ErrInvalidTransportID indicates that provided transport ID is invalid.
	ErrInvalidTransportID = errors.New("transport ID is invalid")
	// ErrBadInput occurs in case of bad input
	ErrBadInput = errors.New("error bad input")
	// WhitelistPKs holds the keys allowed to call the bulk deregistration endpoint.
	WhitelistPKs = httpauth.NewWhitelist()
)

Functions

func UptimePath 

func UptimePath(days int) string

UptimePath returns the TreeStore path for a given day window. Exported so visor-side subscribers don't have to duplicate the format string.

Types

type API 

type API struct {

	// Stats, when set, adds the process and traffic charts to the status page.
	Stats *charts.ServiceStats

	http.Handler

	DmsgServers []string
	// contains filtered or unexported fields
}

API register all the API endpoints. It implements a net/http.Handler.

func New 

func New(log logrus.FieldLogger, s store.Store, nonceStore httpauth.NonceStore,
	enableMetrics bool, m tpdiscmetrics.Metrics, dmsgAddr string, backupPath string) *API

New constructs a new API instance.

func (*API) BackfillDHTMirror 

func (api *API) BackfillDHTMirror(ctx context.Context, log logrus.FieldLogger)

BackfillDHTMirror mirrors the full transport list for every edge PK seen in the discovery on startup. Producing one DHT item per edge (value = all that edge's transports) matches the semantics of GET /transports/edge:{pk} and avoids the pre-#2333 data loss where per-transport mirroring overwrote each other's target.

func (*API) Charts added in v1.3.101 

func (api *API) Charts() *charts.Page

Charts is the charts page, or nil until StartCharts runs.

func (*API) ChartsExtra 

func (api *API) ChartsExtra() []charts.Extra

ChartsExtra is the transport graph and its engine, served next to the charts page.

func (*API) ChartsPage added in v1.3.100 

func (api *API) ChartsPage(w http.ResponseWriter, r *http.Request)

ChartsPage serves the charts page, or 404 when charts are not running.

func (*API) DeregisterTransportFromCXO 

func (api *API) DeregisterTransportFromCXO(ctx context.Context, id uuid.UUID, reporter cipher.PubKey) error

DeregisterTransportFromCXO applies a tombstone published by a visor on its TreeStore feed. An unknown ID is a no-op (idempotent — a tombstone that arrives after TPD-side eviction is normal).

func (*API) GraphPage added in v1.3.100 

func (api *API) GraphPage(w http.ResponseWriter, r *http.Request)

GraphPage serves the transport graph, or 404 when charts are not running.

func (*API) ReconcileTransportsFromCXO 

func (api *API) ReconcileTransportsFromCXO(ctx context.Context, entries []*transport.Entry, reporter cipher.PubKey, version string) error

ReconcileTransportsFromCXO applies a visor's full transport list published as a single snapshot leaf (transports/list). It is the declarative replacement for the per-transport entry (register) + tombstone (deregister) leaves: register/refresh every entry the reporter is an edge of, and deregister any of the reporter's existing transports that are ABSENT from the list (absence = deletion). Idempotent and self-healing — a dropped update is corrected by the next snapshot.

func (*API) RefreshTransportsFromCXO 

func (api *API) RefreshTransportsFromCXO(ctx context.Context, entries []*transport.Entry, reporter cipher.PubKey, version string) error

RefreshTransportsFromCXO registers and refreshes the reporter's listed transports like ReconcileTransportsFromCXO, but deletes nothing. It is for a list that is known to be old — the aggregator re-applying a reporter's last good snapshot after a failed fetch — whose absences prove nothing: a transport created since (say by a browser tab dialing this hub) is simply not in it, and deleting on that basis removed the other edge's fresh entry.

func (*API) RegisterTransportFromCXO 

func (api *API) RegisterTransportFromCXO(ctx context.Context, entry *transport.Entry, reporter cipher.PubKey, version string) error

RegisterTransportFromCXO applies a transport entry published by a visor on its TreeStore feed. Mirrors the work the v3 HTTP register handler does: store write, per-entry heartbeat, mirrorEdges, and per-visor heartbeat. Returns an error on auth failure or store failure; the aggregator logs it at debug.

func (*API) RunBackgroundTasks 

func (api *API) RunBackgroundTasks(ctx context.Context, logger logrus.FieldLogger)

RunBackgroundTasks is function which runs periodic background tasks of API.

func (*API) SeedReconcile added in v1.3.99 

func (api *API) SeedReconcile(ctx context.Context) error

SeedReconcile tells the CXO reconcile which transports the store already holds, so it touches rather than rewrites them after a restart. Call after SetEntryTimeout, and only when the store answers GetAllTransports from memory (its live set): otherwise the read is the whole redis index.

func (*API) SetARBindingsFeed 

func (api *API) SetARBindingsFeed(mgr *cxosub.Manager)

SetARBindingsFeed has the per-visor bandwidth day treat two visors the address resolver sees at one public IP as on one network.

func (*API) SetDHTMirror 

func (api *API) SetDHTMirror(m interface {
	Mirror(subjectPK cipher.PubKey, entry interface{}, seq uint64)
	MirrorMany(subjectPKs []cipher.PubKey, entry interface{}, seq uint64)
	Delete(subjectPK cipher.PubKey)
})

SetDHTMirror sets a mirror that publishes transport lists to the DHT.

func (*API) SetDmsgDiscovery added in v1.3.100 

func (api *API) SetDmsgDiscovery(client *http.Client, url string)

SetDmsgDiscovery lets TPD ask dmsg discovery at url (dmsg://<pk>:<port>) which visors run dmsg servers, over client.

func (*API) SetDmsgDiscoveryFeed added in v1.3.100 

func (api *API) SetDmsgDiscoveryFeed(mgr *cxosub.Manager)

SetDmsgDiscoveryFeed has the role lookup read the servers from mgr's subscription to dmsg discovery's clients-by-server feed, falling back to HTTP while that has no servers. Call after SetDmsgDiscovery.

func (*API) SetEntryTimeout 

func (api *API) SetEntryTimeout(ttl time.Duration)

SetEntryTimeout tells the API the store's transport entry TTL so the CXO reconcile path refreshes an unchanged registration a third of the way through it rather than on every visor heartbeat (see reconcileThrottle).

func (*API) SetUptimeRecorder 

func (api *API) SetUptimeRecorder(r *serviceuptime.Recorder)

SetUptimeRecorder wires a service-self uptime recorder onto the API. Idempotent re-attach is a no-op (first writer wins). Until set, the /uptime/* routes return 503.

func (*API) StartCharts added in v1.3.100 

func (api *API) StartCharts(ctx context.Context, st charts.Store, log logrus.FieldLogger)

StartCharts samples TPD's counts into st until ctx ends and serves them as the charts page.

type DailyStats 

type DailyStats struct {
	Daily      []store.DailyAggregate     `json:"daily"`
	Cumulative *store.CumulativeAggregate `json:"cumulative"`

	// Days is the window requested from the store. TPD returns only the
	// history it holds, so len(Daily) may be shorter.
	Days int `json:"days"`

	ObservedAt time.Time `json:"observed_at"`
	Complete   bool      `json:"complete"`
	Confidence string    `json:"confidence"`
	// TrailingPeak is the transport-count peak the sample was judged
	// against — the same field stats/network publishes, repeated here
	// so this body stands on its own.
	TrailingPeak          int `json:"trailing_peak_transports"`
	TrailingWindowSeconds int `json:"trailing_window_seconds"`
}

DailyStats is the body published at StatsPathDaily. The first two fields are store.NetworkMetricResponse verbatim — a consumer that unmarshals this into store.NetworkMetricResponse gets exactly what GET /metric returns, newest day first — and the rest is the completeness stamp.

The stamp is the TRANSPORT-SET verdict, not one computed from the bandwidth figures. These are sums over the same registry stats/network counts, so "the registry is still refilling" is precisely the thing that makes a day's total a lower bound; judging the sums themselves against a trailing peak would instead flag every ordinary quiet day as partial.

type Error 

type Error struct {
	Error string `json:"error"`
}

Error is the object returned to the client when there's an error.

type HealthCheckResponse 

type HealthCheckResponse = httputil.HealthCheckResponse

HealthCheckResponse is the /health body every service shares.

type MetricsCXOPublisher 

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

MetricsCXOPublisher periodically recomputes the /metrics aggregate and publishes it as one leaf per calendar day. The struct is owned by the API; Close shuts the publisher and stops the ticker.

func StartMetricsCXOPublisher 

func StartMetricsCXOPublisher(ctx context.Context, api *API, dmsgC *dmsg.Client, sk cipher.SecKey, logger logrus.FieldLogger) (*MetricsCXOPublisher, error)

StartMetricsCXOPublisher constructs a publisher backed by the given DMSG client and TPD secret key, then kicks off the recompute ticker. The publisher's allowlist is left open (any subscriber may read) — the metrics endpoint is already a public read on the HTTP side, so the CXO mirror inherits that access policy.

Returns nil + error when the publisher can't be created (no DMSG client, listener bind failure, etc.); the caller should log and continue without it. Treat the publisher as best-effort: the existing HTTP /metrics route stays the source of truth.

func (*MetricsCXOPublisher) Close 

func (m *MetricsCXOPublisher) Close() error

Close stops the ticker and tears down the publisher.

func (*MetricsCXOPublisher) FeedPK 

func (m *MetricsCXOPublisher) FeedPK() cipher.PubKey

FeedPK returns the publisher's feed PK — i.e. TPD's own PK, since the publisher was built with TPD's secret key. Subscribers connect to this PK at port skyenv.DmsgTPDMetricsCXOPort.

func (*MetricsCXOPublisher) LastError 

func (m *MetricsCXOPublisher) LastError() error

LastError returns the most recent error encountered by the publish loop, or nil if the last tick succeeded. Exposed for /health-style introspection if a future caller wants it.

func (*MetricsCXOPublisher) Publisher added in v1.3.98 

func (x *MetricsCXOPublisher) Publisher() *treestore.Publisher

Publisher returns the underlying feed publisher, for introspection.

type NetworkStats 

type NetworkStats struct {
	Total        int            `json:"total_transports"`
	ByType       map[string]int `json:"by_type"`
	UniqueVisors int            `json:"unique_visors"`

	// Partial and MissingBatches are the store's own verdict on the read
	// that produced these counts (see store.TransportSummary): a batched
	// MGET over the transport index whose batches did not all come back.
	// GET /all-transports/stats has carried them since #4542 and this
	// body dropped them, which left a subscriber holding a sample the
	// store already KNEW was an undercount while the only signal on the
	// feed was the trailing-peak heuristic below — an independent
	// estimate that can read "settled" over exactly that body. The two
	// are different claims and both belong here.
	Partial        bool `json:"partial,omitempty"`
	MissingBatches int  `json:"missing_batches,omitempty"`

	ObservedAt time.Time `json:"observed_at"`
	Complete   bool      `json:"complete"`
	Confidence string    `json:"confidence"`
	// TrailingPeak is the highest total_transports seen in the last
	// TrailingWindowSeconds. Published so a consumer can apply its own
	// threshold instead of trusting ours.
	TrailingPeak          int `json:"trailing_peak_transports"`
	TrailingWindowSeconds int `json:"trailing_window_seconds"`
}

NetworkStats is the body published at StatsPathNetwork. The first three fields are store.TransportSummary verbatim — a consumer that unmarshals this into store.TransportSummary gets exactly what GET /all-transports/stats returns — and the rest is the completeness stamp #4513 makes necessary.

type PerKeyCXOPublisher added in v1.3.100 

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

PerKeyCXOPublisher publishes PerKeyStats on a timer. Closed by Close.

func StartPerKeyCXOPublisher added in v1.3.100 

func StartPerKeyCXOPublisher(ctx context.Context, api *API, dmsgC *dmsg.Client, sk cipher.SecKey, logger logrus.FieldLogger) (*PerKeyCXOPublisher, error)

StartPerKeyCXOPublisher constructs the publisher on the given DMSG client and TPD secret key and starts its ticker. The feed is open to any subscriber, like the HTTP endpoint it mirrors. Best-effort: HTTP stays the source of truth, so the caller should log and continue on error.

func (*PerKeyCXOPublisher) Close added in v1.3.100 

func (p *PerKeyCXOPublisher) Close() error

Close stops the ticker and tears down the publisher.

func (*PerKeyCXOPublisher) FeedPK added in v1.3.100 

func (p *PerKeyCXOPublisher) FeedPK() cipher.PubKey

FeedPK returns the publisher's feed PK (TPD's own PK).

func (*PerKeyCXOPublisher) LastError added in v1.3.100 

func (p *PerKeyCXOPublisher) LastError() error

LastError returns the error from the most recent publish attempt, or nil.

func (*PerKeyCXOPublisher) Publisher added in v1.3.100 

func (p *PerKeyCXOPublisher) Publisher() *treestore.Publisher

Publisher returns the underlying feed publisher, for introspection.

type PerKeyStats added in v1.3.100 

type PerKeyStats struct {
	Keys  map[string]map[string]int `json:"keys"`
	Total int                       `json:"total_transports"`

	ObservedAt time.Time `json:"observed_at"`
	Complete   bool      `json:"complete"`
	Confidence string    `json:"confidence"`
	// TrailingPeak is the highest total_transports seen in the last
	// TrailingWindowSeconds.
	TrailingPeak          int `json:"trailing_peak_transports"`
	TrailingWindowSeconds int `json:"trailing_window_seconds"`
}

PerKeyStats is the body published at PerKeyPath. Keys is the /all-transports/per-key-stats body verbatim; the rest is the completeness stamp.

type PubKeyRateLimiter 

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

PubKeyRateLimiter manages rate limiting per public key

func NewPubKeyRateLimiter 

func NewPubKeyRateLimiter(requestsPerMinute int, burst int) *PubKeyRateLimiter

NewPubKeyRateLimiter creates a new rate limiter For 30 requests per minute: rate = 0.5 (30/60 seconds)

func (*PubKeyRateLimiter) Allow 

func (rl *PubKeyRateLimiter) Allow(key string) bool

Allow checks if a request from the given key should be allowed. The key can be a public key (for dmsg requests) or an IP address (for HTTP requests).

func (*PubKeyRateLimiter) GetStats 

func (rl *PubKeyRateLimiter) GetStats() map[string]interface{}

GetStats returns current rate limiter statistics

func (*PubKeyRateLimiter) Middleware 

func (rl *PubKeyRateLimiter) Middleware() func(http.Handler) http.Handler

Middleware returns a Chi middleware that enforces rate limiting per public key

type RoutingCXOPublisher added in v1.3.98 

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

RoutingCXOPublisher publishes the routing feed.

func StartRoutingCXOPublisher added in v1.3.98 

func StartRoutingCXOPublisher(ctx context.Context, api *API, dmsgC *dmsg.Client, sk cipher.SecKey, cfg RoutingFeedConfig, logger logrus.FieldLogger) (*RoutingCXOPublisher, error)

StartRoutingCXOPublisher starts the routing feed on DmsgTPDRoutingCXOPort, configured by cfg.

func (*RoutingCXOPublisher) Close added in v1.3.98 

func (r *RoutingCXOPublisher) Close() error

Close stops the loop and the publisher.

func (*RoutingCXOPublisher) FeedPK added in v1.3.98 

func (r *RoutingCXOPublisher) FeedPK() cipher.PubKey

FeedPK returns the publisher's feed PK (TPD's own PK).

func (*RoutingCXOPublisher) LastError added in v1.3.98 

func (r *RoutingCXOPublisher) LastError() error

LastError reports the most recent publish failure, if any.

func (*RoutingCXOPublisher) Publisher added in v1.3.98 

func (r *RoutingCXOPublisher) Publisher() *treestore.Publisher

Publisher returns the underlying treestore publisher.

type RoutingFeedConfig 

type RoutingFeedConfig struct {
	// Every is the publish interval.
	Every time.Duration
	// SnapshotEvery is how many publishes go by between snapshots; the ones
	// between add a delta each.
	SnapshotEvery int
	Smoothing     RoutingSmoothing
}

RoutingFeedConfig configures the routing feed. Zero values take the defaults.

type RoutingSmoothing 

type RoutingSmoothing struct {
	// LatencyWindow is how many publishes the published latency is the
	// minimum of. The minimum is the path's propagation delay; samples above
	// it are queueing on a busy hop. Zero means DefaultRoutingLatencyWindow.
	LatencyWindow int
	// Hysteresis is the relative change a figure must make before the feed
	// publishes it. Zero means DefaultRoutingSmoothHysteresis; a negative
	// value publishes every change.
	Hysteresis float64
}

RoutingSmoothing sets how the routing feed steadies the figures TPD holds, which are the last value a visor reported.

type StatsCXOPublisher 

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

StatsCXOPublisher publishes the network-aggregate bodies, gating each against a completeness tracker. Closed by Close.

func StartStatsCXOPublisher 

func StartStatsCXOPublisher(ctx context.Context, api *API, dmsgC *dmsg.Client, sk cipher.SecKey, logger logrus.FieldLogger) (*StatsCXOPublisher, error)

StartStatsCXOPublisher constructs a publisher backed by the given DMSG client and TPD secret key, then kicks off the publish ticker. The allowlist is left open (any subscriber may read) — same access policy as the HTTP endpoints it mirrors.

Best-effort: the HTTP routes stay the source of truth, so the caller should log and continue if this returns an error.

func (*StatsCXOPublisher) Close 

func (s *StatsCXOPublisher) Close() error

Close stops the ticker and tears down the publisher.

func (*StatsCXOPublisher) FeedPK 

func (s *StatsCXOPublisher) FeedPK() cipher.PubKey

FeedPK returns the publisher's feed PK (TPD's own PK). Subscribers connect to this PK at port skyenv.DmsgTPDStatsCXOPort.

func (*StatsCXOPublisher) LastError 

func (s *StatsCXOPublisher) LastError() error

LastError returns the most recent error encountered by the publish loop, or nil if the last tick succeeded for both paths.

func (*StatsCXOPublisher) Publisher added in v1.3.98 

func (x *StatsCXOPublisher) Publisher() *treestore.Publisher

Publisher returns the underlying feed publisher, for introspection.

type UptimeCXOPublisher 

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

UptimeCXOPublisher publishes the network-wide visor-uptime aggregate for each of uptimePublishDays as ONE gzipped leaf per window at "uptimes/days/<n>" — the same few-large-objects shape as the all-transports publisher, and for the same reason: a subscriber fills a Root over the transient dmsg conn the publisher's re-announce delivers it on, and that conn closes moments later. A Root of a handful of leaves fills in one short burst; a Root fanned out into one leaf PER VISOR PER WINDOW (the previous shape) needs a request round-trip per object, never completes before the conn drops, and every subscriber fill dies with "no connections to fill from" — so the feed published a Root that no one could ever fill (`cli visor cxo status` → tpd-uptime last_err="timeout waiting for Root after 10s", while the 2-leaf all-transports feed synced fine on the same TPD peer).

The 16 MB MaxObjectSize that drove the earlier per-visor split is not a problem for a single leaf here: each visor is encoded as a compactVisorSummary (36-byte timeline bitmaps, ~8x smaller than the v3 288-char strings that blew past the limit) and the whole window array is gzipped before the Put, so the stored object is well under 16 MB for any realistic fleet. Closed by Close.

func StartUptimeCXOPublisher 

func StartUptimeCXOPublisher(ctx context.Context, api *API, dmsgC *dmsg.Client, sk cipher.SecKey, logger logrus.FieldLogger) (*UptimeCXOPublisher, error)

StartUptimeCXOPublisher constructs a publisher backed by the given DMSG client and TPD secret key, then kicks off the recompute ticker. The publisher's allowlist is left open (any subscriber may read) — same access policy as `GET /uptimes`.

Returns nil + error when the publisher can't be created (no DMSG client, listener bind failure, etc.); the caller should log and continue without it. Best-effort: the existing HTTP /uptimes route remains the source of truth.

func (*UptimeCXOPublisher) Close 

func (u *UptimeCXOPublisher) Close() error

Close stops the ticker and tears down the publisher.

func (*UptimeCXOPublisher) FeedPK 

func (u *UptimeCXOPublisher) FeedPK() cipher.PubKey

FeedPK returns the publisher's feed PK (TPD's own PK). Subscribers connect to this PK at port skyenv.DmsgTPDUptimeCXOPort.

func (*UptimeCXOPublisher) LastError 

func (u *UptimeCXOPublisher) LastError() error

LastError returns the most recent error encountered by the publish loop, or nil if the last tick succeeded for every window.

func (*UptimeCXOPublisher) Publisher added in v1.3.98 

func (x *UptimeCXOPublisher) Publisher() *treestore.Publisher

Publisher returns the underlying feed publisher, for introspection.

type VersionEntry 

type VersionEntry struct {
	PK      string `json:"pk"`
	Version string `json:"version"`
	Online  *bool  `json:"on,omitempty"` // Only included if status=true
}

VersionEntry is a response entry for version endpoints

type VersionStats 

type VersionStats struct {
	Versions map[string]int `json:"versions"`
	Visors   int            `json:"visors"`

	ObservedAt time.Time `json:"observed_at"`
	Complete   bool      `json:"complete"`
	Confidence string    `json:"confidence"`
	// TrailingPeak is the highest visor count seen in the last
	// TrailingWindowSeconds.
	TrailingPeak          int `json:"trailing_peak_visors"`
	TrailingWindowSeconds int `json:"trailing_window_seconds"`
}

VersionStats is the body published at StatsPathVersions. Versions is the GET /version body verbatim; the rest is the completeness stamp. The histogram is derived from the uptime cache, not the transport cache, so it carries its own independent verdict.

type VisorBWCXOPublisher added in v1.3.99 

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

VisorBWCXOPublisher publishes the per-visor bandwidth feed.

func StartVisorBWCXOPublisher added in v1.3.99 

func StartVisorBWCXOPublisher(ctx context.Context, api *API, dmsgC *dmsg.Client, sk cipher.SecKey, logger logrus.FieldLogger) (*VisorBWCXOPublisher, error)

StartVisorBWCXOPublisher starts the per-visor bandwidth feed on DmsgTPDVisorBWCXOPort.

func (*VisorBWCXOPublisher) Close added in v1.3.99 

func (p *VisorBWCXOPublisher) Close() error

Close stops the publisher.

func (*VisorBWCXOPublisher) FeedPK added in v1.3.99 

func (p *VisorBWCXOPublisher) FeedPK() cipher.PubKey

FeedPK is the feed's public key.

func (*VisorBWCXOPublisher) LastError added in v1.3.99 

func (p *VisorBWCXOPublisher) LastError() error

LastError is the most recent publish failure, if any.

func (*VisorBWCXOPublisher) Publisher added in v1.3.99 

func (p *VisorBWCXOPublisher) Publisher() *treestore.Publisher

Publisher returns the underlying treestore publisher.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.