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

transport

package
v1.3.103 Latest Latest
Warning

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

 Go to latest Published: Oct 24, 2026 License: AGPL-3.0 Imports: 37 Imported by: 5

Documentation

Overview

Package transport pkg/transport/deferred_discovery.go c2-net-transport

DeferredDiscoveryClient lets a visor finish booting while the transport discovery is unreachable.

Constructing the real TPD client performs network I/O — an httpauth nonce fetch — so it fails outright when the TPD cannot be reached. The visor used to retry that construction, unbounded, on the boot path. The transport module gates the visor's RPC server, its launcher (and therefore dmsgpty, the hypervisor RPC and every app listener) and its router, so a visor that could not reach the TPD never finished starting: alive, holding dmsg sessions, answering nothing, and unmanageable for as long as the TPD stayed away.

That is a cascade, not just an outage. The deployment services are reached over dmsg, so a visor co-located with them cannot register its transports until they answer — and if the TPD is itself behind a visor that is stuck in the same wait, neither ever proceeds. Observed in production: one host's visor wedged, which took the transport discovery with it, and every visor that restarted afterwards wedged the same way on the missing TPD, spreading with the rollout.

A visor's transports are useful before they are registered, and registration is already retried on a timer, so the TPD is not a boot prerequisite. This wrapper makes that explicit: boot proceeds with a client whose calls fail cleanly until the real one connects in the background, after which every call is forwarded to it.

Package transport pkg/transport/dial_once.go c2-net-transport

Package transport pkg/transport/disc_notfound_native.go c2-net-transport

Package transport pkg/transport/discovery.go c2-net-transport

Package transport pkg/transport/entry.go c2-net-transport

Package transport pkg/transport/errors.go c2-net-transport

Package transport pkg/transport/handshake.go c2-net-transport

Package transport pkg/transport/lock_watchdog.go c2-net-transport

Package transport pkg/transport/log.go c2-net-transport

Per-transport bandwidth counters that the manager exposes to the rest of the visor. The store is in-memory only — historical (per-day) bandwidth is served from pkg/visor/stats's bbolt-backed Tracker, not from this store. See pkg/visor/stats/tracker.go for the daily rollups fed into the visor's CXO publisher and the /stats HTTP endpoints.

Package transport pkg/transport/managed_transport.go c2-net-transport

Package transport pkg/transport/manager.go c2-net-transport

Package transport pkg/transport/manager_ar_native.go c2-net-transport

Package transport pkg/transport/manager_inbound.go c2-net-transport

Package transport pkg/transport/signed_list.go c2-net-transport

Package transport pkg/transport/stream_digest.go c4-net-transport

Package transport pkg/transport/testing_helpers.go c2-net-transport

Package transport pkg/transport/transport.go c2-net-transport operations.

Package transport pkg/transport/vstream.go c2-net-transport

VStream provides virtual bidirectional streams over route ID 0 transport packets. Used by the cascade RPC relay and DHT transport layer to multiplex multiple logical connections over a single managed transport.

Wire format per packet: [streamID:8][senderPK:33][flags:1][data...] Flags: 0x00=data, 0x01=SYN (new stream), 0x02=FIN (close)

A visor also acts as a dmsg-server-style RELAY over its skynet transports: a SYN carrying the Relay flag names a third-party destination PK and is signed by the originator. If the destination is not this visor and a direct transport to it exists, the SYN is forwarded (byte-spliced) to that peer — PK-addressed forwarding with no route, mirroring the dmsg server's forwardViaPeer/bridgeStream. Relaying is always on and bounded (1-hop guard via the Relayed flag + a concurrent-relay cap); it is not configurable off. See pkg/transport/vstream_relay.go and docs/design/dmsg-bootstrap-floor.md.

Relay SYN wire format: 

[streamID:8][senderPK:33][flags:1][dstPK:33][originID:8][sig:65]

where sig = sign_senderSK(originID || senderPK || dstPK). originID is the originator's own stream id, preserved end-to-end (the relay remaps streamID but not originID) so the signature verifies at the destination.

Package transport pkg/transport/vstream_relay.go c2-net-transport

Visor-as-relay for VStreamMux: a visor forwards a signed, PK-addressed SYN to a third party it has a direct (non-dmsg) transport to, then byte-splices the two legs — the skynet analog of the dmsg server's forwardViaPeer/bridgeStream. Always on, bounded, not configurable off.

Index

Examples

Constants

View Source 
const (
	VStreamHeaderSize = 8 + 33 + 1 // streamID + senderPK + flags

	VStreamFlagData = 0x00
	VStreamFlagSyn  = 0x01
	VStreamFlagFin  = 0x02
	// VStreamFlagRelay marks a SYN destined for a third-party PK carried in
	// the extended relay header; the relay forwards it PK-addressed.
	VStreamFlagRelay = 0x04
	// VStreamFlagRelayed is set by a relay on the leg it emits, so the next
	// hop refuses to forward again (1-hop guard, mirrors dmsg's ss.isPeer).
	VStreamFlagRelayed = 0x08

	// DefaultMaxRelayedVStreams bounds concurrent relayed streams a visor
	// carries on behalf of others, so an always-on relay can't be amplified.
	DefaultMaxRelayedVStreams = 4096
)

VStream header constants.

View Source 
const CheckpointBase = 16 << 20

CheckpointBase is the spacing in bytes of the finest checkpoint level. Level m is every CheckpointBase<<(4*m) bytes: 16 MiB, 256 MiB, 4 GiB, 64 GiB.

View Source 
const CheckpointLevels = 4

CheckpointLevels is how many checkpoint levels a digest keeps.

View Source 
const CheckpointRecent = 8

CheckpointRecent is how many of the finest checkpoints a digest keeps, so two edges that sample a little apart on a busy link still share one.

View Source 
const DialFailureRingSize = 512

DialFailureRingSize is how many failed dials the manager keeps. They have their own ring so a burst of failures cannot push opens and closes out.

View Source 
const DmsgRelayFallback = true

DmsgRelayFallback is whether the visor may mint a DMSG-type transport when every direct type failed (EnsureBestTransport and the route-setup hook). A native visor keeps it: behind a NAT that defeats stcpr/sudph, the relayed data plane is what keeps its routes alive. See dmsg_fallback_js.go for why a browser visor does not.

View Source 
const MaxReasonableRTTMs = 30_000.0

MaxReasonableRTTMs is the upper bound for an accepted latency sample. handleTransportPong has no in-flight tracking, so a pong that finally arrives ~30s after its ping (delayed packet, suspended host, NIC queue stall) yields rttMs in the tens of seconds. Min/Avg shrug it off but Max only ever grows, so a single bad sample pins the transport's Max forever (or until the lat:<id> key TTLs out 35 days later). Real intercontinental RTT plus heavy queueing is comfortably under 5s; 30s is a generous floor for "obviously bogus".

Exported so router-side latency producers (RouteGroup.MeasureLatency, RouteGroup.handlePongPacket) can apply the same threshold.

View Source 
const TransportEventRingSize = 2048

TransportEventRingSize is how many events the manager keeps. A public visor is a magnet for autoconnect: on one exit visor 256 entries spanned under three minutes (130 stcpr opens, 55 squicr, 41 sudph, 28 webrtc, 2 dmsg), so the one close that was being chased had already rolled out of the ring by the time anyone read it. 2048 entries cost a few hundred KiB and buy roughly half an hour on that same visor.

View Source 
const TransportLastCloseMax = 1024

TransportLastCloseMax is how many transports the manager remembers the last close of. The ring answers "what has been happening"; this answers "why did transport X die", and it has to survive any amount of open churn — opens never evict a close here, only closes of 1024 other transports do.

Variables

View Source 
var ErrDiscoveryNotReady = errors.New("transport discovery not connected yet")

ErrDiscoveryNotReady is returned by a DeferredDiscoveryClient that has not connected yet. It is deliberately distinct from a transport error: the call did not fail on the wire, it never left.

View Source 
var (
	// ErrEdgeIndexNotFound is returned when no edge index was found.
	ErrEdgeIndexNotFound = errors.New("edge index not found")
)
View Source 
var ErrNotFound = errors.New("transport not found")

ErrNotFound is returned when requested transport is not found

View Source 
var (
	// ErrNotServing is the error returned when a transport is no longer served.
	ErrNotServing = errors.New("transport is no longer being served")
)
View Source 
var ErrSignedListSig = errors.New("transport list: signature does not match its owner")

ErrSignedListSig is returned when a transport list's signature does not match its owner's key.

View Source 
var ErrTransportCreateDisabled = errors.New("transport creation disabled for this type by policy")

ErrTransportCreateDisabled is returned when creating (dialing out) a transport of a given type is forbidden by the manager's transport-creation policy (NoDirectTransports / TransportCreateDeny). It gates only OUTBOUND creation; inbound (accepted) transports and reachability over existing transports/routes are unaffected.

View Source 
var ErrUnknownNetwork = errors.New("unknown network type")

ErrUnknownNetwork occurs on attempt to use an unknown network type.

View Source 
var ErrWrongNetwork = errors.New("wrong network")

ErrWrongNetwork is returned if connection's network differs from the one transport is awaiting.

Functions

func MakeTransportID 

func MakeTransportID(keyA, keyB cipher.PubKey, netType types.Type) uuid.UUID

MakeTransportID generates uuid.UUID from pair of keys + type + public Generated uuid is: - always the same for a given pair - GenTransportUUID(keyA,keyB) == GenTransportUUID(keyB, keyA)

func SortEdges 

func SortEdges(keyA, keyB cipher.PubKey) [2]cipher.PubKey

SortEdges sorts keys so that least-significant comes first

func TypeFromTransportID 

func TypeFromTransportID(tpID uuid.UUID, keyA, keyB cipher.PubKey) types.Type

TypeFromTransportID determines the transport type by comparing the given transport ID against computed IDs for known transport types. A TPID is a one-way SHA-256 of (sorted keys ‖ type string) — the type is not stored in a recoverable field — so the only way to recover it is to replay MakeTransportID for each candidate type (with the endpoint keys) and match. Returns empty string if no match is found.

The candidate set is types.Known() (the canonical single source of truth, so it never drifts as new transport types are added — the reason the old hard-coded {STCPR,SUDPH,STCP,DMSG} list silently returned "" for squicr / webrtc / swsr / swtr) plus the legacy pre-rename wire names, since a transport an older visor registered under "quic"/"ws"/"wt" was hashed with that string and so has a different TPID than its canonical-named equivalent.

Types

type BandwidthData 

type BandwidthData struct {
	SentBytes uint64 `json:"sent_bytes"` // Total bytes sent (cumulative)
	RecvBytes uint64 `json:"recv_bytes"` // Total bytes received (cumulative)
}

BandwidthData represents cumulative bandwidth for a transport

type Checkpoint added in v1.3.101 

type Checkpoint struct {
	Offset uint64
	Digest uint64
}

Checkpoint is the digest of a connection's byte stream from its start up to Offset. Both ends of a reliable connection see the same stream, so equal checkpoints from the two edges prove those bytes passed between them.

type CompactEntry 

type CompactEntry struct {
	Remote cipher.PubKey `json:"r"`
	Type   types.Type    `json:"t"`
	Label  Label         `json:"l,omitempty"` // omitted == LabelAutomatic
}

CompactEntry is the space-optimized wire form of a transport in the transports/list CXO discovery snapshot. The full Entry repeats the reporter's own PK in Edges on EVERY row (a 66-hex-char PK duplicated once per transport) and carries an ID that is a deterministic hash of (edges, type). TPD already knows the reporter and can recompute the ID, so the snapshot ships only the remote edge + type — ~62% smaller on a real transport set. Reconstruct a full Entry with EntryFromCompact(reporter, ce). Label is omitted for the common LabelAutomatic case and reconstructed as such (lossless).

type DeferredDiscoveryClient 

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

DeferredDiscoveryClient is a DiscoveryClient that forwards to the real one once it connects, and fails fast before that.

func NewDeferredDiscoveryClient 

func NewDeferredDiscoveryClient(ctx context.Context, log *logging.Logger, interval time.Duration, connect DiscoveryConnectFunc) *DeferredDiscoveryClient

NewDeferredDiscoveryClient returns a client that is not connected yet and a goroutine that keeps trying to connect until ctx ends. interval bounds how often connection is attempted.

func (*DeferredDiscoveryClient) Connected 

func (d *DeferredDiscoveryClient) Connected() bool

Connected reports whether the real client is in place.

func (*DeferredDiscoveryClient) DeleteTransport 

func (d *DeferredDiscoveryClient) DeleteTransport(ctx context.Context, id uuid.UUID) error

DeleteTransport implements DiscoveryClient.

func (*DeferredDiscoveryClient) DeleteTransports 

func (d *DeferredDiscoveryClient) DeleteTransports(ctx context.Context, ids []uuid.UUID) (int, error)

DeleteTransports implements DiscoveryClient.

func (*DeferredDiscoveryClient) GetAllTransports 

func (d *DeferredDiscoveryClient) GetAllTransports(ctx context.Context) ([]*Entry, error)

GetAllTransports implements DiscoveryClient.

func (*DeferredDiscoveryClient) GetAllTransportsPerKeyStats 

func (d *DeferredDiscoveryClient) GetAllTransportsPerKeyStats(ctx context.Context) (PerKeyStats, error)

GetAllTransportsPerKeyStats implements DiscoveryClient.

func (*DeferredDiscoveryClient) GetAllTransportsStats 

func (d *DeferredDiscoveryClient) GetAllTransportsStats(ctx context.Context) (*NetworkTransportStats, error)

GetAllTransportsStats implements DiscoveryClient.

func (*DeferredDiscoveryClient) GetTransportByID 

func (d *DeferredDiscoveryClient) GetTransportByID(ctx context.Context, id uuid.UUID) (*Entry, error)

GetTransportByID implements DiscoveryClient.

func (*DeferredDiscoveryClient) GetTransportStats 

func (d *DeferredDiscoveryClient) GetTransportStats(ctx context.Context, pk cipher.PubKey) (*TransportStats, error)

GetTransportStats implements DiscoveryClient.

func (*DeferredDiscoveryClient) GetTransportsByEdge 

func (d *DeferredDiscoveryClient) GetTransportsByEdge(ctx context.Context, pk cipher.PubKey) ([]*Entry, error)

GetTransportsByEdge implements DiscoveryClient.

func (*DeferredDiscoveryClient) Ready 

func (d *DeferredDiscoveryClient) Ready() <-chan struct{}

Ready is closed once the real client is connected.

func (*DeferredDiscoveryClient) RegisterTransports 

func (d *DeferredDiscoveryClient) RegisterTransports(ctx context.Context, entries ...*SignedEntry) error

RegisterTransports implements DiscoveryClient.

func (*DeferredDiscoveryClient) RegisterTransportsV3 

func (d *DeferredDiscoveryClient) RegisterTransportsV3(ctx context.Context, version string, entries ...*Entry) error

RegisterTransportsV3 implements DiscoveryClient.

type DialFailure 

type DialFailure struct {
	At     time.Time     `json:"at"`
	Type   string        `json:"type"`
	Remote cipher.PubKey `json:"remote"`
	Label  Label         `json:"label"`
	Error  string        `json:"error"`
	// DurS is how long the dial ran before it failed, which tells a timeout
	// from a refusal.
	DurS float64 `json:"dur_s"`
}

DialFailure is one outbound transport dial that failed. The events ring only holds transports that opened, so a dial that never completed left nothing behind but a debug log line.

type DialStat 

type DialStat struct {
	Attempts    uint64    `json:"attempts"`
	OK          uint64    `json:"ok"`
	Failed      uint64    `json:"failed"`
	LastError   string    `json:"last_error,omitempty"`
	LastErrorAt time.Time `json:"last_error_at,omitzero"`
}

DialStat counts the outbound dials of one transport type since start.

type DigestState added in v1.3.101 

type DigestState struct {
	// Epoch names the connection: its digest at the first checkpoint, the
	// same on both edges. Zero until the stream reaches CheckpointBase.
	Epoch uint64
	// Recent holds the latest finest checkpoints, oldest first, and Coarse the
	// latest of each coarser level. Zero where the stream has not reached one.
	Recent [CheckpointRecent]Checkpoint
	Coarse [CheckpointLevels - 1]Checkpoint
}

DigestState is one direction of a transport's current connection.

type DirectDialHookFn 

type DirectDialHookFn func(remotePK cipher.PubKey, transportKind, appName string) error

DirectDialHookFn is the policy hook fired before a VStreamMux direct dial. Returns a non-nil error to refuse the dial (the error surfaces as the Dial return). Pass through a non-error nil return to allow. nil hook = no policy check.

appName is the originating app's name (skysocks-client, vpn-client, etc.) when the caller supplied one — empty when not threaded (older code paths). The routing-policy bridge uses it to look up per-app policy rules so a script can branch on ctx.app for direct dials just as it does for overlay dials.

Lives as a func type rather than an interface so the transport package doesn't take a dependency on the router/policy package — the visor wires this with a closure that bridges into the existing routing-policy stack.

type DiscoveryClient 

type DiscoveryClient interface {
	RegisterTransports(ctx context.Context, entries ...*SignedEntry) error
	// RegisterTransportsV3 registers transports using the DHT-aligned wire
	// format: bare entries with no per-entry signatures (auth is the
	// outer SW-Sig request signature). Version travels in a header.
	// Implementations should fall back to RegisterTransports when the
	// TPD returns 404 so new clients interoperate with older servers.
	RegisterTransportsV3(ctx context.Context, version string, entries ...*Entry) error
	GetTransportByID(ctx context.Context, id uuid.UUID) (*Entry, error)
	GetTransportsByEdge(ctx context.Context, pk cipher.PubKey) ([]*Entry, error)
	GetAllTransports(ctx context.Context) ([]*Entry, error)
	GetTransportStats(ctx context.Context, pk cipher.PubKey) (*TransportStats, error)
	GetAllTransportsStats(ctx context.Context) (*NetworkTransportStats, error)
	GetAllTransportsPerKeyStats(ctx context.Context) (PerKeyStats, error)
	DeleteTransport(ctx context.Context, id uuid.UUID) error
	// DeleteTransports deletes multiple transports in a single request.
	// Returns (deleted count, error). Falls back to sequential deletion if batch endpoint unavailable.
	DeleteTransports(ctx context.Context, ids []uuid.UUID) (int, error)
}

DiscoveryClient performs Transport discovery operations.

func NewDiscoveryMock 

func NewDiscoveryMock() DiscoveryClient

NewDiscoveryMock construct a new mock transport discovery client.

func NewNoopDiscoveryClient 

func NewNoopDiscoveryClient() DiscoveryClient

NewNoopDiscoveryClient constructs a new no-op transport discovery client.

type DiscoveryConnectFunc 

type DiscoveryConnectFunc func(ctx context.Context) (DiscoveryClient, error)

DiscoveryConnectFunc builds the real discovery client. It performs network I/O and returns an error while the discovery is unreachable.

type Entry 

type Entry struct {

	// ID is the Transport ID that uniquely identifies the Transport.
	ID uuid.UUID `json:"t_id"`

	// Edges contains the public keys of the Transport's edge nodes
	// Edges should always be sorted in ascending order
	Edges [2]cipher.PubKey `json:"edges"`

	// Type represents the transport type.
	Type types.Type `json:"type"`

	Label Label `json:"label"`

	// Latency is the inter-visor round-trip time in milliseconds, measured via ping/pong
	Latency float64 `json:"latency_ms,omitempty"`

	// Bandwidth is total bytes (sent + recv), not included in ToBinary/signatures
	Bandwidth uint64 `json:"bandwidth,omitempty"`

	// ThroughputBps is the PASSIVELY-observed peak goodput (sent+recv bytes/sec)
	// this transport has achieved carrying real traffic — a lower-bound CAPACITY
	// estimate, distinct from Latency (RTT, which is throughput-blind: a webrtc
	// SCTP datachannel has low RTT but poor goodput). 0 until enough traffic has
	// flowed to sample. Lets route ranking prefer transports MEASURED to be fast
	// rather than guessing by type. Not part of ToBinary/signatures.
	ThroughputBps float64 `json:"throughput_bps,omitempty"`
}

Entry is the unsigned representation of a Transport.

func EntryFromCompact 

func EntryFromCompact(reporter cipher.PubKey, ce CompactEntry) *Entry

EntryFromCompact reconstructs the full Entry a publisher would have sent, from a compact row and the reporter PK that published it. Edges are canonicalized (sorted) and the ID recomputed by MakeEntry, exactly matching the full-form Entry.

func MakeEntry 

func MakeEntry(aPK, bPK cipher.PubKey, netType types.Type, label Label) Entry

MakeEntry creates a new transport entry

func Routable 

func Routable(entries []*Entry) []*Entry

Routable returns entries without the local links a host visor publishes to the services it runs, which no route may use. It copies only when it drops one.

func (*Entry) EdgeIndex 

func (e *Entry) EdgeIndex(pk cipher.PubKey) int

EdgeIndex returns the index location of the given public key. Returns -1 if the edge is not found.

func (*Entry) HasEdge 

func (e *Entry) HasEdge(edge cipher.PubKey) bool

HasEdge returns true if the provided edge is present in 'e.Edges' field.

func (*Entry) IsLeastSignificantEdge 

func (e *Entry) IsLeastSignificantEdge(pk cipher.PubKey) bool

IsLeastSignificantEdge returns true if given pk is least significant edge of this entry

func (*Entry) RemoteEdge 

func (e *Entry) RemoteEdge(local cipher.PubKey) cipher.PubKey

RemoteEdge returns the remote edge's public key.

func (*Entry) Signature 

func (e *Entry) Signature(secKey cipher.SecKey) (cipher.Sig, error)

Signature returns signature for Entry calculated from binary representation.

func (*Entry) String 

func (e *Entry) String() string

String implements stringer

func (*Entry) ToBinary 

func (e *Entry) ToBinary() []byte

ToBinary returns binary representation of an Entry

func (*Entry) ToCompact 

func (e *Entry) ToCompact(local cipher.PubKey) CompactEntry

ToCompact returns the compact form of e relative to local (the reporting/publishing visor's PK): the remote edge + type, and the label only when it isn't the default LabelAutomatic.

type Label 

type Label string

Label is a part of transport entry that signifies the origin of this entry 

const (
	// LabelUser signifies a user-created transport entry
	LabelUser Label = "user"
	// LabelAutomatic are transports to publically advertised visors
	LabelAutomatic Label = "automatic"
	// LabelSkycoin are transports created by skycoin system to improve network resiliency
	LabelSkycoin Label = "skycoin"
	// LabelSetup signifies a transport to a Route Setup Node.
	// These transports carry only control-plane traffic and are excluded
	// from data route path calculations by the route finder.
	LabelSetup Label = "setup"

	// LabelSysFeed marks a local link to a service that takes its registrations
	// off the system feed of the transport discovery on the same host.
	LabelSysFeed Label = "sysfeed"
)

type LatencyData 

type LatencyData struct {
	Min int64 `json:"min"` // Minimum observed latency in microseconds
	Max int64 `json:"max"` // Maximum observed latency in microseconds
	Avg int64 `json:"avg"` // Average latency in microseconds
}

LatencyData represents latency statistics for a transport in microseconds

type LatencyStats 

type LatencyStats struct {
	Min float64 `json:"min_ms"` // Minimum observed latency in milliseconds
	Max float64 `json:"max_ms"` // Maximum observed latency in milliseconds
	Avg float64 `json:"avg_ms"` // Average latency in milliseconds
}

LatencyStats holds latency measurement statistics for a transport.

type LogEntry 

type LogEntry struct {
	// atomic requires 64-bit alignment for struct field access
	RecvBytes *uint64 `csv:"recv"` // Total received bytes.
	SentBytes *uint64 `csv:"sent"` // Total sent bytes.
}

LogEntry represents a logging entry for a given Transport. The entry is updated every time a packet is received or sent.

func MakeLogEntry 

func MakeLogEntry(ls LogStore, tpID uuid.UUID, log *logging.Logger) *LogEntry

MakeLogEntry makes a new LogEntry by adding the info from old entry if found

func NewLogEntry 

func NewLogEntry() *LogEntry

NewLogEntry creates a new LogEntry

func (*LogEntry) AddRecv 

func (le *LogEntry) AddRecv(n uint64)

AddRecv records read.

func (*LogEntry) AddSent 

func (le *LogEntry) AddSent(n uint64)

AddSent records write.

func (*LogEntry) GobDecode 

func (le *LogEntry) GobDecode(b []byte) error

GobDecode implements gob.GobDecoder

func (*LogEntry) GobEncode 

func (le *LogEntry) GobEncode() ([]byte, error)

GobEncode implements gob.GobEncoder

func (*LogEntry) MarshalJSON 

func (le *LogEntry) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaller

func (*LogEntry) Reset 

func (le *LogEntry) Reset()

Reset resets LogEntry. Uses atomic stores rather than subtracting a non-atomically-read current value, which could race with a concurrent AddSent/AddRecv and under/over-shoot (or wrap uint64).

type LogStore 

type LogStore interface {
	Entry(id uuid.UUID) (*LogEntry, error)
	Record(id uuid.UUID, entry *LogEntry) error
}

LogStore stores transport log entries. The only implementation is the in-memory store — historical persistence is the stats Tracker's job (pkg/visor/stats), not this store's. The store exists so that a transport that closes and re-opens within the same visor session preserves its cumulative byte counters across the gap; nothing in this package persists across visor restarts.

func InMemoryTransportLogStore 

func InMemoryTransportLogStore() LogStore

InMemoryTransportLogStore implements in-memory TransportLogStore.

type ManagedTransport 

type ManagedTransport struct {
	Entry    Entry
	LogEntry *LogEntry
	// contains filtered or unexported fields
}

ManagedTransport manages a direct line of communication between two visor nodes. There is a single underlying transport between two edges. Initial dialing can be requested by either edge of the transport.

func NewManagedTransport 

func NewManagedTransport(conf ManagedTransportConfig) *ManagedTransport

NewManagedTransport creates a new ManagedTransport.

func NewManagedTransportForTest 

func NewManagedTransportForTest(conn network.Transport) *ManagedTransport

NewManagedTransportForTest creates a minimal ManagedTransport for testing. Only the transport field and basic channels are initialized. Uses a no-op queueDeletion to avoid nil pointer on close.

func (*ManagedTransport) Accept 

func (mt *ManagedTransport) Accept(ctx context.Context, transport network.Transport) error

Accept accepts a new underlying transport.

func (*ManagedTransport) Attached added in v1.3.99 

func (mt *ManagedTransport) Attached() bool

Attached reports whether a desk served by this visor's hypervisor opened the transport through /tp/ws, as against a peer dialing the transport port.

func (*ManagedTransport) Checkpoints added in v1.3.101 

func (mt *ManagedTransport) Checkpoints() (sent, recv DigestState)

Checkpoints returns the stream checkpoints of the transport's current conn, for what it sent and what it received.

func (*ManagedTransport) Close 

func (mt *ManagedTransport) Close() (err error)

Close implements io.Closer It also waits for transport to stop serving before it returns. It only returns an error if transport status update fails.

func (*ManagedTransport) CloseForTest 

func (mt *ManagedTransport) CloseForTest()

CloseForTest closes the ManagedTransport's done channel, marking it as closed. This is for tests that need to simulate a closed transport without going through the full close() path which requires a discovery client.

func (*ManagedTransport) ConnDetails 

func (mt *ManagedTransport) ConnDetails() *network.ConnDetails

ConnDetails returns the underlying network.Transport's curated, secrets-free connection metadata (direct remote/local addr, dmsg relay server PK, QUIC TLS identity, webrtc ICE candidate), or nil when the transport isn't serving or its type exposes no details. The visor's TransportSummary carries this onto `cli visor state` / `cli tp`. No secret key is ever included.

func (*ManagedTransport) Dial 

func (mt *ManagedTransport) Dial(ctx context.Context) error

Dial dials a new underlying transport.

func (*ManagedTransport) DialAsync 

func (mt *ManagedTransport) DialAsync(ctx context.Context, errCh chan error)

DialAsync is asynchronous version of dial that allows dialing in a different goroutine

func (*ManagedTransport) GetBandwidth 

func (mt *ManagedTransport) GetBandwidth() *BandwidthData

GetBandwidth returns the current cumulative bandwidth for this transport.

func (*ManagedTransport) GetLatency 

func (mt *ManagedTransport) GetLatency() float64

GetLatency returns the average inter-visor ping latency in milliseconds. For backwards compatibility, returns the average latency.

func (*ManagedTransport) GetLatencyStats 

func (mt *ManagedTransport) GetLatencyStats() LatencyStats

GetLatencyStats returns the full latency statistics (min/max/avg).

func (*ManagedTransport) GetThroughputBps 

func (mt *ManagedTransport) GetThroughputBps() float64

GetThroughputBps returns the passively-observed peak goodput estimate in bytes/sec (0 if the transport hasn't carried enough traffic to sample yet).

func (*ManagedTransport) Handlers 

func (mt *ManagedTransport) Handlers() []string

Handlers names the route-ID-0 packet handlers wired on this transport. A transport missing one delivers that packet type to the router instead, where it is dropped as unknown — the #4725 failure mode, visible here per transport.

func (*ManagedTransport) IsClosed 

func (mt *ManagedTransport) IsClosed() bool

IsClosed returns true when the transport is closed This instance cannot be used anymore and should be discarded

func (*ManagedTransport) IsInitiator 

func (mt *ManagedTransport) IsInitiator() bool

IsInitiator reports whether this visor originated the transport (dialed out, "outgoing") rather than accepting an inbound dial ("incoming").

func (*ManagedTransport) LastRecvAt 

func (mt *ManagedTransport) LastRecvAt() time.Time

LastRecvAt is when the read loop last received any packet (zero = never).

func (*ManagedTransport) MalformedFrames 

func (mt *ManagedTransport) MalformedFrames() int64

MalformedFrames counts received frames whose type byte is outside the known range — the peer's framing is off. For `visor state`.

func (*ManagedTransport) MissedPongs 

func (mt *ManagedTransport) MissedPongs() int64

MissedPongs is the number of transport pings sent since the last pong (pongMissThreshold closes the transport). For `visor state`.

func (*ManagedTransport) PongSeen 

func (mt *ManagedTransport) PongSeen() bool

PongSeen reports whether the peer has ever answered a transport ping.

func (*ManagedTransport) ProbeLatency 

func (mt *ManagedTransport) ProbeLatency(ctx context.Context) (float64, error)

ProbeLatency actively measures this transport's RTT: it sends one transport-level ping and waits for the pong to land, bounded by ctx. It returns the milliseconds the new sample recorded.

Why it exists: the periodic ping (transportPingInterval) only samples a transport that is READY, so a transport this visor holds but has never carried traffic over can sit with no measurement at all — which is exactly the state a multi-tunnel dial finds its candidate first hops in when it must choose one (measured 2026-09-17: every candidate intermediate reported "unmeasured" at dial time and the ranking had nothing to rank on). One round trip on demand fills that gap.

It returns the transport's existing sample immediately when it already has one, so callers can probe a whole candidate set unconditionally. The error path is "no measurement obtained" — ctx expired, or the transport is not serving; callers treat that exactly as unmeasured, never as a dial failure.

func (*ManagedTransport) Remote 

func (mt *ManagedTransport) Remote() cipher.PubKey

Remote returns the remote public key.

func (*ManagedTransport) RemoteIP 

func (mt *ManagedTransport) RemoteIP() string

RemoteIP returns the host portion of the underlying transport's remote raw address (e.g. "1.2.3.4" for an stcpr transport to 1.2.3.4:5000). Returns "" when the transport isn't serving, or the underlying type doesn't have an IP-shaped remote (notably dmsg, which relays through a server rather than directly to the peer). Used by the routing-policy provider's Geo() lookup so an operator's policy can branch on `geo.country(pk) == "ID"`.

func (*ManagedTransport) SameNetwork added in v1.3.99 

func (tp *ManagedTransport) SameNetwork(self net.IP, localIPs []net.IP) bool

SameNetwork reports whether tp's peer is on this visor's own network: the transport reaches it at a private address, or at this visor's own public IP (self, nil when not yet known) or one of its interface addresses. Bytes on such a transport move between two visors on one IP, which the reward system does not pay for. dmsg transports never qualify: a dmsg peer's address is deliberately unknown.

Stricter than SameLANPeers, which also counts a shared /24 because for routing a neighboring host is the same failure domain. Two visors in one /24 can belong to different operators, so for rewards only the same IP counts.

func (*ManagedTransport) Serve 

func (mt *ManagedTransport) Serve(readCh chan<- routing.Packet)

Serve serves and manages the transport.

Goroutine layout: the Serve goroutine runs readLoop itself, then waits on the done channel until close fires. The previous per-transport logLoop and pingLoop tickers have been centralized into Manager.runTransportMaintenance — on a hub visor with hundreds of automatic transports those two per-transport goroutines (plus their tickers, each waking the Go scheduler every 3s and 30s respectively) were the dominant runtime-scheduler cost even on an otherwise idle visor. The central maintenance loop walks all transports under one ticker apiece, dropping ~2*N goroutines from steady state.

func (*ManagedTransport) SetLatency 

func (mt *ManagedTransport) SetLatency(latencyMs float64)

SetLatency sets the average inter-visor ping latency in milliseconds. For backwards compatibility; prefer SetLatencyStats for full statistics.

Samples outside (0, MaxReasonableRTTMs] are dropped:

  • rttMs <= 0 happens with ns-resolution timestamps on sub-µs loopback; a 0 there clobbers Avg and (via the `latencyMs < Min` branch) Min.
  • rttMs > MaxReasonableRTTMs happens when handleTransportPong receives a long-delayed pong; that 30+ second value would pin Max indefinitely.

func (*ManagedTransport) SetLatencyStats 

func (mt *ManagedTransport) SetLatencyStats(stats LatencyStats)

SetLatencyStats sets the full latency statistics. A snapshot with any field outside (0, MaxReasonableRTTMs] is rejected wholesale: a partial measurement (e.g. all five pings timed out except one that returned in <1µs) would otherwise overwrite a previously good record with a zero in min, and a stragglerpong on the upper end would pin Max forever.

func (*ManagedTransport) Start added in v1.3.100 

func (mt *ManagedTransport) Start(readCh chan<- routing.Packet, onEnd func())

Start serves the transport, and calls onEnd once it has ended. A transport whose reads can be pushed gets no goroutine of its own.

func (*ManagedTransport) Type 

func (mt *ManagedTransport) Type() types.Type

Type returns the transport type.

func (*ManagedTransport) WriteDatagram 

func (mt *ManagedTransport) WriteDatagram(ctx context.Context, packet routing.Packet) error

WriteDatagram writes a (faithful-UDP) packet over the transport's native datagram channel when available (QUIC RFC 9221) — no head-of-line blocking, genuine loss tolerance. If the transport has no datagram channel, or the datagram send fails (e.g. the payload exceeds the path MTU), it falls back to the reliable stream so the packet is still delivered. The payload is already end-to-end AEAD-sealed by the DatagramRouteGroup; QUIC's own TLS adds hop-by-hop encryption to the datagram.

func (*ManagedTransport) WritePacket 

func (mt *ManagedTransport) WritePacket(ctx context.Context, packet routing.Packet) error

WritePacket writes a packet to the remote. Respects context cancellation to prevent blocking forever on dead transports.

func (*ManagedTransport) WriteRawPacket 

func (mt *ManagedTransport) WriteRawPacket(packet routing.Packet) error

WriteRawPacket writes a pre-constructed packet to the transport. Used by the cascade handler and VStreamMux to send route-ID-0 traffic (cascade RPC, DHT, VStream / sky_forward_conn direct dials).

Logs the payload bytes via logSent so bandwidth accounting matches WritePacket's behavior. Without this, every direct- dial byte sent by sky_forward_conn (skysocks-client / vpn- client / skynet-client over a direct stcpr/sudph transport) was invisible to the bandwidth counters — only inbound bytes got counted via readPacket, producing the systematic "sent>>recv" or "recv>>sent" asymmetry seen on per-transport metrics for visors that use direct-dial heavily. The recv counter has always counted all packets because readPacket is the only inbound path.

type ManagedTransportConfig 

type ManagedTransportConfig struct {
	DC              DiscoveryClient
	LS              LogStore
	RemotePK        cipher.PubKey
	TransportLabel  Label
	InactiveTimeout time.Duration

	// QueueDeletion, when set, defers TPD deregistration to the manager's batch
	// deletion loop instead of making an HTTP call per transport close.
	QueueDeletion func(id uuid.UUID)
	// IsInitiator records whether this visor originated the transport (dialed
	// out) rather than accepting an inbound dial. Used only for reporting
	// (outgoing vs incoming counts); the transport itself is bidirectional.
	IsInitiator bool
	// contains filtered or unexported fields
}

ManagedTransportConfig is a configuration for managed transport.

type Manager 

type Manager struct {
	Logger *logging.Logger
	Conf   *ManagerConfig
	// contains filtered or unexported fields
}

Manager manages Transports.

func NewManager 

func NewManager(log *logging.Logger, arClient any, ebc any, config *ManagerConfig, factory network.ClientFactory) (*Manager, error)

NewManager creates a Manager with the provided configuration and transport factories. 'factories' should be ordered by preference. arClient must be an addrresolver.APIClient (native builds); ebc an *appevent.Broadcaster. Both are typed `any` to keep addrresolver (net/http) and appevent (net/rpc) out of the TinyGo graph — concrete callers are unaffected since a concrete value satisfies `any`.

func (*Manager) ARClient 

func (tm *Manager) ARClient() addrresolver.APIClient

ARClient returns the address resolver client used by this transport manager. arClient is stored as `any` (see Manager.arClient) to keep addrresolver — and its net/http dependency — out of the TinyGo graph; this native getter recovers the concrete type so existing callers are unchanged.

func (*Manager) CanCreateTransport 

func (tm *Manager) CanCreateTransport(netType types.Type) bool

CanCreateTransport reports whether the manager is permitted to CREATE (dial out) a transport of netType under the current transport-creation policy. It is the exported, affirmative form of transportCreateDenied — callers that decide what to dial (e.g. autoconnect phases) use it to skip types they'd only be refused at the funnel, avoiding wasted dials and log noise.

func (*Manager) Close 

func (tm *Manager) Close()

Close closes opened transports, network clients and all service tasks of transport manager

func (*Manager) DeleteAllTransports 

func (tm *Manager) DeleteAllTransports()

DeleteAllTransports deregisters all Transports in transport discovery and deletes them locally.

func (*Manager) DeleteTransport 

func (tm *Manager) DeleteTransport(id uuid.UUID)

DeleteTransport deregisters the Transport of Transport ID in transport discovery and deletes it locally.

func (*Manager) DialStats 

func (tm *Manager) DialStats() (map[string]DialStat, []DialFailure)

DialStats returns the outbound dial counts per transport type and the recent failed dials, oldest first.

func (*Manager) EnsureBestTransport 

func (tm *Manager) EnsureBestTransport(ctx context.Context, remote cipher.PubKey, skip ...types.Type) error

EnsureBestTransport makes a DIRECT transport to remote if none exists, trying this visor's creatable direct types in preference order (PreferredDirectCreateOrder) and falling back to the DMSG relay ONLY if every direct type fails. Idempotent: returns nil immediately when a direct transport to remote already exists.

This is the shared "no transport type specified" creation policy — the order `tp add`, the route-setup hooks, and the proxy clients should all use — so the data plane prefers a peer-to-peer transport (the best one each host can make) and only uses the relayed dmsg transport when nothing direct is possible.

Reaching the DMSG fallback is a RED FLAG, not a normal state: a dmsg data-plane transport means every direct type failed for a peer that (usually) should have been reachable p2p — a symptom worth investigating (webrtc/STUN misconfig, the peer not accepting the type it advertises, AR/discovery staleness). Hence it logs at WARN, and callers that must never relay app traffic (e.g. proxy exit SELECTION, which can just pick a different, p2p-reachable exit) should prefer skipping the peer over accepting a dmsg fallback.

skip lists types the caller has already attempted (or wants excluded) — the native route-setup hook passes STCPR/SUDPH, which it tries itself with AR- and STUN-aware gating, so EnsureBestTransport only adds the remaining direct types.

func (*Manager) Events 

func (tm *Manager) Events() []TransportEvent

Events returns the manager's transport open/close history, oldest first.

func (*Manager) GetTransport 

func (tm *Manager) GetTransport(remote cipher.PubKey, netType types.Type) (*ManagedTransport, error)

GetTransport gets transport entity to the given remote

func (*Manager) GetTransportByID 

func (tm *Manager) GetTransportByID(tpID uuid.UUID) (*ManagedTransport, error)

GetTransportByID retrieves transport by its ID, if it exists

func (*Manager) GetTransportsByLabel 

func (tm *Manager) GetTransportsByLabel(label Label) []*ManagedTransport

GetTransportsByLabel returns all transports that have given label

func (*Manager) GetTransportsByLabels 

func (tm *Manager) GetTransportsByLabels(labels ...Label) []*ManagedTransport

GetTransportsByLabels returns all transports matching any of the given labels

func (*Manager) HTTPAcceptor 

func (tm *Manager) HTTPAcceptor(netType types.Type) (http.Handler, bool)

HTTPAcceptor returns the network client for netType as an http.Handler when it can take transport handshakes arriving on a foreign HTTP server — the WS client on native builds does (wsClient.ServeHTTP). The hypervisor UI port mounts it so a desk it serves can open a transport at the address it knows.

func (*Manager) HasTPDLeafPublisher 

func (tm *Manager) HasTPDLeafPublisher() bool

HasTPDLeafPublisher reports whether CXO mirroring is installed.

When it is, this visor publishes its FULL transport list as one snapshot leaf and TPD's aggregator reconciles against it — absence means deletion, and a missed update self-heals on the next publish (see publishTPDList). That makes the HTTP publish-then-verify reconciliation redundant in both directions: a stale TPD entry disappears because it is not in the next snapshot, and a dropped registration is re-asserted by it. Callers use this to skip that work and keep it only for visors where CXO publishing is off or unavailable.

func (*Manager) InitClient 

func (tm *Manager) InitClient(ctx context.Context, netType types.Type, port int)

InitClient initilizes a network client.

Safe to call again for a netType that already has one — ReinitiateModule ("stcpr" et al, api_network.go) and the SUDPH retry in init_transport.go both do. The previous client is closed before being replaced.

func (*Manager) InitDmsgClient 

func (tm *Manager) InitDmsgClient(ctx context.Context, dmsgC *dmsg.Client)

InitDmsgClient initilizes the dmsg client and also adds dmsgC to the factory

func (*Manager) InjectTransportForTest 

func (tm *Manager) InjectTransportForTest(mt *ManagedTransport)

InjectTransportForTest inserts mt into the manager's transport map keyed by its Entry.ID. Test-only: it bypasses dialing/discovery so cross-package tests can populate a Manager without a live network.

func (*Manager) IsKnownNetwork 

func (tm *Manager) IsKnownNetwork(netName types.Type) bool

IsKnownNetwork returns true when netName is a known network type that we are able to operate in.

Wrapper around the lockless helper for external callers. Methods that already hold tm.mx (read or write) MUST call isKnownNetworkLocked directly — calling this exported wrapper from inside a held read lock deadlocks against any pending writer (Go's RWMutex prioritizes pending writers to prevent writer starvation, so a recursive RLock attempt blocks when a writer is waiting). Production case that bit us: GetTransport previously called IsKnownNetwork from inside its own RLock; with cleanupTransports or acceptTransport pending the write lock, the recursive read attempt deadlocked, freezing every subsequent transport operation including ServeRPCClient's redial attempts.

func (*Manager) LastCloses 

func (tm *Manager) LastCloses() map[uuid.UUID]TransportEvent

LastCloses returns the last close event of each of the last TransportLastCloseMax transports that closed, keyed by transport id. Unlike Events it is not flooded out by open churn: on a public visor every peer's autoconnect keeps opening transports, and the one close being chased was gone from the ring minutes later.

func (*Manager) LastInbound added in v1.3.98 

func (tm *Manager) LastInbound() map[types.Type]time.Time

LastInbound returns, per transport type, when this visor last accepted an inbound transport of it. Types never accepted are absent.

func (*Manager) Local 

func (tm *Manager) Local() cipher.PubKey

Local returns Manager.config.PubKey

func (*Manager) Networks 

func (tm *Manager) Networks() []types.Type

Networks returns all the network types contained within the TransportManager.

func (*Manager) OnTransportClosed 

func (tm *Manager) OnTransportClosed(fn func(uuid.UUID) int)

OnTransportClosed registers fn to be called with a transport's id when that transport closes, for either reason (closeWith or the manager shutting down). Hooks are fired outside every manager lock, in registration order; fn must not block. Registration is for the life of the manager.

func (*Manager) PreferredDirectCreateOrder 

func (tm *Manager) PreferredDirectCreateOrder() []types.Type

PreferredDirectCreateOrder returns this visor's DIRECT (non-relay) transport types in global preference order (tptypes.PreferenceOrder — STCPR > QUIC > SUDPH > STCP > WEBRTC > WS > WT), filtered to the network clients this manager actually has initialized. It is the host-aware "auto" transport-creation order: a native visor yields e.g. [stcpr, quic, sudph, stcp, webrtc, ws, wt]; a browser visor (no raw TCP/UDP) yields [webrtc, ws, wt]. DMSG is excluded — it is the relay, tried only as a last resort by EnsureBestTransport.

func (*Manager) ReadPacket 

func (tm *Manager) ReadPacket() (routing.Packet, error)

ReadPacket reads data packets from routes.

func (*Manager) ReadQueue 

func (tm *Manager) ReadQueue() (depth, capacity int)

ReadQueue is the depth and capacity of the shared inbound packet queue every transport's read loop feeds and the router drains. A full queue means the router is not keeping up (a route group whose app stopped reading blocks it), and every transport's read loop — pings and pongs included — stalls behind it.

func (*Manager) Ready 

func (tm *Manager) Ready() <-chan struct{}

Ready checks if the transport manager is ready with atleast one transport

func (*Manager) RepublishTPDList 

func (tm *Manager) RepublishTPDList()

RepublishTPDList publishes the transport list now, as after a change.

func (*Manager) SameLANPeers 

func (tm *Manager) SameLANPeers(self net.IP) []cipher.PubKey

SameLANPeers returns the PKs of directly-connected non-DMSG peers whose transport endpoint IP is on the same local network as this visor. Two cases are treated as same-LAN:

  • the peer's endpoint is a private / RFC1918 (non-public) IP — a transport reached over a private address is by definition a LAN neighbor; and
  • the NAT-hairpin case: the peer's endpoint is a PUBLIC IP that matches the visor's own public IP (self) exactly, or shares its /24, or shares the /24 of any local interface IP. Two visors behind one NAT that both registered their public IP reach each other by hairpinning off that IP.

Such peers make poor routing INTERMEDIATES to a remote destination: routing through them adds a hop but no path diversity (same first-mile link, same NAT and failure domain) and can fold the path back onto the source's own network. Callers feed the result into DialOptions.ExcludeIntermediatePKs. self may be nil when the public IP is not yet known (STUN pending) — the private-IP and local-interface /24 checks still apply.

func (*Manager) SaveTransport 

func (tm *Manager) SaveTransport(ctx context.Context, remote cipher.PubKey, netType types.Type, label Label) (*ManagedTransport, error)

SaveTransport begins to attempt to establish data transports to the given 'remote' visor.

func (*Manager) SaveTransportNoRegister 

func (tm *Manager) SaveTransportNoRegister(ctx context.Context, remote cipher.PubKey, netType types.Type, label Label) (*ManagedTransport, error)

SaveTransportNoRegister is like SaveTransport but skips transport discovery registration. This is only valid for user-labeled transports.

func (*Manager) SaveTransportWithOptions 

func (tm *Manager) SaveTransportWithOptions(ctx context.Context, remote cipher.PubKey, netType types.Type, label Label, opts SaveTransportOptions) (*ManagedTransport, error)

SaveTransportWithOptions creates a transport with the given options.

func (*Manager) Serve 

func (tm *Manager) Serve(ctx context.Context)

Serve starts all network clients and starts accepting connections from all those clients Additionally, it runs cleanup and persistent reconnection routines

func (*Manager) SetAppDirectHandler 

func (tm *Manager) SetAppDirectHandler(h func(p routing.Packet, mt *ManagedTransport))

SetAppDirectHandler sets the handler for direct skywire-network app dial packets (route ID 0). Used by the visor to deliver inbound direct-dial streams to the per-app server-side accept loop without route setup.

func (*Manager) SetCascadeHandler 

func (tm *Manager) SetCascadeHandler(h func(p routing.Packet, mt *ManagedTransport))

SetCascadeHandler sets the handler for cascade protocol packets (route ID 0). Called by the router to register its cascade handler.

func (*Manager) SetDHTHandler 

func (tm *Manager) SetDHTHandler(h func(p routing.Packet, mt *ManagedTransport))

SetDHTHandler sets the handler for DHT protocol packets (route ID 0).

func (*Manager) SetHostedEntries 

func (tm *Manager) SetHostedEntries(f func() []*Entry)

SetHostedEntries sets what returns the local links from this visor to the services it hosts, which every transport list publish carries.

func (*Manager) SetPTpsCache 

func (tm *Manager) SetPTpsCache(pTps []PersistentTransports)

SetPTpsCache sets the PersistentTransportsCache

func (*Manager) SetRouteChecker 

func (tm *Manager) SetRouteChecker(rc RouteChecker)

SetRouteChecker sets the callback used to determine if a transport has active routes. When set, transport re-creation is blocked for transports that are currently referenced by routing rules, protecting in-flight route traffic.

func (*Manager) SetSetupRPCHandler 

func (tm *Manager) SetSetupRPCHandler(h func(p routing.Packet, mt *ManagedTransport))

SetSetupRPCHandler sets the handler for RSN RPC relay packets (route ID 0).

func (*Manager) SetSkynetForwardHandler 

func (tm *Manager) SetSkynetForwardHandler(h func(p routing.Packet, mt *ManagedTransport))

SetSkynetForwardHandler sets the handler for skynet forward packets (route ID 0).

func (*Manager) SetTPDLeafPublisher 

func (tm *Manager) SetTPDLeafPublisher(p TPDLeafPublisher)

SetTPDLeafPublisher installs (or clears, if nil) the CXO publisher hook used to mirror transport register / deregister leaves to TPD. Safe to call after Serve has started — the next register tick or flushDeletionQueue picks it up. Calling with nil disables CXO mirroring without affecting the HTTP register path.

func (*Manager) SetVisorRPCHandler 

func (tm *Manager) SetVisorRPCHandler(h func(p routing.Packet, mt *ManagedTransport))

SetVisorRPCHandler sets the handler for visor RPC packets (route ID 0).

func (*Manager) ShouldRegisterAR 

func (tm *Manager) ShouldRegisterAR() bool

ShouldRegisterAR returns false if the AR transport limit is negative (never register). Callers should check this during visor initialization.

func (*Manager) Stcpr 

func (tm *Manager) Stcpr() (network.Client, bool)

Stcpr returns stcpr client

func (*Manager) Transport 

func (tm *Manager) Transport(id uuid.UUID) *ManagedTransport

Transport obtains a Transport via a given Transport ID.

func (*Manager) TransportCount 

func (tm *Manager) TransportCount() int

TransportCount returns the number of active transports.

func (*Manager) TransportRemoteAddrs 

func (tm *Manager) TransportRemoteAddrs() []string

TransportRemoteAddrs gets remote IPs for all non-DMSG transports (stcpr, sudph, stcp, quic, …).

Used only to seed the vpn-client's direct-route exemptions. Every one of these peer IPs MUST bypass the tunnel: routing an established transport's remote IP into the TUN severs that transport — and if the transport carries the vpn's own path to the vpn-server, it severs the tunnel itself (the client stops receiving the ServerHello → NO-CARRIER). Previously this returned STCPR only, so a visor whose path to the server ran over sudph (or any non-stcpr type) had that peer tunneled and the VPN silently failed.

DMSG is deliberately excluded: it relays through dmsg servers and has no single routable peer IP, and its servers are exempted separately via DirectRoutesEnvConfig.DmsgServers.

func (*Manager) WalkTransports 

func (tm *Manager) WalkTransports(walk func(tp *ManagedTransport) bool)

WalkTransports ranges through all transports.

The caller's walk callback is arbitrary code, so it runs OUTSIDE tm.mx over a snapshot — the same pattern recordAllTransportLogs/pingAllTransports use. Running it under RLock was a latent freeze: a callback that panicked (and got recovered upstream — e.g. an HTTP handler's recovery middleware) leaked the read lock, and a slow callback held it, blocking every transport writer (acceptTransport, SaveTransport) and — via RWMutex writer-preference — every subsequent reader. That is the transport-manager wedge: one stuck holder, the whole management surface frozen with no log line.

type ManagerConfig 

type ManagerConfig struct {
	PubKey                    cipher.PubKey
	SecKey                    cipher.SecKey
	DiscoveryClient           DiscoveryClient
	LogStore                  LogStore
	PersistentTransportsCache []PersistentTransports
	PTpsCacheMu               sync.RWMutex
	Version                   string // Visor version for reporting to TPD
	// ARTransportLimit is a static, config-only switch for AR registration:
	//   >= 0 = register with the address resolver (default)
	//   <  0 = never register
	// It is deliberately NOT a runtime load-shed: a running visor must never
	// deregister itself from AR because it hit some transport count — AR
	// discoverability is what lets peers dial it, and dropping it silently
	// breaks reachability. Runtime load-shedding is the public-visor SD drain
	// (PublicVisorConfig.MaxTransports), which pauses SERVICE-DISCOVERY
	// registration on a DISTINCT-PEER count and resumes when it drops — see
	// pkg/app/appdisc PublicVisorUpdater.
	ARTransportLimit int
	// NoDirectTransports, when true, refuses to CREATE (dial out) any DIRECT
	// peer-to-peer transport (stcpr/sudph/stcp/squicr/swsr/swtr/webrtc). DMSG — a
	// relay through a dmsg server, not a direct p2p link — is still created, so
	// the visor keeps its always-available baseline and can still reach peers
	// over existing transports/routes. Gates only OUTBOUND creation; inbound
	// (accepted) transports are unaffected.
	NoDirectTransports bool
	// TransportCreateDeny is an advanced/testing per-type deny list: creating any
	// type listed here is refused, INCLUDING dmsg (so a test can force a
	// pure-direct or fully-offline visor). Applied on top of NoDirectTransports;
	// stored normalized (aliases resolved). Empty (default) denies nothing.
	TransportCreateDeny []types.Type
}

ManagerConfig configures a Manager.

type NetworkTransportStats 

type NetworkTransportStats struct {
	TotalTransports int            `json:"total_transports"`
	ByType          map[string]int `json:"by_type"`
	UniqueVisors    int            `json:"unique_visors"`
}

NetworkTransportStats contains network-wide transport statistics.

type PerKeyStats 

type PerKeyStats map[string]map[string]int

PerKeyStats is a map from PK hex to transport counts by type (includes "total" key). Format: {"pk1": {"total": 15, "stcpr": 1, "sudph": 14}, ...}

type PersistentTransports 

type PersistentTransports = tspec.PersistentTransports

PersistentTransports is the wire-format pinning entry. Aliased from pkg/transport/spec so existing callers writing `transport.PersistentTransports{...}` keep compiling, while the canonical (WASM-clean) definition lives in the spec leaf package.

type RouteChecker 

type RouteChecker func(tpID uuid.UUID) bool

RouteChecker is called before tearing down an existing transport for re-creation. It returns true if any active routing rule references the given transport ID, meaning the transport is actively carrying route traffic and must not be torn down.

type SaveTransportOptions 

type SaveTransportOptions struct {
	NoRegister bool // skip transport discovery registration
}

SaveTransportOptions contains options for transport creation.

type SettlementHS 

type SettlementHS func(ctx context.Context, dc DiscoveryClient, transport network.Transport, sk cipher.SecKey) (Label, error)

SettlementHS represents a settlement handshake. This is the handshake responsible for registering a transport to transport discovery. It returns the label from the remote entry (useful for the responder to adopt the initiator's label).

func MakeSettlementHS 

func MakeSettlementHS(init bool, log *logging.Logger, label Label) SettlementHS

MakeSettlementHS creates a settlement handshake. `init` determines whether the local side is initiating or responding. `label` is the transport label to include in the entry (propagated to the remote end). The handshake logic only REGISTERS the transport, and does not update the status of the transport.

func (SettlementHS) Do 

func (hs SettlementHS) Do(ctx context.Context, dc DiscoveryClient, transport network.Transport, sk cipher.SecKey) (Label, error)

Do performs the settlement handshake.

type SignedEntry 

type SignedEntry struct {
	Entry      *Entry         `json:"entry"`
	Signatures [2]cipher.Sig  `json:"signatures"`
	Registered int64          `json:"registered,omitempty"`
	Latency    *LatencyData   `json:"latency,omitempty"`   // Latency statistics in microseconds
	Bandwidth  *BandwidthData `json:"bandwidth,omitempty"` // Cumulative bandwidth data
	Version    string         `json:"version,omitempty"`   // Visor version reporting this entry
}

SignedEntry holds an Entry and it's associated signatures. The signatures should be ordered as the contained 'Entry.Edges'.

func NewSignedEntry 

func NewSignedEntry(entry *Entry, pk cipher.PubKey, secKey cipher.SecKey) (*SignedEntry, error)

NewSignedEntry creates a SignedEntry with first signature

func (*SignedEntry) Sign 

func (se *SignedEntry) Sign(pk cipher.PubKey, secKey cipher.SecKey) error

Sign sets Signature for a given PubKey in correct position

Example 
package main

import (
	"fmt"

	"github.com/skycoin/skywire/pkg/cipher"
	"github.com/skycoin/skywire/pkg/transport"
)

func main() {
	pkA, skA := cipher.GenerateKeyPair()
	pkB, skB := cipher.GenerateKeyPair()

	entry := transport.MakeEntry(pkA, pkB, "mock", transport.LabelUser)
	sEntry := &transport.SignedEntry{Entry: &entry}

	if sEntry.Signatures[0].Null() && sEntry.Signatures[1].Null() {
		fmt.Println("No signatures set")
	}

	if err := sEntry.Sign(pkA, skA); err != nil {
		fmt.Println("error signing with skA: ", err)
	}

	if (!sEntry.Signatures[0].Null() && sEntry.Signatures[1].Null()) ||
		(!sEntry.Signatures[1].Null() && sEntry.Signatures[0].Null()) {
		fmt.Println("One signature set")
	}

	if err := sEntry.Sign(pkB, skB); err != nil {
		fmt.Println("error signing with skB: ", err)
	}

	if !sEntry.Signatures[0].Null() && !sEntry.Signatures[1].Null() {
		fmt.Println("Both signatures set")
	} else {
		fmt.Printf("sEntry.Signatures:\n%v\n", sEntry.Signatures)
	}

}

Output:
No signatures set
One signature set
Both signatures set

func (*SignedEntry) Signature 

func (se *SignedEntry) Signature(pk cipher.PubKey) (cipher.Sig, error)

Signature gets Signature for a given PubKey from correct position

Example 
package main

import (
	"fmt"

	"github.com/skycoin/skywire/pkg/cipher"
	"github.com/skycoin/skywire/pkg/transport"
)

func main() {
	pkA, skA := cipher.GenerateKeyPair()
	pkB, skB := cipher.GenerateKeyPair()

	entry := transport.MakeEntry(pkA, pkB, "mock", transport.LabelUser)
	sEntry := &transport.SignedEntry{Entry: &entry}

	if err := sEntry.Sign(pkA, skA); err != nil {
		fmt.Println("Error signing sEntry with (pkA,skA)")
	}

	if err := sEntry.Sign(pkB, skB); err != nil {
		fmt.Println("Error signing sEntry with (pkB,skB)")
	}

	idxA := sEntry.Entry.EdgeIndex(pkA)
	idxB := sEntry.Entry.EdgeIndex(pkB)

	sigA, errA := sEntry.Signature(pkA)
	sigB, errB := sEntry.Signature(pkB)

	if errA == nil && sigA == sEntry.Signatures[idxA] {
		fmt.Println("SignatureA got")
	}

	if errB == nil && (sigB == sEntry.Signatures[idxB]) {
		fmt.Println("SignatureB got")
	}

	// Incorrect case
	pkC, _ := cipher.GenerateKeyPair()
	if _, err := sEntry.Signature(pkC); err != nil {
		fmt.Printf("SignatureC got error: %v\n", err)
	}

}

Output:
SignatureA got
SignatureB got
SignatureC got error: edge index not found

type SignedList added in v1.3.100 

type SignedList struct {
	PK      cipher.PubKey  `json:"pk"`
	At      int64          `json:"at"` // unix seconds when it was built
	Entries []CompactEntry `json:"entries"`
	Sig     cipher.Sig     `json:"sig"`
}

SignedList is one visor's own transport list, signed by that visor. A neighbor can cache and pass it on, and a reader can still check it.

func NewSignedList added in v1.3.100 

func NewSignedList(pk cipher.PubKey, sk cipher.SecKey, at int64, entries []*Entry) (*SignedList, error)

NewSignedList builds and signs pk's list from its transport entries.

func (*SignedList) Transports added in v1.3.100 

func (l *SignedList) Transports() []*Entry

Transports reconstructs the full entries.

func (*SignedList) Verify added in v1.3.100 

func (l *SignedList) Verify() error

Verify checks the signature against the list's own PK.

func (*SignedList) Version added in v1.3.100 

func (l *SignedList) Version() string

Version names the transport set. It changes only when the set does, so a reader holding the same version has nothing to fetch.

type TPDLeafPublisher 

type TPDLeafPublisher interface {
	Put(path string, value []byte) error
	Delete(path string) error
}

TPDLeafPublisher is the contract the transport manager uses to mirror register / deregister events into a CXO publisher tree consumed by TPD's aggregator. *treestore.Publisher already satisfies this; declaring it as an interface keeps the transport package free of a cxo/treestore dependency. Wired by pkg/visor/init_stats.go after the publisher is built.

type TransportEvent 

type TransportEvent struct {
	At        time.Time     `json:"at"`
	Event     string        `json:"event"` // "open" or "close"
	ID        uuid.UUID     `json:"id"`
	Type      string        `json:"type"`
	Remote    cipher.PubKey `json:"remote"`
	Label     Label         `json:"label"`
	Initiator bool          `json:"initiator"`
	// Reason is set on a close: what decided it, in the words of the code that
	// did ("read: EOF", "no pong for 3 consecutive pings", "deleted by
	// request", …). Empty on an open.
	Reason string `json:"reason,omitempty"`
	// AgeS is how long the transport had been open when it closed.
	AgeS float64 `json:"age_s,omitempty"`
}

TransportEvent is one entry in the manager's bounded history of transport opens and closes, with the reason a close happened. It exists because the question "why did that transport go away" had no answer: the visor log ring holds a few minutes, and a transport that was there at one reading of `tp ls` and gone at the next left nothing behind. `visor state --select diag` carries the last TransportEventRingSize of these.

type TransportStats 

type TransportStats struct {
	Total  int            `json:"total"`
	ByType map[string]int `json:"by_type"`
}

TransportStats contains statistics about transports for a given public key.

type VStream 

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

VStream is a virtual bidirectional connection over route ID 0 packets.

func (*VStream) Close 

func (s *VStream) Close() error

Close closes the virtual stream.

func (*VStream) Direct added in v1.3.99 

func (s *VStream) Direct() bool

Direct reports whether the stream arrived over a transport to its own remote peer. A relayed stream arrives over the relay's transport, which splices its bytes between two legs, and is encrypted end to end by its own Noise session.

func (*VStream) Read 

func (s *VStream) Read(p []byte) (int, error)

Read implements io.Reader. A relayed stream is read through its end to end Noise session, see secure.

func (*VStream) RemotePK 

func (s *VStream) RemotePK() cipher.PubKey

RemotePK returns the remote peer's public key.

func (*VStream) TransportID 

func (s *VStream) TransportID() uuid.UUID

TransportID returns the ID of the transport the stream rides.

func (*VStream) Write 

func (s *VStream) Write(p []byte) (int, error)

Write implements io.Writer, segmenting p into frames of at most vstreamMaxData bytes. Before this the whole of p went into ONE packet whose 16-bit size field silently truncated (uint16(len)), so a yamux frame or RPC reply over 64 KiB was written in full but declared short — the receiver parsed the excess as packet headers ("unknown packet type: Unknown(112)" with ASCII payload bytes as the type) and the session on top died with "invalid protocol version".

type VStreamInfo 

type VStreamInfo struct {
	AppName   string        `json:"app_name,omitempty"`
	RemotePK  cipher.PubKey `json:"remote_pk"`
	TpID      uuid.UUID     `json:"transport_id"`
	StreamID  uint64        `json:"stream_id"`
	SentBytes uint64        `json:"sent_bytes"`
	RecvBytes uint64        `json:"recv_bytes"`
	UptimeMS  int64         `json:"uptime_ms"`
	// ReadBufLen/ReadBufCap are this stream's inbound frame queue: how many
	// frames the transport read loop has handed over that the reader has not
	// taken yet, out of vstreamReadBuf. Len near Cap means the reader is the
	// bottleneck — deliver() is back-pressuring the whole transport's read
	// loop, and at vstreamStallTimeout the stream is given up.
	ReadBufLen int `json:"read_buf_len"`
	ReadBufCap int `json:"read_buf_cap"`
}

VStreamInfo describes one live direct (0-hop) stream: which app opened it, which peer it reaches, and over which transport.

This is the direct path's counterpart to a route group's MuxInfo. A dial that takes the AppDirectMux shortcut builds no route group by design, so every surface that reports "the route" — `proxy status`, `proxy tree`, the status.skysocks page — had nothing to show for a working session and said "(no active route group)". The path is perfectly well defined; it just was not being described.

type VStreamMux 

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

VStreamMux multiplexes virtual streams over route ID 0 packets. Multiple instances can coexist (one per packet type, e.g., DHTPacket for DHT, CascadeSetupPacket for RSN relay).

func NewVStreamMux 

func NewVStreamMux(tm *Manager, packetType routing.PacketType, log *logging.Logger) *VStreamMux

NewVStreamMux creates a virtual stream multiplexer for the given packet type.

func (*VStreamMux) Accept 

func (m *VStreamMux) Accept() (*VStream, error)

Accept returns the next incoming virtual stream.

func (*VStreamMux) Close 

func (m *VStreamMux) Close() error

Close shuts down the mux and all streams.

func (*VStreamMux) CloseStreamsOnTransport 

func (m *VStreamMux) CloseStreamsOnTransport(tpID uuid.UUID) int

CloseStreamsOnTransport closes every stream and relay leg riding tpID, returning how many streams it closed. Registered with the transport manager by NewVStreamMux and fired once per transport close.

The streams are closed WITHOUT a FIN: the transport they would go out on is the one that just died. That also keeps this callable from the manager's close path — sending would re-enter the manager to look the transport up.

func (*VStreamMux) Dial 

func (m *VStreamMux) Dial(remotePK cipher.PubKey, appName string) (*VStream, error)

Dial opens a virtual stream to a remote PK over an existing transport. appName is the originating app's name; pass "" when not known. The hook receives appName so per-app policies can branch correctly on the direct-dial path.

func (*VStreamMux) DialByTransportID 

func (m *VStreamMux) DialByTransportID(remotePK cipher.PubKey, tpID uuid.UUID) (*VStream, error)

DialByTransportID opens a virtual stream to remotePK pinned to a specific transport (looked up by ID). Use this when the caller has a specific transport in mind — e.g. ping-tree measuring the latency of *that exact* transport rather than "any direct transport to the peer." Same DMSG-exclusion + closed-transport rules as Dial; returns an error when no matching transport exists or the matching one isn't currently usable.

func (*VStreamMux) DialOnTransport 

func (m *VStreamMux) DialOnTransport(tp *ManagedTransport) (*VStream, error)

DialOnTransport opens a virtual stream on a specific transport.

func (*VStreamMux) DialOnTransportAs 

func (m *VStreamMux) DialOnTransportAs(tp *ManagedTransport, appName string) (*VStream, error)

DialOnTransportAs is DialOnTransport, recording which app the stream belongs to so StreamInfo can report it. Dial and DialByTransportID carry the caller's app name here; DialOnTransport keeps the unattributed form for internal users (setup-RPC, ping-tree) that are not an app.

func (*VStreamMux) DialThroughRelay 

func (m *VStreamMux) DialThroughRelay(relayPK, dstPK cipher.PubKey, appName string) (*VStream, error)

DialThroughRelay opens a virtual stream to dstPK via a relay visor (relayPK) that this node has a direct transport to. The SYN is signed so the relay and the destination can verify the origin, and the stream is encrypted end to end with dstPK before it is returned.

func (*VStreamMux) HandlePacket 

func (m *VStreamMux) HandlePacket(p routing.Packet, mt *ManagedTransport)

HandlePacket processes an incoming route ID 0 packet for this mux. Register this as the handler for the appropriate packet type.

func (*VStreamMux) SetDirectDialHook 

func (m *VStreamMux) SetDirectDialHook(hook DirectDialHookFn)

SetDirectDialHook attaches a hook fired before every Dial. Returning an error from the hook causes Dial to fail with that error — useful for vpn-killswitch-style policies that want to refuse direct dials over the wrong transport kind. Goroutine- safe; the hook itself must be thread-safe.

func (*VStreamMux) Stats 

func (m *VStreamMux) Stats() VStreamMuxStats

Stats snapshots the mux counters. FramesUnknownStream climbing on a healthy link means frames are arriving for streams this side does not have: the peer's transport lost a handler (see Manager.applyHandlers) or ids collided.

func (*VStreamMux) StreamInfo 

func (m *VStreamMux) StreamInfo(appName string) []VStreamInfo

StreamInfo snapshots every live stream on this mux. Pass a non-empty appName to report only that app's streams; "" reports all of them.

type VStreamMuxStats 

type VStreamMuxStats struct {
	PacketType          string `json:"packet_type"`
	Streams             int    `json:"streams"`
	RelayLegs           int64  `json:"relay_legs"`
	FramesUnknownStream int64  `json:"frames_unknown_stream"`
	StalledStreams      int64  `json:"stalled_streams"`
	AcceptDropped       int64  `json:"accept_dropped"`
	// ReadBufCap is the per-stream inbound frame queue's capacity; every
	// stream on the mux has the same one (vstreamReadBuf).
	ReadBufCap int `json:"read_buf_cap"`
	// ReadBufMaxLen is the deepest queue across the live streams right now.
	// A queue that sits near ReadBufCap is the reader — not the network —
	// setting the pace: deliver() blocks the transport's read loop once it
	// fills, and after vstreamStallTimeout it gives the stream up
	// (StalledStreams). Sustained depth on a download is the signal that the
	// receive chain, not the path, is the bottleneck.
	ReadBufMaxLen int `json:"read_buf_max_len"`
	// ReadBufNearFull counts streams whose queue is at least 90% full.
	ReadBufNearFull int `json:"read_buf_near_full"`
}

VStreamMuxStats is the mux's `visor state` view: what the logs used to be the only window on. Counters are cumulative since start.

Source Files

View all Source files

Directories

Path Synopsis
Package deprecated pkg/transport/deprecated/deprecated.go c2-net-transport
 Package deprecated pkg/transport/deprecated/deprecated.go c2-net-transport
Package network pkg/transport/network/addr_canonical.go c2-net-transport
 Package network pkg/transport/network/addr_canonical.go c2-net-transport
addrresolver
Package addrresolver pkg/transport/network/addrresolver/client.go c2-net-transport
 Package addrresolver pkg/transport/network/addrresolver/client.go c2-net-transport
handshake
Package handshake pkg/transport/network/handshake/handshake.go c2-net-transport
 Package handshake pkg/transport/network/handshake/handshake.go c2-net-transport
packetfilter
Package packetfilter pkg/transport/network/packetfilter/address-filter.go c2-net-transport
 Package packetfilter pkg/transport/network/packetfilter/address-filter.go c2-net-transport
porter
Package porter pkg/transport/network/porter/porter.go c2-net-transport
 Package porter pkg/transport/network/porter/porter.go c2-net-transport
spec
Package spec pkg/transport/network/spec/spec.go c2-net-transport Currently holds STCPConfig (the visor-side Skywire-TCP configuration stanza) — a pure-data type that pkg/visor/visorconfig.V1 embeds.
 Package spec pkg/transport/network/spec/spec.go c2-net-transport Currently holds STCPConfig (the visor-side Skywire-TCP configuration stanza) — a pure-data type that pkg/visor/visorconfig.V1 embeds.
stcp
Package stcp pkg/transport/network/stcp/pktable.go c2-net-transport
 Package stcp pkg/transport/network/stcp/pktable.go c2-net-transport
Package setup pkg/transport/setup/rpc.go c2-net-transport
 Package setup pkg/transport/setup/rpc.go c2-net-transport
Package spec pkg/transport/spec/spec.go c2-net-transport Currently holds PersistentTransports — the wire-format entry the visor stores per pinned transport peer.
 Package spec pkg/transport/spec/spec.go c2-net-transport Currently holds PersistentTransports — the wire-format entry the visor stores per pinned transport peer.
Package tpdclient pkg/transport/tpdclient/client.go c2-net-transport
 Package tpdclient pkg/transport/tpdclient/client.go c2-net-transport
Package tptypes pkg/transport/types/preference.go c2-net-transport
 Package tptypes pkg/transport/types/preference.go c2-net-transport

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.