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

dmsg

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: 44 Imported by: 0

Documentation

Overview

Package dmsg pkg/dmsg/dmsg/client.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/client_dial.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/client_relay.go

A dmsg client as a RELAY (#4484 stage 3): a peer that cannot, or should not, hold its own dmsg server sessions attaches to this client over a skywire route (the skynet carrier, see CarrierSkynet) and this client carries the peer's stream requests over its own server sessions with the existing forwardRequest/bridgeStream path. No discovery registration, no TCP listener, no server peering: the relay is a server-role session on the shared EntityCommon, nothing more.

What the relay sees is the signed StreamRequest envelope — enough to route it. What it cannot see is the stream: the Noise KK handshake inside the request runs between the attached peer and the destination, so the relay (and the server after it) only copy ciphertext. The server accepts the forwarded request because its signature verifies against the peer's own key (ServerConfig.AcceptRelayedRequests), and charges it to relay slots. The destination sees the attached peer's key, not the relay's.

Package dmsg pkg/dmsg/dmsg/client_session.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/client_sessions.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/const.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/entity_common.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/errors.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/httpsguard_other.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/listener.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/peer_miss.go

Package dmsg pkg/dmsg/dmsg/port_hits.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/quic_iface.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/quic_native.go c1-net-dmsg

All quic-go-dependent code for dmsg-over-QUIC (#2607) lives here, behind the //go:build !tinygo tag, because quic-go does not compile under TinyGo. The session core (session_common.go / stream.go / server_session.go / …) refers to QUIC only through the quicConn / quicStream interfaces (quic_iface.go); the concrete *quic.Conn / *quic.Stream adapters, the QUIC session constructors, and the client dial + server listen paths are all here. See docs/design/tinygo-dmsg-client.md.

Package dmsg pkg/dmsg/dmsg/relay_inproc.go

The IN-PROCESS attach: a second dmsg client in the same process as the first, riding the first one's sessions. It is the third and cheapest pipe into the relay (skynet in client_relay.go, a unix socket in relay_local.go, an in-memory pipe here), and the only one with no I/O under it at all.

What it is for: a visor that hosts a service which must answer under its OWN key. The resolving dmsg/skynet proxy is the case that forced it — the survey whitelist knows that key, so it cannot rotate, and it is not the visor's. Before this the only ways to run it were a second full dmsg client (its own sessions to every server, its own discovery entry, re-registered on every restart — the identity churn that got the tpviz throwaway client deleted in #4500/#4501) or a separate process attached over a unix socket, which is machinery a co-located service does not need.

Attached here the service keeps its key, holds exactly one session, publishes nothing, and its streams ride the VISOR's sessions and transports. The host pays for dmsg once.

Why this is not the thing #4500/#4501 removed: that client registered. It posted a discovery entry under a throwaway key and churned 660 identities in fourteen days. A guest here is NoRegister by construction — it has no entry to churn, and nothing dials it inbound except through its host. If a future reader is about to delete this for the same reason that one went, that is the difference to check first.

AUTHORIZATION. There is no trust boundary to enforce: both clients are constructed by the same process from the same config, so a guest can only exist because the operator put its key in that config. The allow callback is still threaded through — AcceptRelaySession takes one and the host may want to bound which of its own guests attach — but it is a statement about configuration, not a gate against an untrusted peer. That is the whole difference from relay_local.go, where the socket IS the boundary.

Package dmsg pkg/dmsg/dmsg/relay_local.go

The LOCAL half of the relay (see client_relay.go). AcceptRelaySession takes a plain net.Conn and keeps the attached peer's own identity, so the pipe under it is free: today the visor feeds it from skynet (a skywire route to skyenv.DmsgRelayPort), and here it is fed from a unix socket or a loopback listener on the same host.

What that buys is a STANDALONE process — its own keypair, its own binary, no router, no visor — holding a dmsg identity without holding dmsg server sessions. The dmsgweb proxy is the motivating case: it runs under a key the survey whitelist knows, so the key must NOT rotate, and every restart of it costs the deployment a fresh set of server sessions plus a discovery entry churn. Attached locally it keeps the key, holds exactly one session (to the visor beside it), publishes no entry, and its streams are forwarded over the VISOR's sessions and transports — which is what the visor is already paying for.

The wire under the socket is identical to the skynet one: a hello preamble so the attaching process can learn which key to run the Noise XK handshake against, then the same Noise+yamux dmsg session. Nothing above the pipe knows the difference, which is the point — the carrier stays CarrierSkynet and the session behaves as any other relay attachment (DialStream's phase 0, RelayOnly's "exactly one session", the relay-slot accounting).

AUTHORIZATION. Attaching here is a real grant: the attached process dials dmsg under ITS OWN key but over the visor's sessions, and is charged to the visor's relay slots. Two independent gates, and the acceptor is opt-in:

  • The listener itself. A unix socket's file permissions are the gate — mode 0600 means "a process running as the visor's user", which is already a process that could read the visor's own secret key off disk, so the grant adds nothing it did not have. This is why the unix socket is the default and why it is created 0600.
  • The allow callback, same shape as the skynet acceptor's. The Noise XK handshake PROVES the attacher holds the secret key for the PK it claims, so an allow that checks a configured key list is cryptographic authentication and not a hint. A listener with no filesystem gate (loopback TCP) must therefore be paired with a non-empty key list; the visor refuses to bind one otherwise.

There is deliberately no third option: no network-facing acceptor, no "anyone on localhost" TCP default.

Package dmsg pkg/dmsg/dmsg/relay_stats.go c2-dmsg-core

Package dmsg pkg/dmsg/dmsg/serve_unified.go c1-net-dmsg WebSocket on ONE listener / advertised ip:port.

Package dmsg pkg/dmsg/dmsg/serve_unified_quic_wt.go c1-net-dmsg

ServeUnifiedQUIC serves BOTH dmsg-over-QUIC and dmsg-over-WebTransport on ONE UDP socket, ALPN-demuxed on a shared quic.Transport. This is what makes WebTransport DEFAULT-ON for a dmsg server at zero extra port cost — reachable wherever dmsg-over-QUIC already is (same UDP port) — instead of requiring a separately-bound WTAddress. It mirrors how WS cmuxes the TCP port and how the visor's WT shares its unified transport_port QUIC socket.

A quic.Transport permits exactly ONE listener (transport.go: t.server != nil → errListenerAlreadySet), so dmsg-QUIC and WT CANNOT each call Listen*/ListenEarly on it — that was the original bug (the WT ListenEarly always failed with "already listening", so no server ever advertised WT). Instead we take a SINGLE ListenEarly whose tls.Config.GetConfigForClient returns the dmsg identity cert (skywire ALPN, mutual-TLS PK binding) or the WebTransport cert ("h3", no client cert — browsers present none) depending on the ClientHello's offered ALPNs; the accept loop then dispatches each connection by its negotiated ALPN.

Package dmsg pkg/dmsg/dmsg/server.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/server_inproc.go c1-net-dmsg

In-process sessions. A visor that runs a dmsg server and also hosts services under their own keys has each service's dmsg client hold a session with that server like any other client, over QUIC or TCP on loopback. Every stream a remote visor opens to the service through this server then crosses the loopback session as well, so the host pays the carrier's crypto and syscalls twice. A server opted in with ServeInProcess is reached by clients in the same process over an in-memory pipe instead. The session is otherwise a normal one, with the same Noise handshake and stream mux.

Package dmsg pkg/dmsg/dmsg/server_session.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/session_common.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/stream.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/tcpnodelay_native.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/traffic.go c2-dmsg-core

Package dmsg pkg/dmsg/dmsg/types.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/util.go c1-net-dmsg

Package dmsg pkg/dmsg/dmsg/ws.go c1-net-dmsg

A WebSocket connection is a bidirectional, ordered, reliable byte pipe over HTTP(S) — exactly what dmsg's session layer wants. coder/websocket's NetConn adapts it to a net.Conn, so a WS session reuses the EXISTING Noise+yamux stack (makeServerSession/makeClientSession + yamux) with no new session semantics, no new mux, and no new crypto — the per-hop Noise handshake and the end-to-end per-stream noise are both unchanged. WS is purely a different way to carry the same bytes.

Why it matters: a browser tab (JS or Go-js/wasm) cannot open a raw TCP or UDP socket, so neither the legacy TCP transport nor dmsg-over-QUIC can reach the mesh from a browser. WebSocket is the one transport the browser sandbox allows, which makes a WASM dmsg *client* — a real PK-authenticated leaf node running in a web page — possible. It is also the universal fallback for restrictive networks that only permit HTTP(S)/443 egress and for CDN fronting. coder/websocket is used precisely because it also compiles to js/wasm, so the same dial path serves the future browser client.

Package dmsg pkg/dmsg/dmsg/ws_server.go c1-net-dmsg

Split out of ws.go behind //go:build !tinygo: ServeWS runs an http.Server (net/http), which is broken on the TinyGo js target. The CLIENT dial path (dialSessionWS) stays in ws.go untagged — it uses only coder/websocket, which compiles to js/wasm, so the WASM dmsg client can still dial WS servers.

Package dmsg pkg/dmsg/dmsg/wt.go c1-net-dmsg

Build tag: WebTransport pulls in quic-go + webtransport-go, neither of which compiles under TinyGo (quic-go needs crypto/tls.QUICEncryptionLevel). It is a server-side + browser transport, so a TinyGo (IoT) dmsg client doesn't need it; the //go:build !tinygo tag keeps it out of the TinyGo graph. See docs/design/tinygo-dmsg-client.md.

WebTransport (HTTP/3 over QUIC) is the one browser-reachable transport that does NOT need a CA-issued TLS certificate: the browser's WebTransport constructor accepts a self-signed cert via `serverCertificateHashes` (a pinned SHA-256). So a dmsg server can present a short-lived self-signed ECDSA cert (skyquic.NewWebTransportCertificate), publish its hash in discovery, and a browser connects to it by bare IP — no Caddy, no domain, no Let's Encrypt.

Unlike dmsg-over-QUIC (which authenticates BOTH peers via mutual PK-bound TLS), a browser WebTransport client can't present a client certificate. So WT authenticates only the SERVER (cert-hash pinning); the client→server skywire PK is authenticated in the dmsg Noise handshake, exactly like the TCP/WS paths. Accordingly a WT session here is used as a single bidirectional stream carrying the EXISTING Noise+yamux session — the WS code path, over a different transport. (WT's native stream multiplexing is intentionally not used; yamux multiplexes inside the one WT stream, keeping one session implementation.)

Package dmsg pkg/dmsg/dmsg/wt_advertise.go c1-net-dmsg

What a server advertises about its WebTransport endpoint. Untagged, unlike the WebTransport server itself (wt.go), because callers that only advertise — a visor configuring a dmsg server it hosts — build for every target.

Index

Constants

View Source 
const (
	CarrierTCP  = "tcp"
	CarrierWS   = "ws"
	CarrierWT   = "wt"
	CarrierQUIC = "quic"

	// CarrierSkynet is the skynet carrier: the session rides a skywire route to
	// a visor acting as a dmsg relay (not a discovery-registered server). It is
	// selected by the server entry's address, never by Config.Carriers, and
	// dialed through Config.SessionDialer.
	CarrierSkynet = "skynet"
)

Carrier names for Config.Carriers.

View Source 
const (
	DefaultMinSessions = 1

	DefaultUpdateInterval = time.Minute

	DefaultMaxSessions = 2048

	// DefaultMaxPeerLinks bounds concurrent peer sessions (inbound
	// announced + configured outbound) on an always-on relay server.
	DefaultMaxPeerLinks = 512

	// DefaultMaxRelayedStreams bounds concurrent streams a server relays
	// on behalf of a peer, so an always-open relay can't be amplified.
	DefaultMaxRelayedStreams = 4096

	DefaultDmsgHTTPPort = uint16(80)

	DefaultOfficialDmsgServerType = "official"

	DefaultCommunityDmsgServerType = "community"
)

Constants.

View Source 
const (
	// Type returns the stream type string.
	Type = "dmsg"

	// HandshakePayloadVersion contains payload version to maintain compatibility with future versions
	// of HandshakeData format.
	HandshakePayloadVersion = "2.0"
)
View Source 
const CarrierInProcess = "inproc"

CarrierInProcess names a session over an in-memory pipe to a server in the same process. It is never dialed by preference, only when the server is here.

View Source 
const DefaultClientMaxRelayedStreams = DefaultMaxRelayedStreams

DefaultClientMaxRelayedStreams is the relay-slot cap a visor gives its dmsg client when it runs the relay acceptor.

It was 256, sized for a desk's worth of streams from a handful of attached peers. That is the wrong order of magnitude for a hub: a visor co-resident with a dmsg server is what visors attach to when they stop holding their own server sessions, and one core dmsg server carries several hundred client connections today (771 on a production server, 2026-09-10). A hub fronting a few hundred of those, each with a few concurrent streams, exhausts 256 long before it runs out of anything real, and the failure is a refused stream that looks like an unreachable service.

Any visor with transports can be nominated as a relay by a peer, and a visor co-resident with a dmsg server is doing a server's job outright, so this now matches DefaultMaxRelayedStreams: what the server beside it would allow. Bound it per visor with dmsg.relay_max_streams.

View Source 
const SkynetScheme = "skynet://"

SkynetScheme prefixes the advertised address of a dmsg relay reachable over a skywire route: "skynet://<relay pk>:<dmsg port>". See SkynetAddr.

View Source 
const WTPath = "/dmsg"

WTPath is the HTTP/3 path the WebTransport endpoint is served on. The advertised Server.AddressWT URL includes it (e.g. "https://host:port/dmsg").

Variables

View Source 
var (
	ErrDiscEntryNotFound       = registerErr(Error{code: 100, msg: "entry is not found in discovery"})
	ErrDiscEntryIsNotServer    = registerErr(Error{code: 101, msg: "entry is not of server in discovery"})
	ErrDiscEntryIsNotClient    = registerErr(Error{code: 102, msg: "entry is not of client in discovery"})
	ErrDiscEntryHasNoDelegated = registerErr(Error{code: 103, msg: "client entry in discovery has no delegated servers"})
)

Errors for dmsg discovery (1xx).

View Source 
var (
	ErrEntityClosed               = registerErr(Error{code: 200, msg: "local entity closed"})
	ErrSessionClosed              = registerErr(Error{code: 201, msg: "local session closed"})
	ErrCannotConnectToDelegated   = registerErr(Error{code: 202, msg: "cannot connect to delegated server"})
	ErrSessionHandshakeExtraBytes = registerErr(Error{code: 203, msg: "extra bytes received during session handshake"})
)

Entity Errors (2xx).

View Source 
var (
	ErrReqInvalidSig        = registerErr(Error{code: 300, msg: "request has invalid signature"})
	ErrReqInvalidTimestamp  = registerErr(Error{code: 301, msg: "request timestamp should be higher than last"})
	ErrReqInvalidSrcPK      = registerErr(Error{code: 302, msg: "request has invalid source public key"})
	ErrReqInvalidDstPK      = registerErr(Error{code: 303, msg: "request has invalid destination public key"})
	ErrReqInvalidSrcPort    = registerErr(Error{code: 304, msg: "request has invalid source port"})
	ErrReqInvalidDstPort    = registerErr(Error{code: 305, msg: "request has invalid destination port"})
	ErrReqNoListener        = registerErr(Error{code: 306, msg: "request has no associated listener", temp: true})
	ErrReqNoNextSession     = registerErr(Error{code: 307, msg: "request cannot be forwarded because the next session is non-existent"})
	ErrRelayCapacityReached = registerErr(Error{code: 308, msg: "server is at its peer-relay stream capacity", temp: true})

	ErrDialRespInvalidSig  = registerErr(Error{code: 350, msg: "response has invalid signature"})
	ErrDialRespInvalidHash = registerErr(Error{code: 351, msg: "response has invalid hash of associated request"})
	ErrDialRespNotAccepted = registerErr(Error{code: 352, msg: "response rejected associated request without reason"})

	ErrSignedObjectInvalid = registerErr(Error{code: 370, msg: "signed object is invalid"})
)

Errors for dial request/response (3xx).

View Source 
var (
	ErrPortOccupied    = registerErr(Error{code: 400, msg: "port already occupied"})
	ErrAcceptChanMaxed = registerErr(Error{code: 401, msg: "listener accept chan maxed", temp: true})
)

Listener errors (4xx).

View Source 
var (
	// DialTimeout defines the duration a TCP dial to a dmsg server should take.
	// This prevents blocking for minutes on unresponsive/overloaded servers.
	DialTimeout = 10 * time.Second

	// HandshakeTimeout defines the duration a stream handshake should take.
	// On a healthy network the noise handshake completes in milliseconds.
	// 5s is generous — anything longer means the destination is unreachable.
	HandshakeTimeout = 5 * time.Second

	// StreamIdleTimeout defines how long a stream can be idle (no reads)
	// before it is considered stale and closed. This prevents streams stuck
	// in waitRead from holding ephemeral ports indefinitely.
	StreamIdleTimeout = 2 * time.Minute

	// AcceptBufferSize is the buffer depth of each dmsg Listener's
	// accept channel — how many newly-arrived (but not yet
	// AcceptStream-drained) streams the listener can hold before it
	// starts dropping with "Accept buffer full" + ErrAcceptChanMaxed.
	//
	// Sized for the largest plausible burst of inbound REQUEST
	// frames between consumer drains. Production TPD's CXO
	// subscriber listener (port 50) was empirically hitting steady-
	// state ~16 drops/sec with a peer count in the few-hundreds
	// range — small enough that the constant being 20 (the original
	// value, intended for hobbyist visor accept rates) saturated
	// within ~1s of a deploy-restart traffic spike and never
	// recovered. Bumping to 256 absorbs ~16s of burst at that rate
	// before drops start, and is still tiny in absolute memory
	// terms (256 pointers ≈ 2 KiB per listener). Listeners that see
	// less load (visor-side ports, etc.) pay the same trivial cost.
	AcceptBufferSize = 256
)
View Source 
var DmsghttpJSON = deployment.ServicesJSON

DmsghttpJSON is services-config.json embedded in deployment.ServicesJSON

View Source 
var ErrClosed = errors.New("server closed")

ErrClosed is returned when an operation is attempted on a closed server.

View Source 
var ErrDatagramsUnsupported = errors.New("dmsg: session does not support datagrams (not a QUIC session)")

ErrDatagramsUnsupported is returned by the session datagram API for non-QUIC sessions (TCP+yamux/smux carries reliable streams only).

View Source 
var ErrLocalRelayHello = errors.New("dmsg: not a local dmsg relay acceptor")

ErrLocalRelayHello is returned when the bytes at the head of a local relay conn are not a hello this build understands — most often because the socket path points at something that is not a dmsg relay acceptor at all.

View Source 
var ErrRelayPeerNotAllowed = errors.New("dmsg: relay peer not allowed")

ErrRelayPeerNotAllowed is returned by AcceptRelaySession when the allow callback refuses the handshaken peer.

Functions

func AttachInProcess 

func AttachInProcess(ctx context.Context, host, guest *Client, port uint16, allow func(cipher.PubKey) bool) error

AttachInProcess points guest at host's relay: guest holds one session, to host, carried by an in-memory pipe. port is the relay port the synthetic server entry advertises — it is never dialed, but SetRelayPeers builds the entry from it and the dialer checks the key it names.

Returns an error rather than attaching when the two clients share a key. Two dmsg clients on one key evict each other from every server they share, and a guest that is really its host is a configuration mistake worth naming at startup instead of debugging later.

func DiscAddr 

func DiscAddr() string

DiscAddr returns the dmsg address of the dmsg discovery service in the format "dmsg://<pk>:<port>"

func DiscURL 

func DiscURL() string

DiscURL returns the URL of the dmsg discovery service

func ErrorFromCode 

func ErrorFromCode(code errorCode) (bool, error)

ErrorFromCode returns a saved error (if exists) from given error code.

func ExtractPKFromDmsgAddr 

func ExtractPKFromDmsgAddr(input string) string

ExtractPKFromDmsgAddr returns the public key of the dmsg address input in this format in the format "dmsg://<pk>:<port>"

func InitConfig 

func InitConfig() error

InitConfig initialized the config

func ParseSkynetAddr 

func ParseSkynetAddr(addr string) (cipher.PubKey, uint16, error)

ParseSkynetAddr parses an address produced by SkynetAddr.

func ProtocolLabel 

func ProtocolLabel(carrier, addr string) string

ProtocolLabel renders a human-readable label for the protocol a client used to reach a dmsg server, given the carrier it dialed and that carrier's endpoint address. Every dmsg carrier is Noise-encrypted end to end above the byte pipe; the label names the underlying transport, and for the WebSocket carrier it distinguishes plain ws:// from TLS-secured wss:// by the endpoint scheme. An empty carrier (an accepted, server-side session) renders as "accepted".

func SignBytes 

func SignBytes(b []byte, sk cipher.SecKey) (cipher.Sig, error)

SignBytes signs the provided bytes with the given secret key.

func SkynetAddr 

func SkynetAddr(pk cipher.PubKey, port uint16) string

SkynetAddr renders the advertised address of a dmsg relay listening on the given dmsg port of the visor pk. A server entry carrying it is dialed over the skynet carrier.

func SmuxConfig 

func SmuxConfig() *smux.Config

SmuxConfig returns a tuned smux configuration for dmsg sessions.

func WithServerSweep 

func WithServerSweep(ctx context.Context) context.Context

WithServerSweep opts the dials made with the returned context into a last-resort sweep: after the normal ladder fails, establish NEW sessions to every server in the deployment and try the destination through each.

This is deliberately OPT-IN and must stay that way. The automatic fallback (dialViaConnectedServers) only reuses sessions this client already holds, so it costs a yamux stream per server and nothing more. A sweep costs a full Noise handshake per server. Firing that automatically on every failed resolution would turn one typo'd or dead public key into a handshake storm against the whole server fleet — and would defeat the dial-failure backoff (Client.dialFail), which exists precisely because repeated failed dials are expensive.

It is for an operator dialing a key they know about out of band: a client that is in no discovery, was never seeded, and sits on a server this client holds no session to. That is the one case the normal ladder cannot reach.

func YamuxConfig 

func YamuxConfig() *yamux.Config

YamuxConfig returns a tuned yamux configuration for dmsg sessions.

Types

type Addr 

type Addr struct {
	PK   cipher.PubKey `json:"public_key"`
	Port uint16        `json:"port"`
}

Addr implements net.Addr for dmsg addresses.

func (Addr) Network 

func (Addr) Network() string

Network returns "dmsg"

func (*Addr) Set 

func (a *Addr) Set(s string) error

Set implements pflag.Value for Addr.

func (Addr) ShortString 

func (a Addr) ShortString() string

ShortString returns a shortened string representation of the address.

func (Addr) String 

func (a Addr) String() string

String returns public key and port of node split by colon.

func (Addr) Type 

func (Addr) Type() string

Type implements pflag.Value for Addr.

type CXOKeepaliveFunc added in v1.3.98 

type CXOKeepaliveFunc func() (healthy bool, epoch uint64)

CXOKeepaliveFunc reports whether a registration-over-CXO feed is keeping this client's entry alive, and the epoch of the subscription doing it. The epoch changes whenever the subscription arrives on a new conn.

type Client 

type Client struct {
	EntityCommon

	// Lookup counters for diagnostics.
	LookupCacheHits    atomic.Int64 // resolved from entry cache
	LookupResolverHits atomic.Int64 // resolved from an injected EntryResolver
	LookupHTTPHits     atomic.Int64 // resolved from HTTP discovery
	LookupHTTPMisses   atomic.Int64 // HTTP discovery returned not found
	LookupStaleHits    atomic.Int64 // a failed lookup answered from an expired cached entry
	// contains filtered or unexported fields
}

Client represents a dmsg client entity.

func NewClient 

func NewClient(pk cipher.PubKey, sk cipher.SecKey, dc disc.APIClient, conf *Config) *Client

NewClient creates a dmsg client entity.

func (*Client) AcceptLocalRelaySession 

func (ce *Client) AcceptLocalRelaySession(ctx context.Context, conn net.Conn, allow func(cipher.PubKey) bool) error

AcceptLocalRelaySession is AcceptRelaySession for an attach this HOST made — the unix socket, the loopback listener, or the in-process pipe. Its streams are not charged to the relay budget: that budget bounds what this visor carries for OTHER PEOPLE, and a service the operator runs beside the visor under its own key is not other people. The socket's file mode and allowed_keys are that path's gate.

func (*Client) AcceptRelaySession 

func (ce *Client) AcceptRelaySession(ctx context.Context, conn net.Conn, allow func(cipher.PubKey) bool) error

AcceptRelaySession runs one REMOTE peer's relay session over conn until the peer hangs up, ctx is canceled, or the client closes; it blocks for the session's lifetime, so callers run it in the accept loop's goroutine. The client answers the peer's Noise XK handshake as the server side, learns the peer's key from it, and asks allow (if non-nil) whether to keep the peer; then it serves the yamux session the way a dmsg server serves a client: each stream request is bridged to another peer attached here, or forwarded over one of this client's own server sessions, charged to Config.MaxRelayedStreams. Newest-session-wins per peer key, like servers.

conn is closed on every return.

func (*Client) AddDiscovery 

func (c *Client) AddDiscovery(client disc.APIClient, discPK cipher.PubKey)

AddDiscovery registers an additional dmsg-discovery this client should publish its entry to. discPK, when set, identifies the discovery's own dmsg-server PK so the client can detect inbound sessions originated by this discovery (currently unused on the client side; reserved for symmetry with Server).

Safe to call before or after Serve. Subsequent updateClientEntry iterations pick up the new endpoint automatically.

func (*Client) AddEntryResolver 

func (ce *Client) AddEntryResolver(r EntryResolver)

AddEntryResolver installs a resolver, consulted before the discovery in the order added.

Call it during setup, before the client starts dialing: the resolver list is read without a lock on the dial path, which is safe for an append-only list that is fully built before serving and would not be safe for mutation afterwards.

func (*Client) AllEntries 

func (ce *Client) AllEntries(ctx context.Context) (entries []string, err error)

AllEntries returns all the entries registered in discovery

func (*Client) AllSessions 

func (ce *Client) AllSessions() []ClientSession

AllSessions obtains all established sessions.

func (*Client) AllStreams 

func (ce *Client) AllStreams() (out []*Stream)

AllStreams returns all the streams of the current client.

func (*Client) AllVisorEntries 

func (ce *Client) AllVisorEntries(ctx context.Context) (entries []string, err error)

AllVisorEntries returns all the entries registered in discovery that are visor

func (*Client) AttachLocalRelay 

func (ce *Client) AttachLocalRelay(ctx context.Context, network, addr string) (cipher.PubKey, error)

AttachLocalRelay points this client at the local relay acceptor at network/addr: it probes the acceptor for its identity, installs the session dialer that serves the skynet carrier from that socket, and nominates the acceptor as this client's relay. It returns the acceptor's public key.

It does NOT change this client's identity, and deliberately cannot: the key is fixed at NewClient, the acceptor learns it from the session handshake, and every stream this client dials afterwards is signed with it and arrives at the destination under it. That is the whole reason to attach rather than to run a second client — a key the deployment already knows (a whitelisted survey key, a service PK in someone's config) keeps working.

Pair it with Config.RelayOnly to get the intended end state: one session, to the visor, and no discovery entry at all. Without RelayOnly the client also keeps MinSessions server sessions and publishes an entry naming them, which is a legitimate but different thing to want.

func (*Client) Carriers 

func (ce *Client) Carriers() []string

Carriers returns the client's effective ordered carrier preference (the runtime override if set, else Config.Carriers).

func (*Client) Close 

func (ce *Client) Close() error

Close closes the dmsg client entity and waits for background goroutines to finish.

func (*Client) ConnectToAllServers 

func (ce *Client) ConnectToAllServers(ctx context.Context) (ConnectToAllServersResult, error)

ConnectToAllServers enumerates every server entry in dmsg discovery and ensures the client has an active session to each one. Useful as a one-shot "maximize connectivity now" action for visors that run route/transport setup nodes and need to reach arbitrary destinations without bouncing through phase-3 new-session dials.

The client's configured MinSessions is NOT mutated — once a session dies, the normal reconnect behavior applies. Use SetSessionsCount (or restart the client with sessions_count: 0) if you want the "keep sessions to all servers" behavior to persist.

func (*Client) ConnectedServers deprecated 

func (ce *Client) ConnectedServers() []string

ConnectedServers obtains all the servers client is connected to.

Deprecated: we can now obtain the remote TCP address of a session from the ClientSession struct directly.

func (*Client) ConnectedServersPK 

func (ce *Client) ConnectedServersPK() []string

ConnectedServersPK return keys of all connected dmsg servers

func (*Client) ConnectionsSummary 

func (ce *Client) ConnectionsSummary() ConnectionsSummary

ConnectionsSummary returns a summary of all connected clients, and the associated servers that connect them.

func (*Client) ConvergeCarriers 

func (ce *Client) ConvergeCarriers() int

func (*Client) CountTraffic added in v1.3.100 

func (ce *Client) CountTraffic()

CountTraffic starts counting the streams this client accepts on each of its listening ports, and the bytes they carry. For services, which serve every API and feed on a port of their own.

func (*Client) Dial 

func (ce *Client) Dial(ctx context.Context, addr Addr) (net.Conn, error)

Dial wraps DialStream to output net.Conn instead of *Stream.

func (*Client) DialPaused 

func (ce *Client) DialPaused() bool

DialPaused reports whether SetDialPaused(true) is in force.

func (*Client) DialStream 

func (ce *Client) DialStream(ctx context.Context, addr Addr) (*Stream, error)

DialStream dials to a remote client entity with the given address.

func (*Client) DialStreamVia 

func (ce *Client) DialStreamVia(ctx context.Context, serverPK cipher.PubKey, addr Addr) (*Stream, error)

DialStreamVia dials addr through a PINNED rendezvous server: it establishes a session to serverPK and dials the destination over that session. This is how a browser reaches a direct or hidden client by naming the server it sits on — <server-pk>.<dest-pk>.dmsg — when the destination's own entry names no server the dialer can use.

A relay-attached client cannot do that, and should not have to. It holds one session, the relay, and an empty discovery by construction (see dmsgclient.StartDmsgLocalRelay), so EnsureAndObtainSession(serverPK) fails the entry lookup with error 100 whichever server is named — there is no address for it to dial. But a pin is not a demand for a particular session; it is the caller saying where the destination can be found. A relay answers that question better: DialStream tries relay sessions before any lookup, the relay resolves the destination with the visor's entry cache and forwards over the visor's own sessions, and since #4829 it can deliver over a terminal skynet leg straight to the destination visor. Falling through to the ordinary ladder honors the pin's intent rather than failing on its mechanism.

With no relay the pin stays binding: the caller named a server deliberately and nothing here knows better.

func (*Client) DiscEntry 

func (ce *Client) DiscEntry(ctx context.Context, pk cipher.PubKey) (*disc.Entry, error)

DiscEntry looks up a PK in dmsg-discovery and returns the entry if it exists as a client with at least one delegated server. Returns an error if the PK is not found, is a server entry, or has no delegated servers. This is a lightweight check that does NOT dial — useful to pre-filter before expensive dial attempts.

func (*Client) EnsureAndObtainSession 

func (ce *Client) EnsureAndObtainSession(ctx context.Context, srvPK cipher.PubKey) (ClientSession, error)

EnsureAndObtainSession attempts to obtain a session. If the session does not exist, we will attempt to establish one. It returns an error if the session does not exist AND cannot be established.

sesMx is released across the discovery lookup (getServerEntry) because ce.dc may be a dmsgfirst-wrapped client whose primary path dials over DMSG itself — DialStream → phase 3 → EnsureAndObtainSession on the same goroutine, which would re-acquire sesMx and self-deadlock on a non-reentrant mutex. It is released across the dial too, for the same reason; what serializes concurrent dials to one server is not sesMx but dialSessionOnce, which is where the reason they must be serialized is written down.

We also resolve the server entry from the local entryCache first when possible. If we go to ce.dc here, the dmsgfirst wrapper's primary path is dmsghttp — which is itself a DialStream that lands back in EnsureAndObtainSession to set up its session, calls getServerEntry, and recurses into the same disc.Entry call until the goroutine stack blows up. The cache short-circuit removes that loop entirely for any server PK we already know about (configured servers are pre-seeded by dmsgc.New, so the common case never makes a network lookup; runtime- learned servers fall through to the slower disc-client path which only recurses if we genuinely don't have the entry yet — and even then, the recursion bottoms out as soon as one configured server's session is dialed and seeds itself).

func (*Client) EnsureSession 

func (ce *Client) EnsureSession(ctx context.Context, entry *disc.Entry) error

EnsureSession ensures the existence of a session. It returns an error if the session does not exist AND cannot be established.

func (*Client) ForceReconnect 

func (ce *Client) ForceReconnect() int

ForceReconnect closes every active server session. The reconnect loop (15 s cadence) and delSession → nudgeEntryUpdate path together will re-dial fresh sessions and refresh the discovery entry's delegated_servers list.

This exists as a recovery hook for the visor's self-probe: when a visor's own listeners become unreachable through its existing sessions (typically a one-sided view — server has the session, we don't, or vice versa), nothing in steady-state detects it because our side thinks the session is fine. Closing and re-dialing clears the staleness.

Returns the number of sessions that were torn down. Safe to call concurrently with normal operation; in-flight streams on the closed sessions will fail and the caller is expected to retry on a fresh session.

func (*Client) InProcessRelayDialer 

func (ce *Client) InProcessRelayDialer(ctx context.Context, allow func(cipher.PubKey) bool) SessionDialer

InProcessRelayDialer returns a SessionDialer that attaches the caller to host's relay over an in-memory pipe.

Each dial makes a net.Pipe, hands the host end to AcceptRelaySession on a goroutine and returns the guest end. There is no hello preamble (relay_local needs one only because a separate process cannot know the acceptor's key before the Noise XK handshake; here the caller was handed the host client itself) and no listener (nothing accepts — the dial IS the attach).

The returned dialer refuses any carrier but skynet and any address but the host's own, so a guest whose relay peers were repointed elsewhere fails loudly instead of silently attaching to the wrong end of its own pipe.

func (*Client) Listen 

func (ce *Client) Listen(port uint16) (*Listener, error)

Listen listens on a given dmsg port.

func (*Client) LookupIP 

func (ce *Client) LookupIP(ctx context.Context, servers []cipher.PubKey) (myIP net.IP, err error)

LookupIP dails to dmsg servers for public IP of the client.

func (*Client) LookupIPGeo 

func (ce *Client) LookupIPGeo(ctx context.Context, servers []cipher.PubKey) (StreamResponse, error)

LookupIPGeo dials a dmsg-server for both the client's public IP and its geolocation in a single round-trip. Falls through to dmsg-servers exactly like LookupIP — first a session-mate, then any delegated server. Older dmsg-servers without a geo lookup configured return GeoCountry="" — the caller should fall back to a local geoip lookup in that case.

func (*Client) LookupIPsByFamily 

func (ce *Client) LookupIPsByFamily(_ context.Context) (v4, v6 net.IP)

LookupIPsByFamily iterates the client's connected dmsg sessions, calls LookupIP on each, and collects ONE public IP per family. Used by the visor to learn BOTH its IPv4 and IPv6 public addresses when its dmsg-client has sessions over both families — typically because at least one dmsg-server has AddressV6 set (Phase 2a) and Phase 3's happy-eyeballs dialed it via IPv6.

Each returned IP is validated via netutil.IsPublicIP; non-public values are skipped. Iteration stops once both families are populated; if only one is available, the other return value is nil. Both nil indicates no public IPs could be learned (no sessions, or all sessions returned non-public addresses).

This is family-aware in the SOURCE direction (the visor's own IP observed by each dmsg-server). The session's outbound family — v4 or v6 — determines what LookupIP returns for that session. A dual-stack visor with a v6-dmsg-server session AND a v4-dmsg-server session learns both addresses in a single iteration. Phase 2c uses these to populate LocalAddresses.PublicIP / PublicIPv6 declared fields in AR bind payloads, so the AR can register both family records even when reached via a dmsg-bridge that hides the visor's actual source family.

func (*Client) MaxRelayedStreams 

func (ce *Client) MaxRelayedStreams() int

MaxRelayedStreams is the relay-slot cap RelayedStreams is charged against (Config.MaxRelayedStreams, i.e. dmsg.relay_max_streams). Zero or less means this client refuses to relay at all. Set once at construction.

func (*Client) MinSessions 

func (ce *Client) MinSessions() int

MinSessions returns the current minimum session count.

func (*Client) PinnedFailureCount 

func (ce *Client) PinnedFailureCount() int64

PinnedFailureCount returns the number of consecutive failed Serve-loop passes while the client is pinned to a single dmsg server via the "dmsgServer" context value. Reset to 0 whenever a session is successfully established. Callers (the visor's startup path) use this to bound how long the visor waits for a pinned server before aborting startup.

func (*Client) PorterCount 

func (ce *Client) PorterCount() int

PorterCount returns the number of reserved ephemeral ports.

func (*Client) PorterDiag 

func (ce *Client) PorterDiag() netutil.EphemeralDiagResult

PorterDiag returns detailed diagnostic information about ephemeral port reservations. This helps identify what is holding ports when the porter approaches exhaustion.

func (*Client) Probe 

func (ce *Client) Probe(ctx context.Context, pk cipher.PubKey, port uint16) bool

Probe checks whether a remote dmsg client is reachable by performing a lightweight DialStream to the specified port and immediately closing. On the remote side, the stream is accepted and the noise handshake completes — confirming end-to-end DMSG reachability. The remote's application-level handler (e.g. router.serveSetup on port 136) will typically close the stream quickly (untrusted PK), but that's fine: the dial success is all we need.

This is useful as a pre-flight check before expensive operations like route setup: if Probe fails, the destination is unreachable and there is no point attempting a full route setup that will time out.

func (*Client) ProbeErr 

func (ce *Client) ProbeErr(ctx context.Context, pk cipher.PubKey, port uint16) error

ProbeErr is Probe with the reason kept. Probe answers only "reachable or not", which makes an unreachable result undiagnosable: the caller cannot tell a destination with no listener on that port from one whose dmsg server stopped listening from a dial that ran out of budget. Every one of those reads as a bare "no" in `skywire cli dmsg probe`'s output, whose ERROR column was consequently always empty.

func (*Client) ProbeViaServer 

func (ce *Client) ProbeViaServer(ctx context.Context, pk cipher.PubKey, port uint16, serverPK cipher.PubKey) bool

ProbeViaServer is like Probe but forces the stream through a specific dmsg server (serverPK), so it tests reachability of pk:port VIA that server rather than via whichever delegated/shared server DialStream would otherwise pick. Useful for per-server reachability diagnosis ("is pk reachable through server X but not server Y?"). It ensures a session to serverPK first, then dials the target through that session.

func (*Client) Ready 

func (ce *Client) Ready() <-chan struct{}

Ready returns a chan which blocks until the client has at least one delegated server and has an entry in the dmsg discovery.

func (*Client) RelayBackoffs 

func (ce *Client) RelayBackoffs() map[cipher.PubKey]time.Time

RelayBackoffs lists the relay nominees currently backed off after a failed attach, with the time each backoff lapses. For `visor state`.

func (*Client) RelayDialSkips 

func (ce *Client) RelayDialSkips() map[cipher.PubKey]time.Time

RelayDialSkips lists the destinations DialStream is currently not trying over the relay (the relay failed to reach them), with the skip's end.

func (*Client) RelayPeers 

func (ce *Client) RelayPeers() []cipher.PubKey

RelayPeers returns the current relay nominees.

func (*Client) RelayRefused 

func (ce *Client) RelayRefused() int

RelayRefused reports how many stream requests this client has turned away with ErrRelayCapacityReached since start — either the global budget or one peer's share of it was full. Surfaced in `visor state` so a hub at its cap is diagnosable from the hub rather than inferred from the peers it refused.

func (*Client) RelaySessionStreams 

func (ce *Client) RelaySessionStreams() map[cipher.PubKey]int

RelaySessionStreams is the open-stream count of every peer attached to this client as a relay (the "Relaying for" side).

func (*Client) RelaySessions 

func (ce *Client) RelaySessions() []cipher.PubKey

RelaySessions lists the peers currently attached to this client as a relay.

func (*Client) RelayedSince 

func (ce *Client) RelayedSince(t time.Time) bool

RelayedSince reports whether ANY of this client's sessions has read bytes since t.

The client-wide version of SessionCommon.ReadSince, for a caller deciding whether to tear every session down — the visor's self-probe does, and a visor with traffic moving is not the unreachable visor that recovery is for. See decideReap for the same rule applied per session.

func (*Client) RelayedStreams 

func (ce *Client) RelayedStreams() int

RelayedStreams reports how many relayed streams this client is carrying.

func (*Client) ResetPorter 

func (ce *Client) ResetPorter() int

ResetPorter frees all ephemeral port reservations, recovering from port exhaustion. Well-known ports (listeners) are preserved. Returns the number of ports freed.

func (*Client) SeedEntryCache 

func (ce *Client) SeedEntryCache(pk cipher.PubKey, entry *disc.Entry)

SeedEntryCache injects a client entry into the discovery cache so that DialStream can find the destination without querying the HTTP discovery. This is used for deployment services that run as direct DMSG clients (they don't register in the public discovery). The caller should set entry.Client.DelegatedServers to all known DMSG server PKs so that DialStream can try each connected server.

Seeded entries never expire (timestamp set far in the future).

func (*Client) Serve 

func (ce *Client) Serve(ctx context.Context)

Serve serves the client. It blocks until the client is closed.

func (*Client) ServeLocalRelay 

func (ce *Client) ServeLocalRelay(ctx context.Context, lis net.Listener, port uint16, allow func(cipher.PubKey) bool) error

ServeLocalRelay accepts local attach conns on lis and serves each as a relay session on this client, exactly as the skynet acceptor does — the pipe is the only difference. It blocks until lis fails or ctx is done, so callers run it in their own goroutine; lis is closed on return.

port is what the hello advertises: the dmsg port this visor's relay is published on (skyenv.DmsgRelayPort), so an attached peer nominates the same address whichever pipe it came in over.

allow is the SAME callback shape the skynet acceptor passes to AcceptRelaySession, and carries the same meaning: it is consulted after the Noise handshake has proven the peer's key, and refusing closes the conn. A nil allow admits any key that can open lis — correct ONLY when the listener itself is the gate (a 0600 unix socket). See this file's package comment.

func (*Client) Session 

func (ce *Client) Session(pk cipher.PubKey) (ClientSession, bool)

Session obtains an established session.

func (*Client) SessionCarriers 

func (ce *Client) SessionCarriers() []SessionCarrier

SessionCarriers returns the carrier each live server session is using, plus the most-preferred carrier that server currently advertises. Read-only.

func (*Client) SetCarriers 

func (ce *Client) SetCarriers(carriers []string)

SetCarriers overrides the client's ordered carrier preference at runtime (nil restores Config.Carriers). See ConvergeCarriers.

func (*Client) SetDialPaused 

func (ce *Client) SetDialPaused(p bool)

SetDialPaused stops the client dialing server sessions while p is true. Sessions already up stay up and are not redialed when they drop. False resumes dialing.

func (*Client) SetDiscoveryClients 

func (c *Client) SetDiscoveryClients(clients []disc.APIClient)

SetDiscoveryClients replaces the underlying disc.APIClient on each configured discovery endpoint, preserving PKs. The lengths must match the current endpoint count, otherwise the call is a no-op. Used to upgrade plain HTTP clients to dmsgfirst-wrapped clients once an outbound dmsg.Client is available.

func (*Client) SetLiveDiscovery 

func (ce *Client) SetLiveDiscovery(dc disc.APIClient)

SetLiveDiscovery installs a live-only discovery client used by carrier convergence to read a server's CURRENT WT/QUIC endpoints. The client's default disc (ce.dc) is a static-seed-first fallback that shadows those rotating fields, so convergence would otherwise never see them.

func (*Client) SetMinSessions 

func (ce *Client) SetMinSessions(n int)

SetMinSessions updates the minimum session count at runtime. The reconnect loop will use the new value on its next iteration.

func (*Client) SetRelayPaused 

func (ce *Client) SetRelayPaused(p bool)

SetRelayPaused makes the client refuse every new relayed stream while p is true, whatever dmsg.relay_max_streams allows.

func (*Client) SetRelayPeers 

func (ce *Client) SetRelayPeers(pks []cipher.PubKey, port uint16) bool

SetRelayPeers nominates the peers this client should hold a dmsg session through as RELAYS, at the given dmsg port, replacing any earlier nomination. The visor calls it as its transports change: a nominee is a peer it has a live skywire transport to and a reason to trust with its dmsg traffic (a hypervisor, a pinned persistent peer). The serve loop dials nominees FIRST and treats "no relay session while nominees exist" as one session short, so a client already at MinSessions still attaches; a nominee that refuses is backed off for relayFailureBackoff. Stream dials prefer a relay session (see DialStream). Passing an empty set withdraws every nomination; existing relay sessions are left to close on their own. It reports whether the nomination changed.

func (*Client) SetSessionDialer 

func (ce *Client) SetSessionDialer(d SessionDialer)

SetSessionDialer installs (or, with nil, removes) the dialer used for carriers this client cannot dial natively. See Config.SessionDialer. Safe to call before Serve; takes effect for every subsequent session dial.

func (*Client) SweepStalePorterEntries 

func (ce *Client) SweepStalePorterEntries(maxAge time.Duration) int

SweepStalePorterEntries closes and frees ephemeral port entries older than maxAge. Returns the number of entries swept. This is a safety net for streams that leak through the normal cleanup paths.

func (*Client) Traffic added in v1.3.100 

func (ce *Client) Traffic() []PortTraffic

Traffic returns the per-port totals since CountTraffic, by port.

func (*Client) Type 

func (*Client) Type() string

Type returns the client's type (should always be "dmsg").

func (*Client) UpgradeBrowserSessions 

func (ce *Client) UpgradeBrowserSessions() int

UpgradeBrowserSessions is the browser's wss→WebTransport convergence, kept under its original name for the wasm converge ticker. It is now a thin wrapper over ConvergeCarriers, gated on the browser's WT-over-WS preference.

type ClientCallbacks 

type ClientCallbacks struct {
	OnSessionDial       SessionDialCallback
	OnSessionDisconnect SessionDisconnectCallback
}

ClientCallbacks contains callbacks which a Client uses.

type ClientSession 

type ClientSession struct {
	*SessionCommon
	// contains filtered or unexported fields
}

ClientSession represents a session from the perspective of a dmsg client.

func (*ClientSession) Close 

func (cs *ClientSession) Close() error

Close closes the client session and reaps any porter entries for streams belonging to this session. Without this reaping, ephemeral ports stay reserved after the session dies because nothing calls Stream.Close() on the orphaned streams — eventually exhausting the porter's 16k port range. Streams are matched by their underlying SessionCommon pointer because multiple ClientSession value copies can wrap the same SessionCommon.

func (*ClientSession) DialStream 

func (cs *ClientSession) DialStream(ctx context.Context, dst Addr) (dStr *Stream, err error)

DialStream attempts to dial a stream to a remote client via the dmsg server that this session is connected to. The context is used to cancel the dial if the caller's deadline expires — this prevents ephemeral port leaks when many dials are attempted and the caller gives up before the handshake completes.

func (*ClientSession) LookupIP 

func (cs *ClientSession) LookupIP(dst Addr) (myIP net.IP, err error)

LookupIP attempts to dial a stream to the server for the IP address of the client.

func (*ClientSession) LookupIPGeo 

func (cs *ClientSession) LookupIPGeo(dst Addr) (StreamResponse, error)

LookupIPGeo is the same as LookupIP but additionally asks the dmsg server for geolocation data on the resolved IP. Older servers without a configured GeoLookupFunc return zero geo fields — the caller falls back to a local lookup when GeoCountry is empty.

type Config 

type Config struct {
	MinSessions    int
	UpdateInterval time.Duration // Duration between discovery entry updates.

	// NoRegister disables publishing this client's own discovery entry.
	//
	// Registration and resolution are independent axes. Not registering does
	// NOT make a client undialable — anyone who knows (or is seeded with) the
	// servers it sits on can still reach it; it is only unresolvable by lookup.
	// Set it for a read-only consumer with a fixed set of destinations, or for a
	// deployment running without a discovery at all.
	//
	// The alternative it replaces was handing the client a discovery whose
	// writes silently succeed without publishing (direct.NewClient), which left
	// the client believing it had registered.
	NoRegister bool
	// RelayOnly makes a relay session (SetRelayPeers, the skynet carrier) the
	// only session this client holds while one is attached: the serve loop is
	// satisfied by it, the idle reaper trims server sessions to it, and with no
	// server session there is nothing to publish — the discovery entry is
	// deleted. With no relay attached the client behaves as any other: it holds
	// MinSessions server sessions and publishes an entry naming them. A browser
	// visor sets it: attached to its host it rides the relay; off the host's
	// LAN it falls back to servers over wss and is reachable over dmsg.
	RelayOnly            bool
	Callbacks            *ClientCallbacks
	ClientType           string
	ConnectedServersType string
	Protocol             string

	// Carriers is the ordered preference of dmsg CARRIERS — how this client
	// reaches a dmsg server — by name: "tcp", "ws", "wt", "quic". The first
	// listed carrier the server advertises is dialed (falling back to TCP on dial
	// failure). The carrier is transparent above the session: every dmsg stream is
	// the same Noise+yamux regardless, so this only affects how bytes reach the
	// rendezvous server. Empty (the native default) = QUIC when the server
	// advertises it, else TCP — raw sockets are strictly better for native
	// clients. It exists for restrictive networks (e.g. only 443/HTTPS egress →
	// ["wt"] or ["ws"]) and is set to ["ws"] in the js/wasm build, which has no
	// raw socket. Unknown names are ignored.
	Carriers []string

	// SkipSelfServer, when true, makes the serve loop skip any discovered
	// server entry whose PK equals this client's own PK. It exists for the
	// visor that runs a dmsg SERVER in-process under the same PK: without
	// this the client would try to open a transit session to itself. Default
	// false — the standalone dmsg-server's transit client relies on reaching
	// its own PK (see self_session_test.go), so only the co-resident-server
	// visor sets it.
	SkipSelfServer bool

	// SessionDialer dials the byte pipe to a server whose advertised address
	// names a carrier this client has no native dial for — today the skynet
	// carrier: a "skynet://<pk>:<port>" address is a dmsg relay run by a visor
	// (see Client.AcceptRelaySession) reached over a skywire route. The conn it
	// returns is wrapped in the same Noise+yamux session as every other
	// carrier; the pipe is invisible above it. nil leaves such servers
	// undialable (the seeded entry is skipped with an error, like a browser
	// meeting a tcp-only server).
	SessionDialer SessionDialer

	// MaxRelayedStreams bounds the concurrent streams this client relays for
	// sessions accepted through AcceptRelaySession. 0 (the default) refuses
	// every relayed stream: a client is not a relay unless configured as one.
	MaxRelayedStreams int
}

Config configures a dmsg client entity.

func DefaultConfig 

func DefaultConfig() *Config

DefaultConfig returns the default configuration for a dmsg client entity.

func (*Config) Ensure 

func (c *Config) Ensure()

Ensure ensures all config values are set.

type ConnectToAllServersResult 

type ConnectToAllServersResult struct {
	Total            int                      // total servers found in discovery
	AlreadyConnected int                      // sessions already in place before the call
	NewlyConnected   int                      // sessions established by this call
	Failed           map[cipher.PubKey]string // server PK → error text for any that could not be connected
}

ConnectToAllServersResult summarizes the result of a ConnectToAllServers call.

type ConnectionsSummary 

type ConnectionsSummary map[cipher.PubKey][]cipher.PubKey

ConnectionsSummary associates connected clients, and the servers that connect such clients. Key: Client PK, Value: Slice of Server PKs

type DmsghttpConfig 

type DmsghttpConfig struct {
	DmsgServers        []disc.Entry `json:"dmsg_servers"`
	DmsgDiscovery      string       `json:"dmsg_discovery_dmsg"`
	TransportDiscovery string       `json:"transport_discovery_dmsg"`
	AddressResolver    string       `json:"address_resolver_dmsg"`
	RouteFinder        string       `json:"route_finder_dmsg"`
	UptimeTracker      string       `json:"uptime_tracker_dmsg"`
	ServiceDiscovery   string       `json:"service_discovery_dmsg"`
	// SkycoinNode is the deployment's skycoin node, reachable over the mesh — the
	// default coin node for the HV-served wallet (native + wasm). The resolver
	// VHOST form "https://<vhost>.<base32-pk>.dmsg" routes /api through the node's
	// visor port-80 forward (Caddy, --preserve-host) like clearnet; a bare
	// "dmsg://<pk>:6420" hits the node directly (deployments without a web front).
	SkycoinNode string `json:"skycoin_node_dmsg"`
}

DmsghttpConfig is the struct that corresponds to the _dmsg fields of services-config.json 

var Prod DmsghttpConfig

Prod is the production deployment dmsghttp-config.json services

type EntityCommon 

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

EntityCommon contains the common fields and methods for server and client entities.

func (*EntityCommon) InboundSince 

func (c *EntityCommon) InboundSince(t time.Time) bool

InboundSince reports whether any remote peer has opened a stream to this entity since t.

This is positive, zero-cost evidence that the dmsg server is still forwarding inbound streams to us — the same property a self-dial through the server proves, established by a real remote rather than by talking to ourselves. A caller that would otherwise probe its own reachability on a timer can consult this first and only probe during genuine inbound silence.

False means "no evidence", not "unreachable": a visor nobody has contacted is indistinguishable from one nobody can contact, which is exactly the case where an active probe is worth its cost.

func (*EntityCommon) LocalPK 

func (c *EntityCommon) LocalPK() cipher.PubKey

LocalPK returns the local public key of the entity.

func (*EntityCommon) LocalSK 

func (c *EntityCommon) LocalSK() cipher.SecKey

LocalSK returns the local secret key of the entity.

func (*EntityCommon) Logger 

func (c *EntityCommon) Logger() logrus.FieldLogger

Logger obtains the logger.

func (*EntityCommon) MasterLogger 

func (c *EntityCommon) MasterLogger() *logging.MasterLogger

MasterLogger obtains the master logger.

func (*EntityCommon) NoListenerHits 

func (c *EntityCommon) NoListenerHits() []PortHit

NoListenerHits returns a snapshot of inbound stream requests this client rejected because no listener was bound on the destination port (dmsg error 306), keyed by (source PK, destination port), most-recent first.

func (*EntityCommon) RelayOnly 

func (c *EntityCommon) RelayOnly() bool

RelayOnly reports Config.RelayOnly.

func (*EntityCommon) SessionCount 

func (c *EntityCommon) SessionCount() int

SessionCount returns the number of sessions.

func (*EntityCommon) SetCXOKeepaliveHealthyFunc 

func (c *EntityCommon) SetCXOKeepaliveHealthyFunc(fn CXOKeepaliveFunc)

SetCXOKeepaliveHealthyFunc installs (or, with nil, clears) the predicate that reports whether this client's registration-over-CXO keepalive is healthy. See cxoKeepaliveHealthyFn. Safe to call concurrently with the entry-update loop.

func (*EntityCommon) SetDHTBootstrap 

func (c *EntityCommon) SetDHTBootstrap(enabled bool)

SetDHTBootstrap marks this entity as a DHT bootstrap node. When set, the server's discovery entry includes dht_bootstrap=true so visors know to use this server for DHT bootstrapping.

func (*EntityCommon) SetEntryPublishHook 

func (c *EntityCommon) SetEntryPublishHook(fn EntryPublishFunc)

SetEntryPublishHook installs (or, with nil, clears) a callback invoked with the client's canonical discovery entry after each successful registration with the primary discovery. Used by the visor to mirror the entry onto a CXO feed. Safe to call concurrently with the entry-update loop.

The hook runs synchronously on the entry-update goroutine, so it MUST NOT block — a well-behaved installer only hands the entry to a batched/async sink (e.g. treestore.Publisher.Put, which coalesces into the next BatchWindow tick) and returns immediately.

func (*EntityCommon) SetLogger 

func (c *EntityCommon) SetLogger(log logrus.FieldLogger)

SetLogger sets the internal logger. This should be called before we serve.

func (*EntityCommon) SetMasterLogger 

func (c *EntityCommon) SetMasterLogger(mlog *logging.MasterLogger)

SetMasterLogger sets the internal master logger. This should be called before we serve.

func (*EntityCommon) Unpublished 

func (c *EntityCommon) Unpublished() bool

Unpublished reports whether this entity publishes no discovery entry (Config.NoRegister). For `visor state`.

type EntryPublishFunc 

type EntryPublishFunc = func(*disc.Entry)

EntryPublishFunc receives a copy of the client's discovery entry each time it is successfully (re)registered with the primary discovery. See EntityCommon.SetEntryPublishHook and entryPublishHook.

type EntryResolver 

type EntryResolver interface {
	// Name identifies the resolver in logs and metrics.
	Name() string
	// Entry returns the peer's entry. Any error means "I cannot answer",
	// and resolution falls through to the next resolver.
	Entry(ctx context.Context, pk cipher.PubKey) (*disc.Entry, error)
}

EntryResolver answers "which servers bridge to this public key" without a discovery round-trip.

This is the dependency-inversion seam that lets a resolver live OUTSIDE pkg/dmsg. A CXO-backed resolver is the motivating case: pkg/cxo imports pkg/dmsg/dmsg, so the dmsg client can never import CXO — but the visor imports both and can inject one. The discovery already publishes its whole clients-by-server mapping over CXO (FeedDMSGDClientsByServer), so a subscriber holds every entry locally and needs no per-dial lookup at all.

A resolver MUST be fast and local. It is consulted on the dial path ahead of the discovery, and resolverTimeout bounds it precisely so that a resolver which is cold, syncing, or wedged degrades to a discovery lookup instead of stalling the dial. Answering "I don't know" quickly is always better than answering slowly.

Whether to install one is a per-deployment judgment, not a default: the cost is a resident copy of the mapping (~930 client entries on the production deployment today, growing with the network) and the benefit scales with how many distinct peers this client dials. Right for a setup node, a route finder or a hypervisor; wrong for a microcontroller, a browser tab, or a short-lived CLI invocation that dials one known peer.

type Error 

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

Error represents a dmsg-related error.

func (Error) Error 

func (e Error) Error() string

Error implements error

func (Error) Is 

func (e Error) Is(target error) bool

Is reports whether target is the same dmsg error, comparing the code alone. Without it, wrapping (Error.Wrap, used by dialLadderErr to carry the per-attempt cause behind a 202) would defeat every errors.Is(err, dmsg.ErrX) check in the tree: Error is a comparable struct, so the default equality also compares the wrapped `nxt` and a wrapped error would no longer match its own sentinel.

func (Error) Temporary 

func (e Error) Temporary() bool

Temporary implements net.Error

func (Error) Timeout 

func (e Error) Timeout() bool

Timeout implements net.Error

func (Error) Unwrap 

func (e Error) Unwrap() error

Unwrap exposes the wrapped cause so errors.Is/As can also reach past a dmsg error to what actually failed underneath it.

func (Error) Wrap 

func (e Error) Wrap(err error) Error

Wrap wraps an error and returns the new error.

type GeoLookupFunc 

type GeoLookupFunc func(ip net.IP) (country, region string, lat, lon float64)

GeoLookupFunc returns geolocation data for the given IP. Returns zero values when the IP is non-public or the lookup fails — the caller passes the zero values straight through into the response and the dmsg client falls back to a local lookup on its end.

type Listener 

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

Listener listens for remote-initiated streams.

func LocalListener 

func LocalListener(addr Addr) (*Listener, bool)

LocalListener returns the listener a dmsg client in this process holds on addr, so a host visor can hand a service a connection it took in itself.

func (*Listener) Accept 

func (l *Listener) Accept() (net.Conn, error)

Accept accepts a connection: a dmsg stream, or one handed over in process.

func (*Listener) AcceptStream 

func (l *Listener) AcceptStream() (*Stream, error)

AcceptStream accepts a stream connection.

func (*Listener) Addr 

func (l *Listener) Addr() net.Addr

Addr returns the listener's address.

func (*Listener) Close 

func (l *Listener) Close() error

Close closes the listener.

func (*Listener) DmsgAddr 

func (l *Listener) DmsgAddr() Addr

DmsgAddr returns the listener's address in as `dmsg.Addr`.

func (*Listener) Hand 

func (l *Listener) Hand(conn net.Conn, remote Addr) error

Hand queues conn for Accept as a connection from remote, for a host visor that took it in over a skywire transport for a service it runs.

func (*Listener) Type 

func (l *Listener) Type() string

Type returns the stream type.

type LocalRelayDialer 

type LocalRelayDialer struct {
	Network string
	Addr    string
}

LocalRelayDialer dials the local relay acceptor at network/addr ("unix" + socket path, or "tcp" + a loopback address). It is the raw dial, hello and all: see DialLocalRelay and AttachLocalRelay for the two things callers actually do with it.

func (LocalRelayDialer) DialLocalRelay 

func (d LocalRelayDialer) DialLocalRelay(ctx context.Context, expect *cipher.PubKey) (net.Conn, cipher.PubKey, uint16, error)

DialLocalRelay opens one conn to the acceptor and reads its hello, returning the conn positioned at the first byte of the dmsg session along with the acceptor's key and relay port.

func (LocalRelayDialer) PubKey 

func (d LocalRelayDialer) PubKey(ctx context.Context) (cipher.PubKey, uint16, error)

PubKey probes the acceptor for its public key and relay port, then hangs up. A caller needs both before it can nominate the relay (SetRelayPeers takes a key, and the Noise XK session handshake needs it too).

func (LocalRelayDialer) SessionDialer 

func (d LocalRelayDialer) SessionDialer() SessionDialer

SessionDialer returns a Config.SessionDialer that serves the skynet carrier from this local socket instead of from a skywire route. The address it is handed is the usual "skynet://<pk>:<port>", and the PK in it is checked against the acceptor's hello — so a client that nominated visor A cannot be silently attached to visor B by a swapped socket.

It is the whole client-side counterpart: a standalone process installs this (SetSessionDialer), nominates the visor (SetRelayPeers), and the ordinary serve loop does the rest.

type PeerAnnounce 

type PeerAnnounce struct {
	Timestamp int64
	SrcPK     cipher.PubKey // the announcing server's PK
	// contains filtered or unexported fields
}

PeerAnnounce is sent by a dialing dmsg server on its first stream after connecting to another server, declaring "I am a server-peer; you may forward client streams to me over this connection." It lets a non-public server (not in the discovery, never dialed by anyone) become reachable inbound: the accepting server, if it opts in, files the announced session in its peer set, so streams destined for the announcer's clients are forwarded back down the link the announcer dialed out. Signed by the announcer's SK and bound to the noise- authenticated session remote (Verify enforces SrcPK == session remote PK), so it cannot be spoofed onto someone else's session.

func (PeerAnnounce) Verify 

func (a PeerAnnounce) Verify(remoteStatic cipher.PubKey) error

Verify checks the PeerAnnounce: SrcPK must be non-null, must equal the session's noise-authenticated remote PK (so an announce can only claim the PK that handshook this session — anti-spoof), and the signature must be valid for SrcPK.

type PeerEntry 

type PeerEntry struct {
	PK   cipher.PubKey
	Addr string
}

PeerEntry represents a peer dmsg server to connect to.

type PortHit 

type PortHit struct {
	SrcPK     cipher.PubKey `json:"src_pk"`
	DstPort   uint16        `json:"dst_port"`
	Count     uint64        `json:"count"`
	FirstSeen time.Time     `json:"first_seen"`
	LastSeen  time.Time     `json:"last_seen"`
	// contains filtered or unexported fields
}

PortHit summarizes inbound stream requests from a peer to a local port that had NO listener (dmsg error 306, ErrReqNoListener) — i.e. someone tried to reach a port this client isn't serving. Useful for spotting would-be connectors (e.g. managed visors hitting a disabled hypervisor port), misconfigured peers, or port scans.

type PortTraffic added in v1.3.100 

type PortTraffic struct {
	Port    uint16
	Streams uint64
	In      uint64
	Out     uint64
}

PortTraffic is what the streams accepted on one port carried since counting started: In is bytes read from peers, Out bytes written to them.

type Server 

type Server struct {
	EntityCommon
	// contains filtered or unexported fields
}

Server represents a dsmg server entity.

func NewServer 

func NewServer(pk cipher.PubKey, sk cipher.SecKey, dc disc.APIClient, conf *ServerConfig, m metrics.Metrics) *Server

NewServer creates a new dmsg server entity.

func (*Server) AddDiscovery 

func (s *Server) AddDiscovery(client disc.APIClient, advertisedAddr string, discPK cipher.PubKey)

AddDiscovery registers an additional dmsg-discovery this server should publish its entry to. advertisedAddr, when non-empty, overrides the server's default AdvertisedAddr() for THIS discovery only — useful for advertising a LAN address to a local discovery and a public address to an internet discovery. discPK, when set, lets the server recognize an inbound DMSG session originated by this discovery and trigger an immediate registration push.

Safe to call before or after Serve. Subsequent updateServerEntry iterations pick up the new endpoint automatically.

func (*Server) AddDiscoveryDualStack 

func (s *Server) AddDiscoveryDualStack(client disc.APIClient, advertisedAddr, advertisedAddrV6 string, discPK cipher.PubKey)

AddDiscoveryDualStack is the dual-stack variant of AddDiscovery — adds an additional v6 endpoint advertised to this discovery alongside advertisedAddr. Empty advertisedAddrV6 is equivalent to calling AddDiscovery (v4-only for this endpoint). Use this when the same server should expose distinct {v4, v6} addresses to a particular discovery (e.g. LAN-v6 to a local discovery while still v4-public elsewhere).

func (*Server) AdvertiseQUIC added in v1.3.99 

func (s *Server) AdvertiseQUIC(addr string)

AdvertiseQUIC publishes addr as this server's dmsg-over-QUIC endpoint, for a server whose QUIC connections arrive through ServeQUICConn.

func (*Server) AdvertiseWT added in v1.3.99 

func (s *Server) AdvertiseWT(url string, certHash, nextCertHash [32]byte)

AdvertiseWT publishes url, the certificate hash a client pins, and the hash of the certificate after the next rotation, as this server's WebTransport endpoint, for a server whose sessions arrive through ServeWTSession.

func (*Server) AdvertisedAddr 

func (s *Server) AdvertisedAddr() string

AdvertisedAddr returns the TCP address in which the dmsg server is advertised by. This is the TCP address that should be contained within the dmsg discovery entry of this server.

func (*Server) AdvertisedAddrV6 

func (s *Server) AdvertisedAddrV6() string

AdvertisedAddrV6 returns the optional IPv6 TCP address the dmsg server is advertised as. Empty when the server hasn't been configured for IPv6 (the default — preserves the pre-#1525 single-stack contract). Does NOT block on addrDone: v6 advertisement is independent of the primary v4 advertisement and may legitimately remain empty.

func (*Server) Close 

func (s *Server) Close() error

Close implements io.Closer

func (*Server) GetSessions 

func (s *Server) GetSessions() map[cipher.PubKey]*SessionCommon

GetSessions returns underlying sessions map.

func (*Server) IsPeerPK 

func (s *Server) IsPeerPK(pk cipher.PubKey) bool

IsPeerPK reports whether pk is a known peer SERVER rather than a client. Exported so a co-resident visor can separate the two when it reports what its in-process server is holding: a server-to-server link counted as a client makes the server look busier than it is.

func (*Server) QUICTLSConfig added in v1.3.99 

func (s *Server) QUICTLSConfig() (*tls.Config, error)

QUICTLSConfig is the server tls.Config for dmsg-over-QUIC connections that another listener accepts on this server's behalf — a visor's shared QUIC socket, which owns the transport port and dispatches by ALPN. Only the dmsg ALPN: the transport ALPN older clients also offer belongs to squicr there.

func (*Server) Ready 

func (s *Server) Ready() <-chan struct{}

Ready returns a chan which blocks until the server begins serving.

func (*Server) Serve 

func (s *Server) Serve(lis net.Listener, addr string) error

Serve serves the server.

func (*Server) ServeInProcess 

func (s *Server) ServeInProcess()

ServeInProcess lets dmsg clients in this process reach s over an in-memory pipe until s closes.

func (*Server) ServeQUIC 

func (s *Server) ServeQUIC(udpConn net.PacketConn, advertisedUDPAddr string) error

ServeQUIC starts a QUIC listener for dmsg-over-QUIC (#2607) on the given UDP socket and advertises advertisedUDPAddr in discovery (Server.AddressUDP + Protocol "quic"). It runs alongside the TCP Serve: QUIC-capable clients dial QUIC for native stream multiplexing + a datagram channel, everyone else stays on TCP. Blocks until the listener errors or the server closes.

func (*Server) ServeQUICConn added in v1.3.99 

func (s *Server) ServeQUICConn(qc *quic.Conn)

ServeQUICConn serves one QUIC connection accepted by another listener (see QUICTLSConfig) as a dmsg session. That listener may hand the connection over before the handshake completes, and the peer's PK comes from its client certificate, so wait for it first.

func (*Server) ServeUnifiedQUIC 

func (s *Server) ServeUnifiedQUIC(udpConn net.PacketConn, advertisedUDPAddr, advertisedWTURL string) error

ServeUnifiedQUIC serves dmsg-over-QUIC and (when advertisedWTURL is non-empty) dmsg-over-WebTransport on the same udpConn via ONE ALPN-demuxing listener: skywire's QUIC ALPN → native dmsg QUIC sessions (handleQUICConn); "h3" → WebTransport (handleWTSession). WT setup is best-effort — a cert/build failure logs and the server keeps serving QUIC. advertisedWTURL empty makes this behave exactly like ServeQUIC (QUIC only).

func (*Server) ServeWS 

func (s *Server) ServeWS(lis net.Listener, advertisedWSURL string) error

ServeWS serves dmsg over WebSocket on lis and advertises advertisedWSURL in discovery (Server.AddressWS). It runs alongside the TCP Serve and the QUIC ServeQUIC: WS-capable clients (chiefly the js/wasm build) dial the WS URL, everyone else stays on TCP/QUIC. lis is a plaintext HTTP listener — TLS for wss:// is expected to be terminated by a front proxy, with the dmsg Noise layer providing the actual end-to-end PK authentication regardless. Blocks until the listener errors or the server closes.

func (*Server) ServeWTSession added in v1.3.99 

func (s *Server) ServeWTSession(sess *webtransport.Session)

ServeWTSession serves a WebTransport session that another HTTP/3 server accepted on this server's behalf — a visor's shared transport socket, which also serves the visor's own WebTransport transport on another path.

func (*Server) ServeWebTransport 

func (s *Server) ServeWebTransport(udpConn net.PacketConn, advertisedWTURL string, cert *skyquic.RotatingWebTransportCert) error

ServeWebTransport serves dmsg over WebTransport on the given UDP socket and advertises advertisedWTURL with cert's hash in discovery (Server.AddressWT + CertHashWT). It runs alongside TCP/QUIC/WS Serve. cert is the short-lived self-signed certificate a browser pins by its SHA-256 in its WebTransport serverCertificateHashes constructor; it rotates before browsers stop accepting it, and each new hash is advertised as it is made. Blocks until the listener errors or the server closes.

func (*Server) ServeWithWS 

func (s *Server) ServeWithWS(lis net.Listener, advertisedAddr, advertisedWSURL string) error

ServeWithWS serves BOTH raw-dmsg and dmsg-over-WebSocket on a SINGLE listener, so one advertised ip:port carries both protocols — no separate WS port. It demultiplexes by first bytes: a WebSocket client opens with an HTTP/1 upgrade (cmux.HTTP1Fast → ServeWS), a native dmsg client opens with the Noise handshake (cmux.Any → Serve). They are trivially distinguishable because the WS handshake is ASCII HTTP and the dmsg handshake is binary Noise.

The server's discovery entry then advertises Server.Address (TCP) AND Server.AddressWS (the SAME host:port, ws://host:port/dmsg) — no new port. So every server running this becomes reachable by a browser at ws://<address>/dmsg with no topology change: clients just dial the advertised address over whatever protocol they speak. (QUIC + WebTransport share the matching UDP port the same way — see the WT/QUIC listeners.)

advertisedAddr is registered as Server.Address; advertisedWSURL (same host:port, ending in /dmsg) as Server.AddressWS. Blocks until a sub-server returns.

func (*Server) SetAdvertisedAddr 

func (s *Server) SetAdvertisedAddr(lis net.Listener, addr *string)

SetAdvertisedAddr sets the advertised TCP address in which the dmsg server is advertised by. This should only be called once.

func (*Server) SetAdvertisedAddrV6 

func (s *Server) SetAdvertisedAddrV6(addr string)

SetAdvertisedAddrV6 sets the optional IPv6 advertised address. Must be called BEFORE Serve so the first updateServerEntry tick already carries the v6 endpoint. Empty addr is a no-op (leaves the server IPv4-only). Independent of SetAdvertisedAddr — they can be called in either order, and the absence of a v6 SetAdvertisedAddrV6 call is the default "no v6 advertisement" behavior for backward compat.

func (*Server) SetDiscoveryClients 

func (s *Server) SetDiscoveryClients(clients []disc.APIClient)

SetDiscoveryClients replaces the underlying disc.APIClient on each configured discovery endpoint, preserving advertised addresses and PKs. The lengths must match the current endpoint count, otherwise the call is a no-op. Used to upgrade plain HTTP clients to dmsgfirst-wrapped clients once the server's outbound dmsg.Client is available — registration then prefers DMSG over HTTP, which keeps the server reachable from local discoveries when the public internet is unavailable.

func (*Server) SetGeoLookup 

func (s *Server) SetGeoLookup(fn GeoLookupFunc)

SetGeoLookup registers an IP-to-geo lookup function. The function is called by the IP-info stream handler when the request has GeoInfo=true, so dmsg clients can get their public IP and its geolocation in one round-trip. Pass nil to disable. cmd/dmsg/dmsg- server wires the embedded MaxMind DB; pkg/dmsg/dmsg deliberately does not import pkg/geoip to avoid pulling the 62MB embedded DB into every dmsg-using binary.

func (*Server) Stats added in v1.3.100 

func (s *Server) Stats() ServerStats

Stats returns what the server is carrying now and has relayed so far.

func (*Server) WSHandler 

func (s *Server) WSHandler(advertisedWSURL string) http.Handler

WSHandler returns the dmsg-over-WebSocket handler and records advertisedWSURL as this server's AddressWS, without binding a listener of its own.

ServeWS owns a port. A dmsg server folded into a visor does not have one: it shares the visor's transport port, whose HTTP/1 branch already belongs to the WS transport. This lets that branch route the dmsg path here instead, so one port keeps carrying raw dmsg, stcpr, WS transports AND dmsg-over-WS — which is what the standalone server did on its own main port before the fold, and what a browser visor needs to reach this server at all.

type ServerConfig 

type ServerConfig struct {
	MaxSessions    int
	UpdateInterval time.Duration
	AuthPassphrase string
	Peers          []PeerEntry

	// Version, when non-empty, is advertised as this server's discovery
	// entry Version (disc.Entry.Version) in place of the vestigial "0.0.1"
	// protocol constant — nothing checks the entry version for protocol
	// compatibility (only that it is non-empty), so the field is free to
	// carry the server's real build version, which then shows in the
	// discovery /all_servers listing and `skywire cli mdisc servers`. The
	// caller (pkg/services/dmsgsrv) sets it to buildinfo.Version(), keeping
	// pkg/dmsg decoupled from skywire's buildinfo. Empty preserves "0.0.1".
	Version string

	// AcceptedPeerPKs optionally restricts which servers may announce
	// themselves as forwardable peers over an inbound link. Empty (the
	// default) accepts any announcer — every dmsg server is a relay
	// surface. Peer relaying is always on and cannot be disabled; this
	// allowlist only narrows *who* may be filed as a peer.
	AcceptedPeerPKs []cipher.PubKey

	// MaxPeerLinks bounds the number of concurrent peer sessions (inbound
	// announced + configured outbound) so an always-open relay surface
	// can't accumulate unbounded peer links. Zero uses DefaultMaxPeerLinks.
	MaxPeerLinks int
	// MaxRelayedStreams bounds the number of concurrent streams this
	// server relays on behalf of a peer (either direction of a peer
	// bridge), so relaying for others can't be turned into an amplifier.
	// Zero uses DefaultMaxRelayedStreams. Local client↔client bridging on
	// this same server is unaffected — only peer-relayed streams count.
	MaxRelayedStreams int

	// AcceptRelayedRequests lets a CLIENT session carry stream requests whose
	// source key is not the session's own: a visor forwarding a request on
	// another key's behalf over its own session (#4484 stage 3, dmsg over
	// skynet through a host). The request is still signed by, and verified
	// against, its source key, so the relaying session cannot impersonate
	// anyone; what it spends is this server's relay capacity, charged like a
	// peer bridge. Peer sessions carry foreign sources regardless.
	AcceptRelayedRequests bool
}

ServerConfig configues the Server

func DefaultServerConfig 

func DefaultServerConfig() *ServerConfig

DefaultServerConfig returns the default server config.

type ServerSession 

type ServerSession struct {
	*SessionCommon
	// contains filtered or unexported fields
}

ServerSession represents a session from the perspective of a dmsg server.

func (*ServerSession) Close 

func (ss *ServerSession) Close() error

Close implements io.Closer

func (*ServerSession) Serve 

func (ss *ServerSession) Serve()

Serve serves the session.

type ServerStats added in v1.3.100 

type ServerStats struct {
	// ClientSessions are the clients connected to this server.
	ClientSessions int
	// PeerSessions are this server's links to other servers.
	PeerSessions int
	// ActiveStreams are streams being relayed now.
	ActiveStreams int64
	// StreamsRelayed counts every stream relayed since the server started.
	StreamsRelayed uint64
	// BytesUp is what relayed streams carried from the side that opened them,
	// BytesDown what they carried back.
	BytesUp, BytesDown uint64
}

ServerStats is what a dmsg server is carrying.

type SessionCarrier 

type SessionCarrier struct {
	ServerPK  cipher.PubKey `json:"server_pk"`
	Carrier   string        `json:"carrier"`
	Addr      string        `json:"addr"`
	Preferred string        `json:"preferred,omitempty"`
	Note      string        `json:"note,omitempty"` // why the last converge attempt didn't reach Preferred
}

SessionCarrier summarizes one live dmsg-server session's carrier, for the convergence CLI/RPC. Preferred is the most-preferred carrier the server currently advertises per the client's ordered Carriers preference (empty if no preference is set or none is advertised).

type SessionCommon 

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

SessionCommon contains the common fields and methods used by a session, whether it be it from the client or server perspective.

func (*SessionCommon) Carrier 

func (sc *SessionCommon) Carrier() string

Carrier reports how this client session's byte pipe to the dmsg server was dialed: one of CarrierTCP / CarrierWS / CarrierWT / CarrierQUIC. It is empty for accepted (server-side) sessions. Use Protocol for a human-readable label that also distinguishes ws from wss.

func (*SessionCommon) CarrierAddr 

func (sc *SessionCommon) CarrierAddr() string

CarrierAddr reports the endpoint this session's carrier dialed. Empty for accepted (server-side) sessions.

func (*SessionCommon) Close 

func (sc *SessionCommon) Close() error

Close closes the session.

func (*SessionCommon) GetConn 

func (sc *SessionCommon) GetConn() net.Conn

GetConn returns underlying TCP `net.Conn`.

func (*SessionCommon) GetDecNonce 

func (sc *SessionCommon) GetDecNonce() uint64

GetDecNonce returns value of DecNonce of underlying `*noise.Noise`, or 0 when there is no noise state to read (see NonceValues).

func (*SessionCommon) GetEncNonce 

func (sc *SessionCommon) GetEncNonce() uint64

GetEncNonce returns value of EncNonce of underlying `*noise.Noise`, or 0 when there is no noise state to read (see NonceValues).

func (*SessionCommon) LastPing 

func (sc *SessionCommon) LastPing() time.Duration

func (*SessionCommon) LocalPK 

func (sc *SessionCommon) LocalPK() cipher.PubKey

LocalPK returns the local public key of the session.

func (*SessionCommon) LocalTCPAddr 

func (sc *SessionCommon) LocalTCPAddr() net.Addr

LocalTCPAddr returns the local address of the underlying TCP connection.

func (*SessionCommon) NonceValues 

func (sc *SessionCommon) NonceValues() (dec, enc uint64, ok bool)

NonceValues returns the session's decryption and encryption nonces together, and whether they could be read at all.

`ns` is only assigned once a handshake completes, so a session can be reachable through the entity's session map — GetSessions returns a snapshot — while having no noise state yet, or after being torn down. Callers that merely observe sessions (the server's throughput metrics) must be able to tell "no reading available" apart from "the nonce is 0": treating the second as the first made a counter appear to run backwards, which the throughput calculation reads as a uint64 overflow and turns into an enormous bogus delta. Reading them separately would also let a session be torn down between the two calls; this takes both under their locks in one go.

func (*SessionCommon) NumStreams 

func (sc *SessionCommon) NumStreams() int

NumStreams reports the number of open mux streams on this session (0 = idle; -1 = unmeasurable, treated as busy). Used by the idle-session reaper.

func (*SessionCommon) Ping 

func (sc *SessionCommon) Ping() (time.Duration, error)

Ping obtains the round trip latency of the session.

func (*SessionCommon) PingFails 

func (sc *SessionCommon) PingFails() int

LastPing returns the last measured round-trip latency. Returns 0 if no measurement has been taken yet. PingFails is the consecutive liveness-ping failure count for this session.

func (*SessionCommon) Protocol 

func (sc *SessionCommon) Protocol() string

Protocol renders a human-readable label for the protocol this session used to reach its dmsg server, distinguishing plain ws:// from secured wss://.

func (*SessionCommon) ReadDatagram 

func (sc *SessionCommon) ReadDatagram(ctx context.Context) ([]byte, error)

ReadDatagram receives an unreliable datagram from the session's QUIC connection. Returns ErrDatagramsUnsupported on non-QUIC sessions.

func (*SessionCommon) ReadSince 

func (sc *SessionCommon) ReadSince(t time.Time) bool

ReadSince reports whether any stream on this session has read bytes since t.

Those bytes came through the dmsg server, which is the very thing the liveness ping goes and asks about — so this is the same evidence, produced by real traffic instead of by a probe, and it cannot be wrong in the direction that matters: data does not arrive through a server that has stopped relaying for us.

False means "no evidence", not "dead". An idle session reads nothing and is exactly the case the ping exists for.

func (*SessionCommon) RemotePK 

func (sc *SessionCommon) RemotePK() cipher.PubKey

RemotePK returns the remote public key of the session.

func (*SessionCommon) RemoteTCPAddr 

func (sc *SessionCommon) RemoteTCPAddr() net.Addr

RemoteTCPAddr returns the remote address of the underlying TCP connection.

func (*SessionCommon) SetLastPing 

func (sc *SessionCommon) SetLastPing(d time.Duration)

SetLastPing records a latency measurement.

func (*SessionCommon) SupportsDatagrams 

func (sc *SessionCommon) SupportsDatagrams() bool

SupportsDatagrams reports whether this session can carry unreliable datagrams — true only for QUIC sessions (#2607 dmsg-over-QUIC).

func (*SessionCommon) WriteDatagram 

func (sc *SessionCommon) WriteDatagram(b []byte) error

WriteDatagram sends an unreliable datagram over the session's QUIC connection (RFC 9221) — UDP-over-dmsg. Returns ErrDatagramsUnsupported on non-QUIC sessions. Datagrams ride the same PK-bound-TLS-encrypted QUIC connection as the session's streams; this gives dmsg a genuine datagram channel that the reliable yamux/smux mux cannot.

type SessionDialCallback 

type SessionDialCallback func(network, addr string) (err error)

SessionDialCallback is triggered BEFORE a session is dialed to. If a non-nil error is returned, the session dial is instantly terminated.

type SessionDialer 

type SessionDialer func(ctx context.Context, network, addr string) (net.Conn, error)

SessionDialer dials the byte pipe for a session over a carrier the client cannot dial natively (Config.SessionDialer). network is the carrier name (CarrierSkynet) and addr the server entry's advertised address.

type SessionDisconnectCallback 

type SessionDisconnectCallback func(network, addr string, err error)

SessionDisconnectCallback triggers after a session is closed.

type SessionManager 

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

SessionManager holds the active multiplexer for a session. Exactly one of yamux / smux / quic is non-nil. yamux and smux multiplex streams over a single reliable conn (TCP); quic is a QUIC connection whose native streams ARE the dmsg streams (no separate mux) and which also carries an unreliable datagram channel — #2607 dmsg-over-QUIC.

func (*SessionManager) NumStreams 

func (sm *SessionManager) NumStreams() int

NumStreams reports the number of open mux streams on this session. Returns -1 when the count is unavailable — a quic session's native streams aren't cheaply countable, and a not-yet-established manager has no mux — which the idle-session reaper treats as "busy" so it never reaps a session it can't measure.

type SignedObject 

type SignedObject []byte

SignedObject represents a gob-encoded structure prepended with a signature.

func MakeSignedPeerAnnounce 

func MakeSignedPeerAnnounce(a *PeerAnnounce, sk cipher.SecKey) (SignedObject, error)

MakeSignedPeerAnnounce encodes and signs a PeerAnnounce into a SignedObject.

func MakeSignedStreamRequest 

func MakeSignedStreamRequest(req *StreamRequest, sk cipher.SecKey) (SignedObject, error)

MakeSignedStreamRequest encodes and signs a StreamRequest into a SignedObject format.

func MakeSignedStreamResponse 

func MakeSignedStreamResponse(resp *StreamResponse, sk cipher.SecKey) (SignedObject, error)

MakeSignedStreamResponse encodes and signs a StreamResponse into a SignedObject format.

func (SignedObject) Hash 

func (so SignedObject) Hash() cipher.SHA256

Hash returns the hash of the SignedObject.

func (SignedObject) Object 

func (so SignedObject) Object() []byte

Object returns the bytes of the SignedObject that contain the encoded object.

func (SignedObject) ObtainPeerAnnounce 

func (so SignedObject) ObtainPeerAnnounce() (PeerAnnounce, error)

ObtainPeerAnnounce obtains a PeerAnnounce from the encoded object bytes.

func (SignedObject) ObtainStreamRequest 

func (so SignedObject) ObtainStreamRequest() (StreamRequest, error)

ObtainStreamRequest obtains a StreamRequest from the encoded object bytes.

func (SignedObject) ObtainStreamResponse 

func (so SignedObject) ObtainStreamResponse() (StreamResponse, error)

ObtainStreamResponse obtains a StreamResponse from the encoded object bytes.

func (SignedObject) Sig 

func (so SignedObject) Sig() cipher.Sig

Sig returns the prepended signature section of the SignedObject.

func (SignedObject) Valid 

func (so SignedObject) Valid() bool

Valid returns true if the SignedObject has a valid length.

type Stream 

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

Stream represents a dmsg connection between two dmsg clients.

Exactly one of yStr / sStr / qStr is non-nil — the underlying mux stream that carries this dmsg stream (yamux, smux, or a native QUIC stream for #2607 dmsg-over-QUIC). The end-to-end client↔client Noise (ns/nsConn) rides on top of whichever it is, so client↔client confidentiality through the relay is identical across transports.

func (*Stream) Carrier 

func (s *Stream) Carrier() string

Carrier reports how the stream's session reaches its dmsg server, such as "skynet" for a session carried over a skywire transport.

func (*Stream) Close 

func (s *Stream) Close() error

Close closes the dmsg stream.

func (*Stream) LocalAddr 

func (s *Stream) LocalAddr() net.Addr

LocalAddr returns the local address of the dmsg stream.

func (*Stream) Logger 

func (s *Stream) Logger() logrus.FieldLogger

Logger returns the internal logrus.FieldLogger instance.

func (*Stream) RawLocalAddr 

func (s *Stream) RawLocalAddr() Addr

RawLocalAddr returns the local address as dmsg.Addr type.

func (*Stream) RawRemoteAddr 

func (s *Stream) RawRemoteAddr() Addr

RawRemoteAddr returns the remote address as dmsg.Addr type.

func (*Stream) Read 

func (s *Stream) Read(b []byte) (int, error)

Read implements io.Reader. It refreshes the idle timeout on each successful read so that active streams are never killed, while stale streams stuck in waitRead with no incoming data will time out and release their ephemeral port.

On a terminal read error (EOF, remote close, deadline exceeded), we call s.close (the porter freer) to release the ephemeral port immediately, without waiting for the caller to call Close(). This is necessary because rpc.Client.input() does not call codec.Close() when the remote side closes the stream — it just marks the client as shutdown — so the port would otherwise leak until the owning session dies (which may be hours or days on a long-lived visor). Auto-releasing on read error is safe because a dmsg stream is single-use: once Read fails, the stream is dead and the caller must either close it or discard it. Releasing the port is idempotent (sync.Once in makePortFreer), and we keep the underlying yamux stream intact for the caller's Close() to handle.

func (*Stream) RemoteAddr 

func (s *Stream) RemoteAddr() net.Addr

RemoteAddr returns the remote address of the dmsg stream.

func (*Stream) RemotePK 

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

RemotePK returns the public key of the client at the other end.

Established by the end-to-end client<->client Noise handshake, not asserted by the peer, so a caller may use it to authenticate one. appnet's directConn offers the same method, which lets a transport-agnostic consumer ask either carrier who it is talking to.

func (*Stream) ServerPK 

func (s *Stream) ServerPK() cipher.PubKey

ServerPK returns the remote PK of the dmsg.Server used to relay frames to and from the remote client.

func (*Stream) SetDeadline 

func (s *Stream) SetDeadline(t time.Time) error

SetDeadline implements net.Conn

func (*Stream) SetReadDeadline 

func (s *Stream) SetReadDeadline(t time.Time) error

SetReadDeadline implements net.Conn

func (*Stream) SetWriteDeadline 

func (s *Stream) SetWriteDeadline(t time.Time) error

SetWriteDeadline implements net.Conn

func (*Stream) StreamID 

func (s *Stream) StreamID() uint32

StreamID returns the stream ID.

func (*Stream) Write 

func (s *Stream) Write(b []byte) (int, error)

Write implements io.Writer. Like Read, a terminal write error means the stream is dead, so we release the porter reservation immediately to prevent a leak when the caller doesn't follow up with Close().

type StreamCensusRow added in v1.3.100 

type StreamCensusRow struct {
	Initiator bool    `json:"initiator"`
	Port      uint16  `json:"port"`
	State     string  `json:"state"`
	Count     int     `json:"count"`
	OldestS   float64 `json:"oldest_s"`
}

StreamCensusRow counts the streams still in memory that share a direction, a service port and a lifecycle state.

func StreamCensus added in v1.3.100 

func StreamCensus() []StreamCensusRow

StreamCensus drops the entries of collected streams and groups the rest.

type StreamRequest 

type StreamRequest struct {
	Timestamp int64
	SrcAddr   Addr
	DstAddr   Addr
	IPinfo    bool
	// GeoInfo, when set, asks the destination dmsg-server to also
	// populate the geo fields on StreamResponse using its own embedded
	// MaxMind database. Only meaningful when IPinfo is also set, since
	// the geo lookup runs on the same TCP source IP. Older
	// dmsg-servers ignore this flag and return zero geo fields, so the
	// client must be prepared to fall back to a local lookup when the
	// response's geo fields are empty.
	GeoInfo  bool
	NoiseMsg []byte
	// contains filtered or unexported fields
}

StreamRequest represents a stream dial request object.

func (StreamRequest) Verify 

func (req StreamRequest) Verify(lastTimestamp int64) error

Verify verifies the StreamRequest.

type StreamResponse 

type StreamResponse struct {
	ReqHash  cipher.SHA256 // Hash of associated dial request.
	Accepted bool          // Whether the request is accepted.
	IP       net.IP        // IP address of the node.
	// Geo fields are populated only when the request had both
	// IPinfo=true and GeoInfo=true and the dmsg-server has a geo
	// lookup function configured (cmd/dmsg/dmsg-server wires the
	// embedded MaxMind DB; other dmsg.Server users may leave it
	// unset, in which case these fields stay zero). The client
	// falls back to a local lookup when GeoCountry is empty.
	GeoCountry string  // ISO country code
	GeoRegion  string  // ISO region/subdivision code
	GeoLat     float64 // latitude
	GeoLon     float64 // longitude
	ErrCode    errorCode
	NoiseMsg   []byte
	// contains filtered or unexported fields
}

StreamResponse is the response of a StreamRequest.

func (StreamResponse) Verify 

func (resp StreamResponse) Verify(req StreamRequest) error

Verify verifies the StreamResponse.

Source Files

View all Source files

Directories

Path Synopsis
Package metrics pkg/dmsg/dmsg/metrics/delta.go c1-net-dmsg
 Package metrics pkg/dmsg/dmsg/metrics/delta.go c1-net-dmsg

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.