peer

package
v0.0.0-...-cae741c Latest Latest
Warning

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

 Go to latest Published: Jul 27, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Overview

Package peer implements persistent QUIC peer connections: shared UDP transports, connection-scoped Hello, roster coordination, and discovery.

Index

Constants

View Source 
const (
	DefaultDiscoveryConcurrency = 32
	DefaultDiscoveryInterval    = 10 * time.Second
	DefaultDialTimeout          = 30 * time.Second
)

Default discovery tuning.

View Source 
const ALPN = "tailsync"

ALPN is the TLS NextProto for tailsync peer sessions. Required by quic-go; not a wire-protocol version (that is proto.Version).

View Source 
const DefaultMaxIncomingStreams = 32

DefaultMaxIncomingStreams allows concurrent op streams (notify, pull, heartbeat). Inbound heavy serve work is also capped by the daemon serveSem.

Variables

View Source 
var ErrDial = errors.New("dial")

ErrDial marks outbound dial-phase failures (timeouts, refused, already connected, …).

Functions

func IsSoftStreamErr 

func IsSoftStreamErr(err error) bool

IsSoftStreamErr reports caller-side cancel/timeout that should not tear down the peer connection (heartbeat handles soft unreachability).

func PreferredDialer 

func PreferredDialer(localID, remoteID string) string

PreferredDialer returns the node ID that should initiate the connection between local and remote (lexicographically smaller). Shared by inbound decide and simultaneous-dial race resolution so the rule cannot drift.

Types

type Candidate 

type Candidate struct {
	// Addr is host:port.
	Addr string
	// NodeID is optional identity when already known.
	NodeID string
}

Candidate is a dial target from status, pins, or config.

type Config 

type Config struct {
	NodeID string
	Port   int

	ServerTLS *tls.Config
	ClientTLS *tls.Config
	Logger    *slog.Logger

	DiscoveryConcurrency int
	DiscoveryInterval    time.Duration
	DialTimeout          time.Duration
	HandshakeTimeout     time.Duration
	HeartbeatInterval    time.Duration
	HeartbeatTimeout     time.Duration

	// Candidates returns dial targets (status Online, pins). Required for discovery.
	Candidates func(ctx context.Context) []Candidate
}

Config configures the peer manager.

type Discovery 

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

Discovery continuously dials candidates that are not yet connected. The in-flight semaphore is released before backoff sleep so other peers can dial while a failed address waits. Ticks are single-flighted; Kick is non-blocking.

func (*Discovery) BackoffStreak 

func (d *Discovery) BackoffStreak(addr string) int

BackoffStreak returns the failure streak for addr (tests).

func (*Discovery) ClearBackoff 

func (d *Discovery) ClearBackoff(addr string)

ClearBackoff clears backoff for addr (e.g. after successful session).

func (*Discovery) InBackoff 

func (d *Discovery) InBackoff(addr string) bool

InBackoff reports whether addr is currently backed off (tests).

func (*Discovery) Kick 

func (d *Discovery) Kick()

Kick requests an immediate discovery pass without blocking the caller. Coalesces with an in-progress or already-queued tick.

func (*Discovery) Run 

func (d *Discovery) Run(ctx context.Context)

Run watches candidates until ctx is cancelled. Single tick loop so Kick and the interval cannot stack concurrent ticks beyond one in-flight + one pending.

func (*Discovery) SoftFailAddr 

func (d *Discovery) SoftFailAddr(addr string)

SoftFailAddr records a failure from outside discovery (e.g. session drop).

type DiscoveryConfig 

type DiscoveryConfig struct {
	Concurrency      int
	Interval         time.Duration
	DialTimeout      time.Duration
	HandshakeTimeout time.Duration
	Candidates       func(ctx context.Context) []Candidate
	DialPeer         func(ctx context.Context, c Candidate) error
	Log              *slog.Logger
}

DiscoveryConfig tunes discovery.

type Endpoint 

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

Endpoint owns one or more shared UDP PacketConns, each wrapped in a quic-go Transport so Listen and Dial share the same socket. Sessions must never close the PacketConn; only Endpoint.Close does.

func NewEndpoint 

func NewEndpoint(serverTLS, clientTLS *tls.Config, log *slog.Logger) (*Endpoint, error)

NewEndpoint builds an endpoint. serverTLS must include certificates; clientTLS is used for dials (typically InsecureSkipVerify + same ALPN).

func (*Endpoint) Accept 

func (e *Endpoint) Accept(ctx context.Context) (*quic.Conn, error)

Accept returns the next inbound QUIC connection.

func (*Endpoint) AddPacketConn 

func (e *Endpoint) AddPacketConn(pc net.PacketConn) error

AddPacketConn registers pc for shared listen+dial and starts accepting QUIC connections. The endpoint owns pc and closes it on Close.

func (*Endpoint) AddrsString 

func (e *Endpoint) AddrsString() string

AddrsString joins local addresses for logging.

func (*Endpoint) Close 

func (e *Endpoint) Close() error

Close stops accepting, closes transports and owned PacketConns.

func (*Endpoint) Dial 

func (e *Endpoint) Dial(ctx context.Context, addr string) (*quic.Conn, error)

Dial opens a QUIC connection to addr using a family-matched shared transport. The returned connection does not own the PacketConn.

func (*Endpoint) DialAddr 

func (e *Endpoint) DialAddr(ctx context.Context, raddr net.Addr) (*quic.Conn, error)

DialAddr dials a pre-resolved UDP address (e.g. after tsnet name resolution).

func (*Endpoint) LocalAddrs 

func (e *Endpoint) LocalAddrs() []string

LocalAddrs returns bound listen addresses.

type Manager 

type Manager struct {

	// ResolveAddr optionally maps host:port to a concrete net.Addr (tsnet).
	// When nil, Endpoint.Dial parses the address directly.
	ResolveAddr func(ctx context.Context, addr string) (net.Addr, error)

	// VerifyRemoteID optionally validates a Hello NodeID against the remote
	// UDP address (e.g. Tailscale WhoIs). When set, mismatched claims are
	// rejected. Plain/local tests leave this nil.
	VerifyRemoteID func(ctx context.Context, remoteAddr, claimedNodeID string) error

	// OnStream is invoked for each inbound application stream after the first
	// message has been decoded (Ping is handled internally).
	OnStream StreamHandler

	// OnPeerUp is called after a session is installed (inbound or outbound).
	OnPeerUp func(s *Session)
	// OnPeerDown is called when a session is removed from the roster.
	OnPeerDown func(nodeID, addr string)
	// contains filtered or unexported fields
}

Manager owns the endpoint, roster, discovery, and connection lifecycle.

func NewManager 

func NewManager(cfg Config) (*Manager, error)

NewManager builds a manager (does not start accept/discovery).

func (*Manager) AddPacketConn 

func (m *Manager) AddPacketConn(pc net.PacketConn) error

AddPacketConn registers a shared UDP socket for listen+dial.

func (*Manager) Close 

func (m *Manager) Close() error

Close stops loops, closes sessions, and shuts down the endpoint.

func (*Manager) Discovery 

func (m *Manager) Discovery() *Discovery

Discovery returns the discovery service (for tests / Kick).

func (*Manager) Endpoint 

func (m *Manager) Endpoint() *Endpoint

Endpoint returns the shared transport endpoint.

func (*Manager) KickDiscovery 

func (m *Manager) KickDiscovery()

KickDiscovery requests a non-blocking discovery pass (coalesced).

func (*Manager) Roster 

func (m *Manager) Roster() *Roster

Roster returns the session roster.

func (*Manager) Session 

func (m *Manager) Session(nodeID string) *Session

Session returns the session for nodeID, if any.

func (*Manager) Snapshot 

func (m *Manager) Snapshot() []SessionInfo

Snapshot returns connected peers.

func (*Manager) Start 

func (m *Manager) Start(ctx context.Context)

Start begins accept and discovery loops. Call after adding PacketConns.

type Roster 

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

Roster maps nodeID → at most one live Session. Concurrent dials and accepts are coordinated with deterministic node-ID race rules.

func NewRoster 

func NewRoster() *Roster

NewRoster returns an empty roster.

func (*Roster) CloseAll 

func (r *Roster) CloseAll()

CloseAll closes and clears every session.

func (*Roster) ConnectedAddrs 

func (r *Roster) ConnectedAddrs() map[string]struct{}

ConnectedAddrs returns dial addresses of healthy sessions.

func (*Roster) ConnectedNodeIDs 

func (r *Roster) ConnectedNodeIDs() map[string]struct{}

ConnectedNodeIDs returns node IDs with a non-closed session.

func (*Roster) Get 

func (r *Roster) Get(nodeID string) *Session

Get returns the live session for nodeID, or nil.

func (*Roster) Install 

func (r *Roster) Install(sess *Session) (result installResult, old *Session)

Install attempts to register sess for its NodeID.

Rules:

  • No existing → accept
  • Existing unhealthy (or closed) → replace (discard old)
  • Existing healthy → reject new (AlreadyConnected), unless simultaneous-dial race: keep the connection initiated by the lexicographically smaller node ID

On replace, the previous session is Closed asynchronously by the caller via the returned old session pointer.

func (*Roster) Len 

func (r *Roster) Len() int

Len returns the number of tracked sessions (may include closing).

func (*Roster) RemoveIfCurrent 

func (r *Roster) RemoveIfCurrent(sess *Session) bool

RemoveIfCurrent removes sess only if it is still the mapped session.

func (*Roster) Snapshot 

func (r *Roster) Snapshot() []SessionInfo

Snapshot returns a copy of connected session metadata.

func (*Roster) String 

func (r *Roster) String() string

String for debugging.

type Session 

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

Session is one persistent QUIC connection to a peer after a successful Hello.

func SessionForTest 

func SessionForTest(nodeID, localID string) *Session

SessionForTest builds a healthy session with no QUIC connection for unit tests that exercise Healthy/Closed/Invalidate without a live transport.

func (*Session) Addr 

func (s *Session) Addr() string

func (*Session) Close 

func (s *Session) Close()

Close tears down the QUIC connection (not the shared PacketConn). Idempotent. Waits for in-flight stream handlers after canceling accept so callers (Manager.Close) do not race handler use of the session.

closed is set under mu before Wait so the accept loop cannot Add after Wait: spawnStreamHandler checks closed under the same mutex before Add.

func (*Session) Closed 

func (s *Session) Closed() bool

func (*Session) Conn 

func (s *Session) Conn() *quic.Conn

func (*Session) Healthy 

func (s *Session) Healthy() bool

func (*Session) Invalidate 

func (s *Session) Invalidate()

Invalidate marks the session unhealthy and closes it so discovery can replace it. Idempotent. Callers should not use this for soft cancel/timeout (see IsSoftStreamErr); use it for mid-stream framing/connection failures.

func (*Session) IsDialer 

func (s *Session) IsDialer() bool

func (*Session) LocalID 

func (s *Session) LocalID() string

func (*Session) NodeID 

func (s *Session) NodeID() string

func (*Session) OpenStream 

func (s *Session) OpenStream(ctx context.Context) (net.Conn, error)

OpenStream opens a bidirectional application stream for one op. Context cancel/deadline on the open call does not tear down the session (short notify timeouts must not kill healthy peers). Connection-level failures mark unhealthy and Close so discovery can replace the session.

Heartbeat pings use a separate internal open path (see ping) that does not call OpenStream: ping failures always Invalidate the session, while OpenStream preserves soft caller cancel/timeout.

func (*Session) Since 

func (s *Session) Since() time.Time

type SessionConfig 

type SessionConfig struct {
	NodeID  string
	LocalID string
	Addr    string
	Dialer  bool
	Conn    *quic.Conn
	Log     *slog.Logger
	OnClose func(*Session)
}

SessionConfig configures a new session after QUIC + Hello succeed.

type SessionInfo 

type SessionInfo struct {
	NodeID  string
	Addr    string // dial-back host:port when known
	Dialer  bool   // true if we initiated the connection
	Healthy bool
	Since   time.Time
	Remote  string // raw remote UDP address
}

SessionInfo is a snapshot of a connected peer (safe for callers without locks).

type StreamHandler 

type StreamHandler func(ctx context.Context, s *Session, first proto.Message, stream net.Conn)

StreamHandler is invoked for each inbound application stream after the first message has been decoded. Ping is handled internally and never reaches the handler.

Lifecycle: the peer accept path always closes stream when the handler returns (see handleInboundStream). Handlers may Close earlier when finished; Close is idempotent (streamConn closeOnce). Do not use stream after the handler returns.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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