link

README

link

github.com/nfltr/link multiplexes many byte-stream sessions, each a net.Conn with its own flow control, over connections you dial and secure (plain TCP, TLS, mutual TLS). A slow reader stalls only its own session, a dead peer is noticed within a bound you pick, and the main module imports only the Go standard library (Apache-2.0).

It is for Go programs that run many streams between two endpoints (proxies, tunnels, relays, RPC fabrics) and want per-stream backpressure and bounded dead-peer detection without building them. Status: v0.10.12, pre-1.0 (Status and compatibility). In the nfltr stack: link is the bottom layer. nfltr/relay (SDK and protocol) carries its agent data plane on link, nfltr/relay-server is the relay that speaks it, and nfltr is the product on top; link stands alone and does not import any of them. Docs: this file, WIRE.md, docs/, CHANGELOG.md. Checks: make ci (Checks).

Quick start

Two peers in one program: a dials b over plaintext TCP, then each opens a session to the other on that one connection. It needs Go 1.27 or later (go.mod). Run it with go run github.com/nfltr/link/examples/quickstart@latest, or save it as main.go in a module of your own (go get github.com/nfltr/link@latest):

// Command quickstart runs two peers, a and b, in one process: a dials b
// over plaintext TCP, then each opens a session to the other on that one
// connection.
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"time"

	"github.com/nfltr/link"
	"github.com/nfltr/link/peerlink"
)

func main() {
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		log.Fatal(err)
	}

	// b listens. NameAuth: each peer sends its name and is known by it
	// (a trusted network only; see docs/recipes.md for mutual TLS).
	b, err := peerlink.New(peerlink.Config{
		Listener: ln,
		Link:     link.Config{Auth: link.NameAuth("b")},
	})
	if err != nil {
		log.Fatal(err)
	}
	go b.Serve(reply("b"))

	// a dials b, and keeps a connection to it (Connect) so that b can
	// open sessions to a as well.
	a, err := peerlink.New(peerlink.Config{
		Link:    link.Config{Auth: link.NameAuth("a")},
		Connect: []string{"b"},
		Dial: func(ctx context.Context, peer string) (net.Conn, error) {
			var d net.Dialer
			return d.DialContext(ctx, "tcp", ln.Addr().String()) // peer is "b"
		},
	})
	if err != nil {
		log.Fatal(err)
	}
	go a.Serve(reply("a"))

	fmt.Println(ask(a, "b")) // a opens a session to b
	fmt.Println(ask(b, "a")) // b opens one to a, on the connection a dialled

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_ = a.Drain(ctx)
	_ = b.Drain(ctx)
}

// reply serves each session a peer opens: it reads the request and
// answers it.
func reply(me string) func(peer string, header []byte, s peerlink.Session) {
	return func(peer string, header []byte, s peerlink.Session) {
		defer s.Close()
		req, _ := io.ReadAll(s) // until the opener's CloseWrite
		fmt.Fprintf(s, "%s got %q (%s) from %s", me, req, header, peer)
	}
}

// ask opens a session to peer, sends a request and reads the answer.
func ask(l peerlink.Link, peer string) string {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	s, err := l.Open(ctx, peer, []byte("greeting"))
	if err != nil {
		log.Fatal(err)
	}
	defer s.Close()
	if _, err := io.WriteString(s, "hello"); err != nil {
		log.Fatal(err)
	}
	// The peer reads io.EOF after "hello"; this side can still read.
	if err := s.CloseWrite(); err != nil {
		log.Fatal(err)
	}
	answer, err := io.ReadAll(s)
	if err != nil {
		log.Fatal(err)
	}
	return string(answer)
}

It prints:

b got "hello" (greeting) from a
a got "hello" (greeting) from b

In two processes, one runs b's half (net.Listen, a Node with Listener, Serve) and the other a's (a Node whose Dial reaches b's address, with Connect: []string{"b"}, and Serve: b opens sessions to a only on the connection Connect keeps up, and only Serve takes them); either then opens sessions to the other. examples/twoproc does that:

go run github.com/nfltr/link/examples/twoproc@latest -role b &   # listens on -addr (127.0.0.1:7000)
go run github.com/nfltr/link/examples/twoproc@latest -role a     # dials it; exits once both sessions ran
kill %1

For mutual TLS (examples/mtls), gRPC or HTTP over a link, liveness, draining, reconnects and metrics, see the recipes.

Chaining

Since v0.10.0 sessions join end to end: a path origin -> hop -> ... -> far end, each arrow a connection of its own (its own TLS, liveness and credit). link.Forward(route, cfg) is an accept function that makes a Mux (or Pool) a hop: your Route opens the next leg (on any Mux, Pool or peer you choose), Forward passes the far end's answer back and splices the two sessions with link.Join, holding no goroutine while they are idle. Where both ends of a connection announce FeatureChain (every v0.10.0 side does; WIRE.md section 12), each open is answered: Session.Answered returns nil only once the far end accepted (Accept, or its first write), or the *link.Abort of a refusal (Refuse(code, msg)) or reset (Abort(code, msg)): its Code (link's below CodeApp, yours from it), its Msg, and Hops, which hop it came from (0: your peer). A dead hop reaches the origin as CodeConnLost from the hop before it; a draining hop refuses new opens (CodeDraining) while what it joined finishes, and cuts the rest at its deadline (CodeDrainDeadline); a routing loop stops at ForwardConfig.MaxHops (CodeHopLimit). A connection to a v0.9 peer is unchanged: its sessions do not answer (Answers() false, Answered returns ErrNoAnswers), and a hop in front of one accepts once its Route returned.

// ExampleForward joins two links end to end: the origin opens through a
// forwarding hop to a far end, which answers each open; the origin reads
// the answer (accepted, with the path's length, or a typed refusal
// saying which hop and why) before it trusts the session.
func ExampleForward() {
	o, h1 := bind(link.NoAuth) // origin -> hop
	h2, f := bind(link.NoAuth) // hop -> far end

	// The far end: an echo, or a refusal with the application's code.
	go link.NewMux(f).Serve(func(header []byte, s *link.Session) {
		if string(header) != "echo" {
			_ = s.Refuse(link.CodeApp+1, "no such service")
			return
		}
		_ = s.Accept()
		_, _ = io.Copy(s, s)
		_ = s.CloseWrite()
	})

	// The hop: every session goes on to the far end (the Route decides
	// where; Forward answers, joins and holds no goroutine while idle).
	onward := link.NewMux(h2)
	go onward.Serve(nil)
	go link.NewMux(h1).Serve(link.Forward(func(ctx context.Context, in *link.Session) (*link.Session, error) {
		c, _ := link.ChainFromContext(ctx)
		return onward.OpenChain(in.Header(), c)
	}, link.ForwardConfig{}))

	// The origin.
	m := link.NewMux(o)
	go m.Serve(nil)
	defer m.Close()
	s, _ := m.Open([]byte("echo"))
	_, _ = io.WriteString(s, "hello") // before the answer: no round trip per hop
	_ = s.CloseWrite()
	if err := s.Answered(context.Background()); err != nil {
		panic(err)
	}
	reply, _ := io.ReadAll(s)
	fmt.Printf("%s, %d hop(s) beyond the peer\n", reply, s.PathLen())

	r, _ := m.Open([]byte("ftp"))
	var a *link.Abort
	if errors.As(r.Answered(context.Background()), &a) {
		fmt.Printf("refused %d hop(s) beyond the peer: code %d, %s\n", a.Hops, a.Code-link.CodeApp, a.Msg)
	}
	// Output:
	// hello, 1 hop(s) beyond the peer
	// refused 1 hop(s) beyond the peer: code 1, no such service
}

Over any peerlink.Link (a Node, quiclink, a mix): node.Serve(peerlink.Forward(route, cfg)) for a hop, peerlink.Join(a, b) to splice any two conns (a session and the TCP connection to its target at the far end, or a client's local connection), and the peerlink.ChainSession interface for the answers (quiclink v0.4.0 negotiates them with ALPN peerlink-chain). Every hop's Observer gets Hop events (opened, answered with its latency, ended with bytes each way) carrying the origin's trace id.

Bytes in flight per session and direction are bounded by the windows of the links plus one 256 KiB write batch per hop (3.5 MiB over three links with the default 1 MiB windows), so a stuck far reader stalls only its own session. Hops are transports: a hop sees the header, the bytes and the answers and can forge them; encrypt end to end inside the session (TLS with a pinned certificate) where that matters (SECURITY.md). The design is docs/design/chaining.md.

Measured (Apple M5, TCP loopback, in process):

Forward hops	links	MB/s, 1 session	MB/s, 32 sessions	64 B echo round trip, µs	OpenChain -> Answered, µs
0	1	1,812	1,922	55	27
1	2	2,433	1,708	205	55
2	3	1,256	2,434	321	112
4	5	1,725	1,173	505	220

Medians of three runs (go test -run '^$' -bench BenchmarkChain -benchmem -count 3; 8 MiB per session; a shared host, load average 14-26 on 10 CPUs, so the spread between runs is wide). A joined session at rest holds 0 goroutines and about 7.0 KiB of heap for its three sessions and joint (origin, one Forward hop, far end; TestChainIdleCost, 200 sessions). In-flight bytes for a stalled far reader stay within the links' windows plus 256 KiB per hop (TestChainStalledFarReader).

Performance

One connection, in process, Apple M5 (10 CPUs), v0.10.9 against v0.10.8, medians of 8 alternating runs on a shared machine (docs/performance.md has every cell, the method and each optimization):

message	sessions	TCP MB/s	TLS 1.3 MB/s	allocations per message
64 B	1	23 -> 32	21 -> 28	2 -> 0
64 B	8	56 -> 335	160 -> 295	2 -> 0
1 KiB	8	676 -> 1,180	694 -> 884	2 -> 0
4 KiB	1	1,465 -> 1,946	993 -> 1,402	1 -> 0
64 KiB	8	7,557 -> 7,953	2,951 -> 3,003	1 -> 0
1 MiB	1	8,834 -> 9,701	2,940 -> 2,915	20 -> 0

The 64 KiB and 1 MiB rows differ within the noise of a shared machine. A 64-byte echo on a held session takes 20 µs (p50) either way. Per message: one data frame (up to 64 KiB), at most one write of the connection (fewer when sessions share a write), and the receiver reads the connection about once per burst of small frames (v0.10.8: three reads per frame). An idle session holds no goroutine, and an idle connection no pooled buffer. TestSteadyStateAllocs and TestFramesAndReadsPerMessage fail if an allocation, a frame or a read comes back onto that path; scripts/perf.sh ab OLD NEW OUT compares two refs.

Liveness and deadlines

Liveness is per link. A connection pings once nothing arrived for its Interval and is dead once nothing arrived for Interval + Timeout (link.ErrDeadPeer), so an idle connection costs a few bytes whatever its session count: an idle link carried 156 bytes in 20 s with one session open on it and 208 with 500 (measured below). Set a preset on the dialer (choosing one); a listener with none of its own adopts the dialer's. On a chain each link keeps its own: when a link in the middle dies, the hops beside it notice within that link's bound and abort both ways with CodeConnLost, so both ends read a *link.Abort, never a clean EOF, within that bound plus one trip along the chain (Example_deadMiddleHop). A hop's drain does the same at its deadline with CodeDrainDeadline.

Deadlines are each side's own; link carries none from hop to hop. Answered(ctx) (or Pool.Open(ctx, ...), Node.Open(ctx, ...)) bounds the open across every hop. SetReadDeadline and SetWriteDeadline are local, as on a net.Conn. A side that gives up calls Abort (or Close), which reaches the far end through every hop as a typed *link.Abort. A Forward hop sets no read deadline and no idle limit; a write deadline set on a spliced session aborts both sides with CodeTimeout, and ForwardConfig.AnswerTimeout refuses an open whose onward answer is late (CodeTimeout, with the hop). Where the far end needs the caller's budget, send it in the header, as gRPC sends grpc-timeout in band:

// Example_deadlines: link carries no deadline value from hop to hop.
// The origin bounds the open across every hop with Answered's context,
// sets its own read deadline (local, as on a net.Conn), sends its budget
// in the header where the far end needs it, and propagates its giving up
// with Abort, which reaches the far end through the hop as a typed
// *link.Abort.
func Example_deadlines() {
	gaveUp := make(chan error, 1)
	m := chain3(func(header []byte, s *link.Session) {
		budget, _ := time.ParseDuration(string(header)) // the origin's budget, if the far end needs it
		_ = s.Accept()
		_ = s.SetReadDeadline(time.Now().Add(budget)) // the far end's own deadline
		_, err := io.Copy(io.Discard, s)              // works, never answers
		gaveUp <- err
	}, nil, link.Liveness{})
	defer m.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	s, err := m.Open([]byte("5s"))
	if err != nil {
		panic(err)
	}
	if err := s.Answered(ctx); err != nil { // the open, across every hop, within ctx
		panic(err)
	}
	stop := context.AfterFunc(ctx, func() { _ = s.Abort(link.CodeApp+1, "caller gave up") })
	defer stop()

	_ = s.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) // this side only
	_, err = s.Read(make([]byte, 1))
	fmt.Println("origin read:", errors.Is(err, os.ErrDeadlineExceeded))
	cancel() // give up: Abort, both ways along the chain

	var a *link.Abort
	if errors.As(<-gaveUp, &a) {
		fmt.Printf("far end: code %d, %d join(s) away: %s\n", a.Code-link.CodeApp, a.Hops, a.Msg)
	}
	// Output:
	// origin read: true
	// far end: code 1, 2 join(s) away: caller gave up
}

link adds no deadline field of its own: the application's protocol already carries one where it needs it (gRPC, HTTP), a wall-clock deadline would need hops to agree on time or subtract their own delay, and the case that matters, a side giving up, already reaches every hop as an abort.

What per-link liveness cannot see: a process that is stuck while its connection still answers pings (only a deadline finds it); a plain TCP connection beyond the chain, such as the one a far end dials to its target, which has TCP keepalive only; and a reader that stopped reading, which flow control bounds instead (it stalls its own session only). Measured with LivenessDefault (dead after 15 s) on every link of origin -> hop -> hop -> far end, the middle link black-holed (both TCP connections kept, every byte dropped): the origin read CodeConnLost after 14.0 s and the far end after 15.0 s, each from the hop beside the dead link (Hops 0); docs/design/chaining.md says how.

Which API

you want	use
sessions between named peers over TCP or TLS: start here	peerlink.Node (peerlink)
the same over QUIC (UDP), surviving address changes	quiclink.Node (peerlink/quiclink, a module of its own)
the same over a network that passes only gRPC	grpclink.Node (peerlink/grpclink, a module of its own)
gRPC or HTTP served over a link	peerlink.Listener (recipes)
code that works with any of these, and tests that check them	the peerlink.Link interface; peerlinktest.Run (the conformance suite)
advanced: your own connection management	link.Pool (connections to one peer), link.Mux (sessions on one connection), link.Client/link.Server (the handshake)
advanced: your own receive loop or transport	link.Conn (frames), link.Credit (per-session flow control)

Constructors check their configuration and name the field to fix (peerlink.New, link.NewPool, link.Config.Validate); a zero link.Config is valid for a dialer, and a listener needs Auth. Each package's documentation (go doc, or pkg.go.dev) has an Example for each common use.

Errors

Test errors with errors.Is (or errors.As for *link.VersionError); the messages name what to change.

error	where you see it	what it means	what to change
link.ErrHandshake (wraps the network's error)	Open (dialing), Client, Server, peerlink.Config.HandshakeError (func(net.Conn, error))	the hello or its reply did not arrive whole: the other end closed or timed out, is not link, speaks TLS where this end does not, or TLS refused a certificate	TLS on both ends or neither; the address of a link listener; the certificates (the CA each end trusts, the names, expiry), as the message says
link.ErrRefused	Open (dialing), Client	the listener's Authenticator refused the dialer's credential	the dialer's Config.Auth: what the listener accepts (CertAuth's certificate, the token)
link.ErrAuth (wraps Verify's error)	Server, peerlink.Config.HandshakeError (func(net.Conn, error))	this listener refused a dialer	the reason it wraps; the dialer's credential or this Config.Auth
link.ErrVersion, *link.VersionError	Client, Server	the peer speaks another wire version	upgrade the older end (every release since v0.5.0 speaks version 2)
link.ErrDeadPeer (a net.Error with Timeout() true)	session reads, Mux.Serve, Observer.ConnDown	nothing arrived for the liveness bound: the peer or the path died	nothing, if it did; else a preset with a longer bound for a path that goes silent (LivenessNAT, LivenessMobile)
link.ErrGoAway	Mux.Open, Conn.SendOpen	the connection is going away (a drain)	open on another connection (Pool and Node do)
link.ErrFull	Open	the peer's most sessions at once are open	close sessions sooner, raise the peer's Config.MaxSessions, or allow more connections (Pool.Max)
peerlink.ErrDraining	Node.Open	this Node is draining	nothing: it is shutting down
peerlink.ErrPeerUnreachable (and the context's error)	Node.Open without Dial	no connection from that peer before the context ended	the peer's name (the error lists the connected ones); that peer's Connect and Dial; a longer context
link.ErrReset	session reads	the peer reset the session	the peer's reason
*link.Abort (errors.As), link.ErrOpenRefused, link.ErrReset	Session.Answered, session reads (FeatureChain)	the far end (or the hop Hops says) refused the open or reset the session, with a Code and a Msg	what the code says: CodeDraining open elsewhere, CodeHopLimit a routing loop, CodeConnLost a hop's connection died, CodeApp+ the application's own
link.ErrNoAnswers	Session.Answered	the connection's peer does not answer opens (link before v0.10.0, or no FeatureChain)	nothing: use the session without an answer, or refuse such a leg in your Route
link.ErrConnClosed (wraps why: ErrDeadPeer, the network's error)	session reads; writes return io.ErrClosedPipe wrapping it	the connection closed under the session	open a new session (Pool and Node redial)
link.ErrFrame, link.ErrCredit	Conn.Recv, Mux.Serve	the peer broke the wire protocol	a peer that is not link, or a bug: please report it
a message naming a field	peerlink.New, link.NewPool, Validate, Client, Server	a configuration that cannot work	the field it names, as it says
io.EOF	session reads	the peer ended its writes on purpose (CloseWrite, Close)	nothing: the normal end
os.ErrDeadlineExceeded	session reads	the read deadline passed	nothing, or a later deadline

quiclink adds its own codes (quiclink.Code*, in the QUIC errors its sessions read); grpclink's are gRPC statuses.

Finding misuse: linkvet and debug mode

linkvet (github.com/nfltr/link/linkvet, a module of its own) is a go vet tool: a session some path leaves open (one Open returns, or the one an accept handler is given), a read loop with no deadline and no cancel path, a write after CloseWrite, an Open given context.Background(), and five checks for gRPC clients (apps behind the connect sidecar):

go install github.com/nfltr/link/linkvet/cmd/linkvet@latest
go vet -vettool=$(command -v linkvet) ./...

Debug mode is a build tag, linkdebug; a build without it carries nothing of it (no field in a Session, no call: the hooks sit behind a constant the compiler drops; TestDebugOffAddsNothing, and BenchmarkLatency/BenchmarkThroughput against v0.10.11: the same ns/op within noise, 0 allocs/op both). With it (go test -tags linkdebug ./..., or go build -tags linkdebug), link reports on stderr:

report	means
session 3 (header "x") opened and garbage-collected, never closed, with the stack that opened it	this side never called Close, CloseWrite, Reset or Abort (a session Join or Forward took counts as closed)
concurrent Reads on session ... (or Writes), with both stacks	two goroutines in Read/WriteTo (or Write/WriteBuffers) of one session at once: their bytes interleave
session ... and open 30s with no Read or Write, with the stack that opened it	with LINKDEBUG_IDLE=30s (any duration): a session open that long with no IO, once each; the leak a collector cannot see, since the Mux holds every session until it ends

Each Read and Write records a stack in this mode: for tests and debugging, not production.

We run both on ourselves: relay, relay-server and nfltr have a make check-linkvet, a CI step (fails on a finding outside tests that their scripts/linkvet-allow.txt does not list with a reason, and on a stale entry), and nfltr's kind relay soak and single-relay soak (LINKDEBUG=1) build the relay and its agents with -tags linkdebug and fail on any report.

Packages

package	what
link	Client/Server (the hello, an Authenticator's credential, Config, Observer), Conn (frames, write coalescing, OPEN/CLOSE, ping liveness, GOAWAY), Credit (per-session flow control, also over a caller's own transport), Mux and Session (sessions as net.Conns), Pool (several connections per peer, growth, backoff, make-before-break drain), Liveness and its presets
link/peerlink	a transport-neutral API for sessions between peers (Link: Open, Serve, Drain), its contract, Node (an implementation over Pool and Mux) and Listener (peerlink/README.md)
link/peerlink/peerlinktest	the contract's conformance suite (Run) and the fault-injecting Proxy (TCP) and UDPProxy, for any implementation
link/examples/quickstart	the quick start above
link/linkvet	go vet checks for code using link, and for gRPC clients (a module of its own, make check-linkvet; linkvet/README.md)

peerlink/grpclink (a module of its own, make check-grpclink) implements peerlink.Link over gRPC streams and passes the same suite (one direction: the dialer opens). peerlink/quiclink (a module of its own, make check-quiclink) implements it over QUIC streams and passes the suite in both directions; its connections survive the dialler's address changing (peerlink/quiclink/README.md).

Wire format version 2 is specified in WIRE.md, with test vectors for every message (testdata/wire_vectors.json); testdata/v2_session.golden holds a recorded session (and testdata/v2_chain_session.golden one with FeatureChain). Every release is tested against them, byte for byte.

Checks

make check: build, vet, tests with -race (the Examples, and TestDocCodeBlocksAreCompiled: every Go block in the Markdown here is a verbatim excerpt of a Go file, so a doc cannot rot), fuzz-short (every decoder, FUZZTIME each, default 10s), gosec, semgrep (docker), govulncheck. make check-api: the exported API against the latest release, with apidiff; the changes must be the ones listed in api/ (an API change is never an accident). make check-grpclink, make check-quiclink: the gRPC and QUIC implementations' modules; make check-soak: the soak module builds and vets; make check-linkvet: linkvet's tests, then its checks over link's own packages; make test-linkdebug (in check): debug mode's tests. make ci (alias check-all) runs every one of them: the steps .github/workflows/check.yml runs on a push, read from that file (scripts/ci.sh). CI is make ci, not GitHub Actions: make hooks installs a pre-push hook that runs it on the commit pushed to main or a tag (bypass NFLTR_CI_BYPASS=<reason>, logged), and a Linux runner on the maintainer's machine runs it for each new sha of main and each tag. The workflow still runs on GitHub as an extra signal, and every fuzz target for 5 minutes weekly. make fuzz-long FUZZTIME=30m is the long fuzz run (FUZZING.md).

Status and compatibility

The latest release is v0.10.12; CHANGELOG.md lists every release. link is pre-1.0 and follows semantic versioning as Go modules do for v0:

• The wire is the stronger promise: version 2 (WIRE.md) is held byte for byte by its test vectors and golden recording. A change to it is a new wire version, negotiated in the hello, and announced in the changelog; feature bits add optional frames without one.
• The Go API may change in a minor release (v0.x.0), with the change and its migration in the changelog; a patch release (v0.x.y) never removes or changes an exported name. Removals are announced a minor release ahead where possible.
• Behaviour that the docs state (bounds, errors to test with errors.Is, the peerlink contract) changes only in a minor release, except to fix a bug, which the changelog flags.
• peerlink/grpclink and peerlink/quiclink are modules of their own, versioned on their own (peerlink/<name>/vX.Y.Z). They stay v0 until their transport's module is v1: quiclink's API exposes quic-go's types.
• Interfaces that link calls (Observer) may gain methods in minor releases; implementations embed NopObserver, which the compiler requires. Interfaces that callers call (peerlink.Link, peerlink.Session, Authenticator) never change: new capabilities arrive as optional interfaces.
• peerlinktest may add checks in minor releases.

v1.0.0 will freeze the API once the maturity criteria below hold: from then on the Go API follows the Go 1 compatibility rules, and a breaking change is a /v2 module.

Maturity

Evidence that link works, provable from this repository alone:

criterion	evidence	status
wire specified, with test vectors	WIRE.md, testdata/wire_vectors.json (TestWireVectors), testdata/v2_session.golden	met
every decoder fuzzed, corpus checked in, 24 h cumulative without a crasher	FUZZING.md (hours per target, crashers), testdata/fuzz/	4.85 h, no crasher in the library (target: 24 h)
security review of the hello and the Authenticator contract	SECURITY.md (threat model, review record)	met (2026-10-14)
standard library only	TestModuleBoundary	met
CI green on every change, with history	GitHub Actions (check)	since 2026-10-14
cumulative soak hours under faults, with leak and integrity checks	SOAK.md (make soak), summaries in soak/results/	2.89 h (macOS, Linux CI; chain-3hop since v0.10.0), clean (target: 24 h)
benchmarks published in the repository	compare/README.md (link vs yamux), peerlink/quiclink/README.md	met
API reviewed for v1.0	the checklist below; make check-api (apidiff against the latest release, in CI); TestExportedNamesDocumented	open (v0.9.0: the review against yamux, smux, quic-go and libp2p applied; the peerlink contract changed in v0.9.0)

v1.0 API checklist: every exported name documented with an example where it is not obvious; errors to test with errors.Is listed per call; no exported name that exists only for one caller; option structs instead of positional growth; the peerlink contract unchanged for two minor releases; deprecated names removed. Use in production by its first user (an HTTP and TCP relay) is supporting evidence, not a criterion.

License

Apache License 2.0; see LICENSE and NOTICE.

Documentation

Overview

Package link multiplexes many byte-stream sessions, each a net.Conn with its own flow control, over one connection that the caller dials and secures (plain TCP, TLS, a pipe). A session whose reader is slow stalls only itself, dead peers are noticed within a bound the caller picks, and the package imports only the Go standard library.

Where to start

Most programs want github.com/nfltr/link/peerlink: a Node opens sessions to named peers and serves the sessions they open, keeping the connections (dialling, redialling, draining) for it. The README's quick start and docs/recipes.md (plaintext and mutual TLS, liveness presets, gRPC and HTTP over a link, graceful drain, reconnects, metrics) build on it.

This package is the layer below, for callers that manage connections themselves:

• Client and Server run the handshake on a connection (the hello, an Authenticator's credential) and return a Conn.
• Mux carries sessions over a Conn as [Session]s (net.Conns).
• Pool keeps several connections to one peer, with growth, dial backoff and make-before-break Pool.Drain.
• Conn and Credit are frames and per-session flow control, for a caller with its own receive loop or its own transport.

Authenticators: NameAuth on a trusted network, CertAuth over mutual TLS, NoAuth where peers need no names, or the caller's own (SECURITY.md). Config.Validate and PoolConfig.Validate name the field to fix.

Liveness

A connection is dead once nothing arrived for Interval + Timeout (ErrDeadPeer): each side pings once it received nothing for Interval. Pick a preset by how fast a dead peer must be noticed against how many idle frames you can afford (one small frame each way per Interval per connection):

• LivenessFast (4.5 s): failover and interactive use on a good network.
• LivenessDefault (15 s): most services; what a zero Config gets.
• LivenessNAT (30 s): paths through NATs, conntrack or load balancers that drop mappings idle for 30 s or more.
• LivenessDatacenter (45 s): stable networks where fewer wakeups matter more than fast failover.
• LivenessMobile (60 s): laptops and phones whose radio sleeps or hands over: the fewest wakeups, and a short silence is not taken for death.

ParseLiveness reads a preset by name, with overrides ("nat,timeout=40s"), for a flag or a config file. The listener takes the dialer's liveness unless its Config sets one.

Errors

Every error to test with errors.Is, by where it comes from (the README has the same as a table, with what to change):

• Client: ErrRefused (the listener refused the credential), ErrVersion (errors.As a *VersionError), ErrHandshake (the reply did not arrive: not a link listener, or a TLS mismatch; it wraps the network's error), ErrFrame, ctx's error.
• Server: ErrAuth (the Authenticator refused; it wraps Verify's error), ErrVersion (a version-1 hello among them), ErrHandshake (with ErrFrame when the first bytes are not a hello), ctx's error.
• Conn.Recv, Mux.Serve: ErrFrame, ErrCredit, ErrFull (the peer broke the protocol), ErrDeadPeer (liveness ended the connection), the network's (io.EOF once the peer closed).
• Conn.SendOpen, Mux.Open, Pool.Open: ErrFull (the peer's max sessions), ErrGoAway (the connection is going away); Mux.Open after Serve returned: net.ErrClosed.
• Session reads: io.EOF (the peer ended its writes on purpose), ErrReset (the peer's Reset), ErrConnClosed (the connection ended under the session; it wraps why, ErrDeadPeer among them), os.ErrDeadlineExceeded. Writes past such an end: io.ErrClosedPipe, wrapping the same error.
• Config.Validate, PoolConfig.Validate, NewPool, Client, Server: a message naming the field and what to set.

Stability

The wire (version 2, WIRE.md) is held byte for byte by its test vectors; a change to it is a new version negotiated in the hello. Before v1.0.0 the Go API may change in a minor release, with the migration in the changelog; from v1.0.0 it follows the Go 1 compatibility rules, and a breaking change is a /v2 module. Observer may gain methods in minor releases (implementations embed NopObserver); interfaces that callers call (Authenticator, peerlink.Link, peerlink.Session) never change, and new capabilities arrive as optional interfaces.

Wire format, version 2

WIRE.md in the module is the full specification, with test vectors for every message (testdata/wire_vectors.json); in short:

Version 2 is the only version since v0.5.0 (version 1, the hello of a bare token and opens and closes in the caller's messages, was removed; a version-1 hello gets the version reply, below, and Server ErrVersion).

hello  "NFDP" version(1)=2 max_version(1) features(4) window(4)
       max_sessions(4) interval_ms(4) timeout_ms(4)
       auth_method(1) auth_len(2) auth(auth_len, at most 4096)
reply  status(1):
         0 bound:   version(1)=2 features(4) window(4) max_sessions(4)
         1 refused (the credential)
         2 version: max_version(1), the highest the listener speaks
frame  both ways: length(4, of the body) type(1) body
  type 0, data:    sid_len(1) sid payload
  type 1, message: opaque bytes of the caller's (never empty; Conn.SendMsg
                   and Conn.Recv carry them, a Mux drops them)
  type 2, window:  sid_len(1) sid increment(4)
  type 3, open:    sid_len(1) sid header (the opener's bytes, at most 64 KiB)
  type 4, close:   sid_len(1) sid code(1): 0 the sender neither sends
                   nor takes more; 1 it sends no more (half-close);
                   2 as 0, and the session failed (reset; only to a
                   peer that announced feature bit 0)
  type 5, ping:    8 opaque bytes
  type 6, pong:    the ping's 8 bytes
  type 7, goaway:  drain_ms(4, 0 = none) reason (UTF-8, at most 1024)

Integers are big endian. A session id (sid) is at most 255 bytes. A window is how many bytes of one session the side that announces it buffers (see Credit.Acquire). Session ids that begin with '_' name control frames, which Mux leaves to the caller.

A listener answers a version it does not speak with status 2. A listener of link v0.4 or older that speaks only version 1 reads 37 bytes of a version-2 hello before it closes the connection.

max_sessions is how many sessions the peer may have open toward the side that announces it at once; an open past it ends the connection. Feature bits: a side announces the ones it reads and ignores the bits it does not know. Bit 0, reset (since v0.3.0): the side reads CLOSE code 2; v0.2 sides announce none and end the connection on code 2, so a reset to them is sent as code 0. Liveness is the dialer's (interval 1s to 15s, timeout at least one interval); a listener takes it unless it has its own: each side pings once it received nothing for the interval and ends the connection when nothing arrived for the timeout after that (dead within interval + timeout of the peer's last frame, idle or busy); every ping is answered (the latest, when several wait). An open starts the session's credit on both sides with the peer's window; a close of code 0 ends it. A Mux picks session ids as uvarints, odd on the dialer and even on the listener. After a goaway the receiver opens no more sessions on the connection. The sender follows its goaway with a ping: opens that arrive before that ping's pong crossed the goaway and run like any other session (since v0.4.2; before, they were refused with a close), opens after it are refused; the sender closes the connection once its last session ended and the pong arrived, after writing every frame.

testdata/v2_session.golden holds a recorded version-2 session.

Example

Example is a listener and a dialer end to end over TLS on loopback: the listener authenticates the dialer and serves each session it opens (here, upper-casing what it reads); the dialer keeps a Pool of connections to it and opens a session per request.

package main

import (
	"bytes"
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/subtle"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"io"
	"math/big"
	"net"
	"time"

	"github.com/nfltr/link"
)

// tokenAuth is a shared-secret Authenticator: the dialer sends the token,
// the listener compares it. Method 128 and up are the caller's to define.
type tokenAuth []byte

func (t tokenAuth) Hello(context.Context, tls.ConnectionState) (byte, []byte, error) {
	return 128, t, nil
}

func (t tokenAuth) Verify(_ context.Context, _ tls.ConnectionState, method byte, payload []byte) (link.Identity, error) {
	if method != 128 || subtle.ConstantTimeCompare(payload, t) != 1 {
		return link.Identity{}, errors.New("bad token")
	}
	return link.Identity{Method: "token", Subject: "example-client"}, nil
}

func main() {
	ctx := context.Background()
	auth := tokenAuth("s3cret")
	serverTLS, clientTLS := exampleTLS()

	// Listener: one link per accepted connection, one goroutine per session.
	ln, err := tls.Listen("tcp", "127.0.0.1:0", serverTLS)
	if err != nil {
		panic(err)
	}
	defer ln.Close()
	who := make(chan link.Identity, 1)
	go func() {
		for {
			nc, err := ln.Accept()
			if err != nil {
				return
			}
			go func() {
				// The handshake's deadline: a dialer that never finishes its
				// hello does not hold the connection (SECURITY.md).
				hctx, cancel := context.WithTimeout(ctx, 10*time.Second)
				c, id, err := link.Server(hctx, nc, link.Config{Auth: auth})
				cancel()
				if err != nil {
					nc.Close()
					return
				}
				who <- id
				_ = link.NewMux(c).Serve(func(header []byte, s *link.Session) {
					defer s.Close()
					body, _ := io.ReadAll(s) // until the dialer's CloseWrite
					fmt.Fprintf(s, "%s: %s", header, bytes.ToUpper(body))
				})
			}()
		}
	}()

	// Dialer: a Pool dials (TLS is the caller's), authenticates, and opens
	// sessions on its connections; the listener adopts the dialer's
	// liveness, so both ends notice a dead path within 1.5 s + 3 s.
	p, err := link.NewPool(link.PoolConfig{
		Link: link.Config{Auth: auth, Liveness: link.LivenessFast},
		Dial: func(ctx context.Context) (net.Conn, error) {
			var d tls.Dialer
			d.Config = clientTLS
			return d.DialContext(ctx, "tcp", ln.Addr().String())
		},
		Max:     2,
		Backoff: func(attempt int) time.Duration { return time.Duration(attempt) * 100 * time.Millisecond },
	})
	if err != nil {
		panic(err)
	}
	defer p.Close()

	for _, msg := range []string{"hello", "world"} {
		s, err := p.Open(ctx, []byte("upper"))
		if err != nil {
			panic(err)
		}
		_, _ = io.WriteString(s, msg)
		_ = s.CloseWrite() // the listener reads io.EOF after msg
		reply, err := io.ReadAll(s)
		s.Close()
		if err != nil {
			panic(err)
		}
		fmt.Println(string(reply))
	}
	fmt.Println("listener saw:", (<-who).Subject)
}

// exampleTLS is a throwaway self-signed certificate for 127.0.0.1 and the
// client config that trusts it.
func exampleTLS() (server, client *tls.Config) {
	key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	tmpl := &x509.Certificate{
		SerialNumber: big.NewInt(1),
		IPAddresses:  []net.IP{net.ParseIP("127.0.0.1")},
		NotBefore:    time.Now().Add(-time.Minute),
		NotAfter:     time.Now().Add(time.Hour),
	}
	der, _ := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
	cert, _ := x509.ParseCertificate(der)
	roots := x509.NewCertPool()
	roots.AddCert(cert)
	return &tls.Config{Certificates: []tls.Certificate{{Certificate: [][]byte{der}, PrivateKey: key}}, MinVersion: tls.VersionTLS13},
		&tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS13}
}

Output:
upper: HELLO
upper: WORLD
listener saw: example-client

Example (DeadMiddleHop)

Example_deadMiddleHop: liveness is per link. The middle link of origin -> hop1 -> hop2 -> far end goes silent; the hops beside it declare it dead within its own bound (here interval 1 s, timeout 2 s: 3 s) and abort both ways, so both ends read CodeConnLost, never a clean end, while the links they hold themselves stay up.

off := new(atomic.Bool)
farErr := make(chan error, 1)
m := chain3(func(_ []byte, s *link.Session) {
	_ = s.Accept()
	_, err := io.Copy(io.Discard, s)
	farErr <- err
}, func(c net.Conn) net.Conn { return silent{c, off} }, link.Liveness{Interval: time.Second, Timeout: 2 * time.Second})
defer m.Close()

s, _ := m.Open(nil)
if err := s.Answered(context.Background()); err != nil {
	panic(err)
}
start := time.Now()
off.Store(true) // the middle link goes silent
_, originErr := io.Copy(io.Discard, s)
took := time.Since(start)

for _, end := range []struct {
	name string
	err  error
}{{"origin", originErr}, {"far end", <-farErr}} {
	var a *link.Abort
	if errors.As(end.err, &a) {
		fmt.Printf("%s: CodeConnLost %v, from the hop beside the dead link: %v\n", end.name, a.Code == link.CodeConnLost, a.Hops == 0)
	}
}
fmt.Println("within the link's bound plus a trip:", took < 3*time.Second+time.Second)

Output:
origin: CodeConnLost true, from the hop beside the dead link: true
far end: CodeConnLost true, from the hop beside the dead link: true
within the link's bound plus a trip: true

Example (Deadlines)

Example_deadlines: link carries no deadline value from hop to hop. The origin bounds the open across every hop with Answered's context, sets its own read deadline (local, as on a net.Conn), sends its budget in the header where the far end needs it, and propagates its giving up with Abort, which reaches the far end through the hop as a typed *link.Abort.

gaveUp := make(chan error, 1)
m := chain3(func(header []byte, s *link.Session) {
	budget, _ := time.ParseDuration(string(header)) // the origin's budget, if the far end needs it
	_ = s.Accept()
	_ = s.SetReadDeadline(time.Now().Add(budget)) // the far end's own deadline
	_, err := io.Copy(io.Discard, s)              // works, never answers
	gaveUp <- err
}, nil, link.Liveness{})
defer m.Close()

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
s, err := m.Open([]byte("5s"))
if err != nil {
	panic(err)
}
if err := s.Answered(ctx); err != nil { // the open, across every hop, within ctx
	panic(err)
}
stop := context.AfterFunc(ctx, func() { _ = s.Abort(link.CodeApp+1, "caller gave up") })
defer stop()

_ = s.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) // this side only
_, err = s.Read(make([]byte, 1))
fmt.Println("origin read:", errors.Is(err, os.ErrDeadlineExceeded))
cancel() // give up: Abort, both ways along the chain

var a *link.Abort
if errors.As(<-gaveUp, &a) {
	fmt.Printf("far end: code %d, %d join(s) away: %s\n", a.Code-link.CodeApp, a.Hops, a.Msg)
}

Output:
origin read: true
far end: code 1, 2 join(s) away: caller gave up

Index

• Constants
• Variables
• func AppendWindow(b []byte, sid string, n int) []byte
• func CheckWindow(w uint32) (int, error)
• func ContextWithChain(ctx context.Context, c Chain) context.Context
• func Forward(route Route, cfg ForwardConfig) func(header []byte, s *Session)
• func IsHello(prefix []byte) bool
• func Join(a, b *Session) error
• func ParseWindow(b []byte) (sid string, n int, err error)
• func Server(ctx context.Context, nc net.Conn, cfg Config) (*Conn, Identity, error)
• type Abort
  • func (a *Abort) Error() string
  • func (a *Abort) Is(target error) bool
• type Authenticator
  • func NameAuth(name string) Authenticator
• type Chain
  • func ChainFromContext(ctx context.Context) (Chain, bool)
• type Config
  • func (cfg Config) Validate() error
• type Conn
  • func Client(ctx context.Context, nc net.Conn, cfg Config) (*Conn, error)
  • func (c *Conn) Close() error
  • func (c *Conn) Credit() *Credit
  • func (c *Conn) Done() <-chan struct{}
  • func (c *Conn) Drained() <-chan struct{}
  • func (c *Conn) Err() error
  • func (c *Conn) Flush() error
  • func (c *Conn) GoAway(drain time.Duration, reason string) error
  • func (c *Conn) GoingAway() bool
  • func (c *Conn) Identity() Identity
  • func (c *Conn) PeerGoAway() <-chan struct{}
  • func (c *Conn) PeerMaxSessions() int
  • func (c *Conn) Recv() (Frame, error)
  • func (c *Conn) SendClose(sid string, half bool) error
  • func (c *Conn) SendData(sid string, data []byte) error
  • func (c *Conn) SendMsg(b []byte) error
  • func (c *Conn) SendOpen(sid string, header []byte) error
  • func (c *Conn) SendReset(sid string) error
  • func (c *Conn) Version() int
• type Credit
  • func NewCredit(window, peerWindow int, grant func(sid string, n int)) *Credit
  • func (c *Credit) Acquire(sid string, n int) bool
  • func (c *Credit) Close()
  • func (c *Credit) Consumed(sid string, n int)
  • func (c *Credit) Forget(sid string)
  • func (c *Credit) Grant(sid string, n int) error
  • func (c *Credit) Open(sid string)
  • func (c *Credit) PeerWindow() int
  • func (c *Credit) Received(sid string, n int) error
  • func (c *Credit) Window() int
• type ForwardConfig
• type Frame
• type HopEvent
• type HopKind
• type Identity
• type Liveness
  • func ParseLiveness(spec string) (Liveness, error)
  • func (l Liveness) DeadAfter(stated time.Duration) time.Duration
  • func (l Liveness) String() string
  • func (l Liveness) Validate() error
• type Mux
  • func NewMux(c *Conn) *Mux
  • func (m *Mux) Close() error
  • func (m *Mux) Conn() *Conn
  • func (m *Mux) Drain(ctx context.Context) error
  • func (m *Mux) Open(header []byte) (*Session, error)
  • func (m *Mux) OpenChain(header []byte, c Chain) (*Session, error)
  • func (m *Mux) Serve(accept func(open []byte, s *Session)) error
  • func (m *Mux) Sessions() int
• type NopObserver
  • func (NopObserver) ConnDown(Identity, error)
  • func (NopObserver) ConnUp(Identity)
  • func (NopObserver) CreditWait(time.Duration)
  • func (NopObserver) GoAway(Identity)
  • func (NopObserver) Hop(HopEvent)
  • func (NopObserver) PingRTT(time.Duration)
  • func (NopObserver) SessionClose(error)
  • func (NopObserver) SessionOpen()
• type Observer
• type OwnedWriter
• type Pool
  • func NewPool(cfg PoolConfig) (*Pool, error)
  • func (p *Pool) Close() error
  • func (p *Pool) Drain(ctx context.Context) error
  • func (p *Pool) Open(ctx context.Context, header []byte) (*Session, error)
  • func (p *Pool) Stats() PoolStats
• type PoolConfig
  • func (cfg PoolConfig) Validate() error
• type PoolStats
• type Route
• type Session
  • func (s *Session) Abort(code uint32, msg string) error
  • func (s *Session) AbortWith(a *Abort) error
  • func (s *Session) Accept() error
  • func (s *Session) AcceptPath(pathLen uint8) error
  • func (s *Session) Answered(ctx context.Context) error
  • func (s *Session) Answers() bool
  • func (s *Session) Chain() Chain
  • func (s *Session) Close() error
  • func (s *Session) CloseWrite() error
  • func (s *Session) Done() <-chan struct{}
  • func (s *Session) Header() []byte
  • func (s *Session) LocalAddr() net.Addr
  • func (s *Session) PathLen() uint8
  • func (s *Session) Pump(w io.Writer, done func(n int64, err error))
  • func (s *Session) Read(b []byte) (int, error)
  • func (s *Session) Refuse(code uint32, msg string) error
  • func (s *Session) RemoteAddr() net.Addr
  • func (s *Session) Reset() error
  • func (s *Session) SetDeadline(t time.Time) error
  • func (s *Session) SetReadDeadline(t time.Time) error
  • func (s *Session) SetWriteDeadline(t time.Time) error
  • func (s *Session) Write(b []byte) (int, error)
  • func (s *Session) WriteBuffers(bufs [][]byte) (int64, error)
  • func (s *Session) WriteTo(w io.Writer) (int64, error)
• type Trace
• type VersionError
  • func (e *VersionError) Error() string
  • func (e *VersionError) Is(target error) bool

Examples

• Package
• Package (DeadMiddleHop)
• Package (Deadlines)
• Client
• Conn
• Forward
• Mux
• NewPool
• ParseLiveness

Constants

const (
	CodeUnspecified   uint32 = 0
	CodeRefused       uint32 = 1 // not taking sessions: no accept function, or no room
	CodeDraining      uint32 = 2 // GOAWAY or Drain: open elsewhere
	CodeHopLimit      uint32 = 3 // MaxHops reached: a loop, or a path too long
	CodeNoRoute       uint32 = 4 // the Route failed with an error that is no *Abort
	CodeConnLost      uint32 = 5 // a hop's connection ended (dead peer, network)
	CodeTimeout       uint32 = 6 // no answer within ForwardConfig.AnswerTimeout, or a write deadline
	CodeDrainDeadline uint32 = 7 // a draining hop's deadline cut the session
	CodeApp           uint32 = 1024
)

Codes of an Abort: link's below CodeApp, the application's from it.

const (
	MethodNone  byte = 0 // no credential (NoAuth, CertAuth)
	MethodToken byte = 1 // a bearer token the listener issued
	MethodName  byte = 2 // a name the dialer claims (NameAuth)
)

Auth methods with a meaning in this package; the others are the Authenticator's to define (128 and up are never assigned here).

const (
	MinWindow = 2 * MaxChunk
	MaxWindow = 1 << 30
)

Window bounds: a side announces a window in [MinWindow, MaxWindow]. Credit returns in steps of a quarter window; a window of two MaxChunks always returns a waiting sender the MaxChunk it asks for.

const (
	MinLivenessInterval = time.Second
	MaxLivenessInterval = 15 * time.Second
)

Bounds Validate enforces. A heartbeat sender keeps an interval in [MinLivenessInterval, MaxLivenessInterval]: below it the frames cost more than they tell, above it a 45 s limit for a silent peer (one that answers each frame) would drop the connection. Timeout is at least one more interval, since a peer skips a frame right after sending data.

const DefaultMaxHops = 8

DefaultMaxHops is ForwardConfig's MaxHops when it names none.

const DefaultWindow = 1 << 20

DefaultWindow is a Config's window when it names none.

const FeatureChain uint32 = 1 << 1

FeatureChain is feature bit 1 (WIRE.md section 6): the side reads ACCEPT, ABORT and CHAIN-OPEN frames, answers every session opened to it (Accept or Refuse) and opens sessions with CHAIN-OPEN. A connection where both sides announced it carries answers and typed aborts (Session.Answers); one where either did not is byte for byte the v0.9.4 connection.

const Magic = "NFDP"

Magic opens a link connection, where an HTTP/2 connection starts "PRI ": a listener that shares its port can tell the two apart.

const MaxAbortMsg = 1024

MaxAbortMsg bounds an Abort's message: Refuse and Abort cut a longer one at a UTF-8 boundary, decoders refuse it.

const MaxAuth = 4096

MaxAuth bounds a hello's auth payload.

const MaxChunk = 64 << 10

MaxChunk is the most data a sender acquires credit for at once: larger writes are split.

const MaxSID = 255

MaxSID is the longest session id.

const MaxSessions = 1024

MaxSessions is the most sessions one Mux carries at once by default: Open refuses more (ErrFull), Serve closes a session the peer opens past it. The peer's hello sets the limit toward it.

const MaxVersion = 2

MaxVersion is the highest wire version this package speaks.

Variables

var (
	// ErrDeadPeer ends a connection on which nothing arrived for its
	// liveness's interval plus timeout, a ping included. It is a net.Error
	// whose Timeout() is true.
	ErrDeadPeer error = deadPeerError{}
	// ErrGoAway is Open's refusal on a connection that is going away
	// (either side sent GOAWAY).
	ErrGoAway = errors.New("link: connection going away")
	// ErrHandshake is a hello or its reply that did not arrive whole:
	// the other end closed the connection, timed out, or is not a link
	// end speaking the same TLS-or-not (the error says which, and wraps
	// the network's).
	ErrHandshake = errors.New("link: handshake failed")
	// ErrAuth is a listener's refusal of a dialer its Authenticator did
	// not verify.
	ErrAuth = errors.New("link: peer not authenticated")
	// ErrVersion is a peer that does not speak the version asked for
	// (VersionError says which it speaks).
	ErrVersion = errors.New("link: wire version not supported")
)

var (
	// LivenessDefault: dead within 15 s, a frame each way every 5 s.
	LivenessDefault = Liveness{Interval: 5 * time.Second, Timeout: 10 * time.Second}
	// LivenessFast: quick failover and interactive use; dead within 4.5 s,
	// a frame each way every 1.5 s.
	LivenessFast = Liveness{Interval: 1500 * time.Millisecond, Timeout: 3 * time.Second}
	// LivenessNAT: long idle behind NATs, conntrack and load balancers that
	// drop a mapping idle for 30 s or more; dead within 30 s, a frame each
	// way every 10 s.
	LivenessNAT = Liveness{Interval: 10 * time.Second, Timeout: 20 * time.Second}
	// LivenessDatacenter: stable networks; dead within 45 s, a frame each
	// way every 15 s.
	LivenessDatacenter = Liveness{Interval: 15 * time.Second, Timeout: 30 * time.Second}
	// LivenessMobile: laptops and phones on flaky Wi-Fi or metered links:
	// the fewest wakeups, and a radio that sleeps or hands over is not
	// taken for a dead peer; dead within 60 s, a frame each way every 15 s.
	LivenessMobile = Liveness{Interval: 15 * time.Second, Timeout: 45 * time.Second}
)

The presets. Idle traffic is one small frame each way per Interval per connection; the bound is Interval + Timeout from the peer's last frame.

var ErrConnClosed = errors.New("link: connection closed")

ErrConnClosed is what a session reads once its connection closed: the error wraps it and why (ErrDeadPeer, the network's error; the peer's close in its text only, never as io.EOF; nothing when this side closed it), and says to open a new session.

var ErrCredit = errors.New("link: peer exceeded the session window")

ErrCredit is the receive error of a peer that broke the credit rules: it sent a session more data than this side's window, or granted more credit than its own window.

var ErrDrainDeadline = errors.New("link: the drain's deadline cut the session")

ErrDrainDeadline is matched (errors.Is) by the error of a session that a draining Mux's deadline cut, on the side that drained: a splice (Join, peerlink.Forward) sends it on as CodeDrainDeadline.

var ErrFrame = errors.New("link: malformed frame")

ErrFrame reports bytes that break the wire format (WIRE.md): a frame, or a hello or reply with a field out of its bounds.

var ErrFull = errors.New("link: connection carries its most sessions")

ErrFull is Open's refusal of a session past the most sessions allowed.

var ErrNoAnswers = errors.New("link: the peer does not answer opens (no FeatureChain)")

ErrNoAnswers is Answered on a session whose connection did not negotiate FeatureChain: its peer does not answer opens.

var ErrOpenRefused = errors.New("link: session refused")

ErrOpenRefused is what a refused session's Answered and reads match (errors.Is): the *Abort says who and why.

var ErrRefused = errors.New("link: peer refused the connection (check the dialer's Config.Auth against what the listener accepts: its token, CertAuth's client certificate)")

ErrRefused is the listener's refusal of a hello: an unknown token, or one that binds no more connections.

var ErrReset = errors.New("link: session reset by the peer")

ErrReset is what a session reads once the peer reset it (Reset).

var LivenessPresets = []string{"default", "fast", "nat", "datacenter", "mobile"}

LivenessPresets names the presets, in the order help and docs list them.

Functions

func AppendWindow

func AppendWindow(b []byte, sid string, n int) []byte

AppendWindow appends a window update's body, sid_len(1) sid increment(4), to b: the body of a window frame, and what a caller with a transport of its own carries for a Credit's grant. sid is at most MaxSID bytes and n at most a window.

func CheckWindow

func CheckWindow(w uint32) (int, error)

CheckWindow returns w as a window, or an error when it is out of range.

func ContextWithChain

func ContextWithChain(ctx context.Context, c Chain) context.Context

ContextWithChain returns ctx carrying c: Pool.Open (and peerlink.Node's Open) open with it (Mux.OpenChain). Forward hands its Route a ctx carrying the inbound session's Chain with Hops+1.

func Forward

func Forward(route Route, cfg ForwardConfig) func(header []byte, s *Session)

Forward returns an accept function for Mux.Serve (or a Pool's Accept) that makes this side a hop of a path: it refuses an open whose Chain already crossed MaxHops joins (a routing loop), calls route, forwards the bytes the origin sent meanwhile, waits for the onward answer and answers in with it (an ACCEPT with the path's length, or the refusal one hop further), then joins the two sessions (Join) and returns: a forwarded session holds no goroutine while idle. Toward an onward session without answers (Answers false) it accepts in once route returned; a Route that wants answers end to end refuses such a session itself. Observer gets HopOpened, HopAnswered and HopEnded.

Example

ExampleForward joins two links end to end: the origin opens through a forwarding hop to a far end, which answers each open; the origin reads the answer (accepted, with the path's length, or a typed refusal saying which hop and why) before it trusts the session.

o, h1 := bind(link.NoAuth) // origin -> hop
h2, f := bind(link.NoAuth) // hop -> far end

// The far end: an echo, or a refusal with the application's code.
go link.NewMux(f).Serve(func(header []byte, s *link.Session) {
	if string(header) != "echo" {
		_ = s.Refuse(link.CodeApp+1, "no such service")
		return
	}
	_ = s.Accept()
	_, _ = io.Copy(s, s)
	_ = s.CloseWrite()
})

// The hop: every session goes on to the far end (the Route decides
// where; Forward answers, joins and holds no goroutine while idle).
onward := link.NewMux(h2)
go onward.Serve(nil)
go link.NewMux(h1).Serve(link.Forward(func(ctx context.Context, in *link.Session) (*link.Session, error) {
	c, _ := link.ChainFromContext(ctx)
	return onward.OpenChain(in.Header(), c)
}, link.ForwardConfig{}))

// The origin.
m := link.NewMux(o)
go m.Serve(nil)
defer m.Close()
s, _ := m.Open([]byte("echo"))
_, _ = io.WriteString(s, "hello") // before the answer: no round trip per hop
_ = s.CloseWrite()
if err := s.Answered(context.Background()); err != nil {
	panic(err)
}
reply, _ := io.ReadAll(s)
fmt.Printf("%s, %d hop(s) beyond the peer\n", reply, s.PathLen())

r, _ := m.Open([]byte("ftp"))
var a *link.Abort
if errors.As(r.Answered(context.Background()), &a) {
	fmt.Printf("refused %d hop(s) beyond the peer: code %d, %s\n", a.Hops, a.Code-link.CodeApp, a.Msg)
}

Output:
hello, 1 hop(s) beyond the peer
refused 1 hop(s) beyond the peer: code 1, no such service

func IsHello

func IsHello(prefix []byte) bool

IsHello reports whether a connection's first bytes are a link hello's: a listener that shares its port (with TLS or HTTP/2) tells them apart by its first len(Magic) bytes.

func Join

func Join(a, b *Session) error

Join splices a and b, sessions of the same Mux or of two, until both directions ended, and returns nil after a clean end, else the *Abort it sent on (the first). It answers either one the peer opened and that is not answered yet (Accept).

Each direction carries the bytes written, in order, intact. A clean end of one side's writes (CloseWrite) becomes CloseWrite on the other once every byte before it was written there, and a full close becomes Close; each direction ends on its own. An error on either side (an Abort or Reset, its connection's end, a write deadline set on a session that passed) aborts the other: an *Abort received goes on with Hops+1 (a plain reset as CodeUnspecified, Hops 1), a local cause as an Abort with Hops 0 and CodeConnLost, CodeTimeout or CodeDrainDeadline.

Join takes a batch (at most 256 KiB) from one side's receive queue only when it writes it to the other, and the queue returns credit as bytes leave it: what a joined session holds is its two windows plus two batches. It holds no goroutine while the sessions are idle (Pump): only the caller's waits in Join. Join sets no deadline; idle limits are the caller's, who may Abort either session. An Observer of a's connection (else b's) gets a HopEnded event.

func ParseWindow

func ParseWindow(b []byte) (sid string, n int, err error)

ParseWindow reads a window update's body (AppendWindow).

func Server

func Server(ctx context.Context, nc net.Conn, cfg Config) (*Conn, Identity, error)

Server binds nc, whose security (TLS) is the caller's, as the listener: it reads the dialer's hello (Magic first: a caller that sniffed it replays it), verifies its credential with cfg.Auth, replies, and returns the connection and who the dialer is. It speaks version 2 (MaxVersion). A version-1 hello (a dialer of link v0.4 or older) gets the version reply (status 2 and MaxVersion), which such a dialer reads as a refusal, and Server returns an error that is ErrVersion. A refused dialer gets the refusal and Server ErrAuth. On error the caller closes nc. nc should have CloseWrite, as for Client.

ctx bounds the handshake: give it a deadline. Without one, a dialer that sends its hello slowly, or not at all, holds nc and this goroutine for as long as it likes.

Types

type Abort

type Abort struct {
	Refused bool
	Code    uint32 // link's below CodeApp, the application's from CodeApp
	Hops    uint8  // joins the abort crossed: 0 = this side's peer itself
	Msg     string // at most MaxAbortMsg bytes of UTF-8
}

Abort is a session's typed end on a FeatureChain connection: a refusal (the accepter's answer instead of its accept) or a reset (either side, any time). Reads of the session, and Answered, return it. errors.Is(a, ErrOpenRefused) for refusals; errors.Is(a, ErrReset) for both (a refused open ends as a reset did before v0.10.0).

func (*Abort) Error

func (a *Abort) Error() string

Error says what ended the session, where on the path and why.

func (*Abort) Is

func (a *Abort) Is(target error) bool

Is matches ErrOpenRefused for a refusal and ErrReset for any Abort.

type Authenticator

type Authenticator interface {
	// Hello returns the dialer's auth method and payload (at most MaxAuth
	// bytes).
	Hello(ctx context.Context, cs tls.ConnectionState) (method byte, payload []byte, err error)
	// Verify checks a dialer's method and payload and returns who it is.
	// An error refuses the connection.
	Verify(ctx context.Context, cs tls.ConnectionState, method byte, payload []byte) (Identity, error)
}

Authenticator proves a dialer to a listener. The connection's TLS state (zero on a plaintext one) is there for channel binding.

Verify runs before the listener has bound anything, on bytes from an unauthenticated peer (SECURITY.md): it compares secrets in constant time (crypto/subtle), bounds its own work (ctx carries the handshake's deadline), and returns an Identity whose Subject names the peer uniquely where callers route by it (peerlink names a peer by Subject). Its error is not sent to the dialer, which reads only a refusal.

var CertAuth Authenticator = certAuth{}

CertAuth is an Authenticator for mutual TLS: the dialer sends no credential, and the listener accepts a dialer whose client certificate its tls.Config verified (ClientAuth: tls.RequireAndVerifyClientCert), named by the certificate's Subject CommonName, else its first DNS name. Give Server (or peerlink.Config.Listener) a *tls.Conn.

var NoAuth Authenticator = noAuth{}

NoAuth sends no credential and accepts only peers that send none: for connections whose security is elsewhere (an mTLS peer port, a pipe). Every peer it accepts has the zero Identity: a peerlink.Node names them all "" (see NameAuth and CertAuth for peers with names).

func NameAuth

func NameAuth(name string) Authenticator

NameAuth is an Authenticator for a trusted network (plaintext inside a cluster, a pipe): the dialer sends name, and the listener accepts any name of 1 to MaxSID bytes as the peer's Identity.Subject. It proves nothing: anyone who reaches the listener can claim any name. Over an untrusted network use CertAuth with mutual TLS, or an Authenticator that checks a secret.

type Chain

type Chain struct {
	Trace Trace // the origin's; zero in OpenChain: link picks one (crypto/rand)
	Hops  uint8 // joins the open crossed: 0 from the origin
}

Chain is where an open is on its path.

func ChainFromContext

func ChainFromContext(ctx context.Context) (Chain, bool)

ChainFromContext returns the Chain ctx carries (ContextWithChain).

type Config

type Config struct {
	// Window is how many bytes of one session this side buffers: 0 is
	// DefaultWindow.
	Window int
	// MaxSessions is how many sessions the peer may have open toward this
	// side at once: 0 is MaxSessions.
	MaxSessions int
	// Liveness: this side declares the connection dead when it
	// received nothing for Interval, sent a ping, and received nothing for
	// Timeout more (ErrDeadPeer). A dialer's zero is LivenessDefault; a
	// listener's zero is the dialer's, so both ends detect a dead path
	// within the same bound.
	// A listener that must bound how long a vanished dialer holds its
	// connection sets its own: the dialer's Timeout has no upper bound.
	Liveness Liveness
	// Auth builds the dialer's credential (Hello) and verifies it on the
	// listener (Verify). A dialer's nil is NoAuth; a listener needs one:
	// NameAuth on a trusted network, CertAuth over mutual TLS, NoAuth
	// where peers need no names, or the caller's own.
	Auth Authenticator
	// Observer receives the connection's events; nil is none.
	Observer Observer
	// contains filtered or unexported fields
}

Config is one side's settings for a connection (Client, Server).

func (Config) Validate

func (cfg Config) Validate() error

Validate reports the first field of cfg out of its range, naming it and what to set; the zero Config is valid. Client and Server (and NewPool, for PoolConfig.Link) refuse a Config it refuses.

type Conn

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

Conn sends and receives frames on a bound connection. One goroutine Receives; the sends, Flush and the credit calls (its Credit's) are safe from any.

Reads go through a read-ahead buffer (read.go; docs/performance.md, P3): a burst of small frames is one read of the connection, and an idle connection holds no pooled buffer (only rsmall, its own).

Example

ExampleConn drives a Conn without a Mux: message frames, the caller's own bytes, one receive loop.

package main

import (
	"context"
	"fmt"
	"net"
	"time"

	"github.com/nfltr/link"
)

// bind binds both ends of one connection (here an in-memory pipe; in
// practice a TLS connection the caller dialled and accepted).
func bind(auth link.Authenticator) (dialer, listener *link.Conn) {
	a, b := net.Pipe()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	type bound struct {
		c   *link.Conn
		err error
	}
	ch := make(chan bound, 1)
	go func() {
		c, _, err := link.Server(ctx, b, link.Config{Auth: auth})
		ch <- bound{c, err}
	}()
	dialer, err := link.Client(ctx, a, link.Config{Auth: auth})
	l := <-ch
	if err != nil || l.err != nil {
		panic(fmt.Sprint(err, l.err))
	}
	return dialer, l.c
}

func main() {
	dc, lc := bind(link.NoAuth)
	defer dc.Close()
	defer lc.Close()

	go func() {
		for _, msg := range []string{"one", "two"} {
			_ = dc.SendMsg([]byte(msg))
		}
		_ = dc.Flush() // a send buffers; Flush writes
	}()
	for range 2 {
		f, err := lc.Recv()
		if err != nil {
			panic(err)
		}
		fmt.Println(string(f.Msg))
	}
}

Output:
one
two

func Client

func Client(ctx context.Context, nc net.Conn, cfg Config) (*Conn, error)

Client binds nc, whose security (TLS) is the caller's, as the dialer: it sends the hello with cfg.Auth's credential and reads the listener's reply. A listener that does not speak the version answers with a VersionError (errors.Is ErrVersion) naming the versions it speaks; a listener of link v0.4 or older that speaks only version 1 closes the connection after it read 37 bytes. A refusal is ErrRefused. On error the caller closes nc.

nc should have CloseWrite() error, as *net.TCPConn and *tls.Conn have (a wrapper of one passes it on): a graceful close (a drain, a Pool's Close) ends nc's write side and waits for the peer's close. Without it, it closes nc, and a TCP connection closed with bytes still arriving resets, losing the bytes in flight to the peer.

Example

ExampleClient binds the dialer's end of a connection and Server the listener's (here an in-memory pipe; TLS, when wanted, is the caller's: wrap the connection first). The listener learns who dialled it from its Authenticator.

package main

import (
	"context"
	"fmt"
	"net"
	"time"

	"github.com/nfltr/link"
)

func main() {
	a, b := net.Pipe()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	who := make(chan string, 1)
	go func() {
		c, id, err := link.Server(ctx, b, link.Config{Auth: link.NameAuth("listener")})
		if err != nil {
			b.Close()
			who <- err.Error()
			return
		}
		defer c.Close()
		who <- id.Subject
	}()
	c, err := link.Client(ctx, a, link.Config{Auth: link.NameAuth("dialer")})
	if err != nil {
		panic(err)
	}
	defer c.Close()
	fmt.Println("the listener knows the dialer as", <-who)
}

Output:
the listener knows the dialer as dialer

func (*Conn) Close

func (c *Conn) Close() error

Close closes the connection; senders waiting for credit give up. ConnDown gets why it ended: ErrDeadPeer, what the peer or the network did (lost), or nil when this side closed it; SessionClose, the same error, for each session still open, before it.

func (*Conn) Credit

func (c *Conn) Credit() *Credit

Credit is the connection's per-session flow control: a caller with its own receive loop acquires credit before SendData (Credit.Acquire) and acknowledges what it consumed (Credit.Consumed).

func (*Conn) Done

func (c *Conn) Done() <-chan struct{}

Done is closed once the connection is closed.

func (*Conn) Drained

func (c *Conn) Drained() <-chan struct{}

Drained is closed once the peer answered a ping this side sent after its GOAWAY (Mux.Drain sends one at once): the peer has read the GOAWAY, so it opens nothing more here, and every OPEN it sent before has arrived (frames keep their order). An OPEN that crossed the GOAWAY is a session like any other; one after Drained breaks the protocol. A peer before v0.4.2 answers only one of two pings that arrive together: then a later ping of the liveness's closes it, or the drain deadline ends the connection.

func (*Conn) Err

func (c *Conn) Err() error

Err is why the connection ended when the link ended it (ErrDeadPeer), else nil.

func (*Conn) Flush

func (c *Conn) Flush() error

Flush writes the buffered frames: inline when no write is in flight, else the write in flight takes them next. The buffer goes back to the pool once written, so an idle connection holds none.

func (*Conn) GoAway

func (c *Conn) GoAway(drain time.Duration, reason string) error

GoAway tells the peer to open no more sessions on this connection: the open ones may run for drain, then this side closes it. It also stops this side opening more (ErrGoAway). It writes at once.

func (*Conn) GoingAway

func (c *Conn) GoingAway() bool

GoingAway reports whether either side sent GOAWAY.

func (*Conn) Identity

func (c *Conn) Identity() Identity

Identity is who the peer is, as the listener's Authenticator said (zero on the dialer).

func (*Conn) PeerGoAway

func (c *Conn) PeerGoAway() <-chan struct{}

PeerGoAway is closed once the peer sent GOAWAY.

func (*Conn) PeerMaxSessions

func (c *Conn) PeerMaxSessions() int

PeerMaxSessions is how many sessions this side may have open toward the peer at once.

func (*Conn) Recv

func (c *Conn) Recv() (Frame, error)

Recv returns the next data, message, open or close frame; it applies the others itself and charges data frames against their session's window. A peer that breaks the credit rules gets ErrCredit (credit.go); the caller closes the connection on any error. A connection the link ended returns why (ErrDeadPeer).

func (*Conn) SendClose

func (c *Conn) SendClose(sid string, half bool) error

SendClose buffers a CLOSE frame for sid: half, the side sends no more of the session and still takes the peer's; else it neither sends nor takes more, and the session's credit ends.

func (*Conn) SendData

func (c *Conn) SendData(sid string, data []byte) error

SendData buffers sid's data as data frames, as SendMsg does. The caller acquired credit for it (Credit().Acquire). On a FeatureChain connection the first data or close of a session the peer opened is preceded by an ACCEPT when nothing answered it yet (a Mux's Session answers itself).

func (*Conn) SendMsg

func (c *Conn) SendMsg(b []byte) error

SendMsg buffers a message frame; Flush, or a full buffer, writes it. The caller does not modify b afterwards. The peer reads it from Recv (Frame.Msg); a Mux on the peer drops it, so messages are for callers that drive the Conn themselves.

func (*Conn) SendOpen

func (c *Conn) SendOpen(sid string, header []byte) error

SendOpen buffers an OPEN frame for session sid, with header (the opener's bytes, at most 64 KiB), and starts the session's credit. It refuses a session past the peer's max sessions (ErrFull) and any on a connection that is going away (ErrGoAway). On a FeatureChain connection the frame is a CHAIN-OPEN with a zero Chain.

func (*Conn) SendReset

func (c *Conn) SendReset(sid string) error

SendReset is SendClose of the whole session saying it failed: a peer that reads CLOSE code 2 (link v0.3.0 and later) reads it as a reset, an older one as a close.

func (*Conn) Version

func (c *Conn) Version() int

Version is the connection's wire version: 2, the only one since v0.5.0.

type Credit

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

Credit is the per-session flow control of one connection. A Conn has one and carries its window updates in window frames; a caller whose sessions travel on a transport of its own (a gRPC stream) makes one with NewCredit and carries the updates itself (AppendWindow, ParseWindow). Its methods are safe from any goroutine.

func NewCredit

func NewCredit(window, peerWindow int, grant func(sid string, n int)) *Credit

NewCredit returns the credit of a connection whose side buffers window bytes per session and whose peer buffers peerWindow. grant sends the peer a window update for n bytes of sid (AppendWindow); it is called from Consumed, never with a lock held.

func (*Credit) Acquire

func (c *Credit) Acquire(sid string, n int) bool

Acquire waits until sid may send n more bytes and takes them. n must not exceed the peer's window (MaxChunk does not). It reports false once the session is forgotten or the credit closed.

Before it waits it flushes: the peer returns credit only for bytes it received, so a sender waiting with its own frames still buffered (a Session.Write of several chunks) would wait for good.

func (*Credit) Close

func (c *Credit) Close()

Close makes every sender waiting in Acquire, and every later Acquire, give up.

func (*Credit) Consumed

func (c *Credit) Consumed(sid string, n int)

Consumed acknowledges n bytes of sid that left the receive buffer; a window update goes out once half a window has.

func (*Credit) Forget

func (c *Credit) Forget(sid string)

Forget ends a session's credit: a sender waiting in Acquire gives up.

func (*Credit) Grant

func (c *Credit) Grant(sid string, n int) error

Grant adds a window update's bytes to sid's credit, which never exceeds the peer's window.

func (*Credit) Open

func (c *Credit) Open(sid string)

Open starts the credit of a session on both sides: the opener before it sends the open frame, the other side when it receives it. Opening an id that is open starts it again: its sender waiting in Acquire gives up.

func (*Credit) PeerWindow

func (c *Credit) PeerWindow() int

PeerWindow is the other side's window.

func (*Credit) Received

func (c *Credit) Received(sid string, n int) error

Received charges n bytes of data for sid against this side's window: a Conn does for data frames, a caller for data it carries otherwise. Data of a session this side does not know is not buffered by anyone and is not charged.

func (*Credit) Window

func (c *Credit) Window() int

Window is this side's window.

type ForwardConfig

type ForwardConfig struct {
	MaxHops       uint8         // refuse an open that crossed this many joins (CodeHopLimit); 0: 8
	AnswerTimeout time.Duration // refuse when no onward answer came by then (CodeTimeout); 0: wait while in is open
	// Observer gets the Hop events; nil: link.Forward uses in's
	// connection's Observer, peerlink.Forward emits none.
	Observer Observer
}

ForwardConfig is a forwarding hop's settings.

type Frame

type Frame struct {
	SID    string
	Data   []byte
	Msg    []byte
	Open   bool
	Header []byte
	Close  bool
	Half   bool
	Reset  bool // a close that says the session failed (code 2, or an ABORT)
	// FeatureChain connections only. Chain: an open's (CHAIN-OPEN); an
	// accept's hops in Hops. Accept: the accepter's answer (ACCEPT).
	// Abort: an ABORT, which also sets Close and Reset.
	Chain  Chain
	Accept bool
	Abort  *Abort
}

Frame is a received frame: a session's data (SID, Data), a message (Msg, the caller's bytes), a session's open (SID, Open, Header: the opener's bytes) or close (SID, Close; Half when the peer only stopped sending). Window, ping, pong and goaway frames never surface: Recv applies them.

type HopEvent

type HopEvent struct {
	Kind    HopKind
	Chain   Chain         // the trace and this hop's Hops (the inbound session's)
	Latency time.Duration // HopAnswered: Route called -> onward answer
	Err     error         // HopAnswered: the refusal; HopEnded: the abort, nil on a clean end
	In, Out int64         // HopEnded: bytes origin->far and far->origin
}

HopEvent is one hop's event on a path of joined sessions (Observer.Hop).

type HopKind

type HopKind uint8

HopKind is a HopEvent's kind.

const (
	HopOpened   HopKind = iota // Forward took an open
	HopAnswered                // the onward answer came (or Forward refused)
	HopEnded                   // the joined sessions ended
)

Hop event kinds.

type Identity

type Identity struct{ Method, Subject string }

Identity is who the peer is, as the listener's Authenticator says.

type Liveness

type Liveness struct {
	Interval time.Duration
	Timeout  time.Duration
}

Liveness is how a connection notices its peer is gone without a FIN (it lost power, a NAT dropped the path): the peer sends a frame at least every Interval while idle, and the connection is dead once Interval + Timeout pass without any frame. Where one side sends the frames (its heartbeat), the other asks for Interval, the sender states the interval it keeps, and the asker waits DeadAfter(stated).

func ParseLiveness

func ParseLiveness(spec string) (Liveness, error)

ParseLiveness reads a preset name, raw values ("interval=2s,timeout=6s"), or a preset with raw values over it ("mobile,timeout=90s"); "" is LivenessDefault. The result is validated.

Example

ExampleParseLiveness: the presets, and a preset with an override (from a flag, say). A connection is dead once nothing arrived for Interval + Timeout; idle, each side sends a frame every Interval.

package main

import (
	"fmt"

	"github.com/nfltr/link"
)

func main() {
	for _, spec := range []string{"fast", "default", "nat", "datacenter", "mobile", "nat,timeout=40s"} {
		l, err := link.ParseLiveness(spec)
		if err != nil {
			panic(err)
		}
		fmt.Printf("%-16s a frame every %v idle, dead after %v\n", spec, l.Interval, l.Interval+l.Timeout)
	}
}

Output:
fast             a frame every 1.5s idle, dead after 4.5s
default          a frame every 5s idle, dead after 15s
nat              a frame every 10s idle, dead after 30s
datacenter       a frame every 15s idle, dead after 45s
mobile           a frame every 15s idle, dead after 1m0s
nat,timeout=40s  a frame every 10s idle, dead after 50s

func (Liveness) DeadAfter

func (l Liveness) DeadAfter(stated time.Duration) time.Duration

DeadAfter is how long to wait for any frame once the peer stated the interval it keeps: stated + Timeout, and never under three stated intervals (a peer that keeps another interval than the one asked for, or skips a frame after data). 0 when the peer stated none: liveness is then the transport keepalive's.

func (Liveness) String

func (l Liveness) String() string

String is the spec ParseLiveness reads back.

func (Liveness) Validate

func (l Liveness) Validate() error

Validate refuses values a heartbeat sender would not keep or that would take a live connection for a dead one.

type Mux

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

Mux carries sessions over a Conn, each a net.Conn. Either side opens sessions (Open), the other takes them (Serve's accept). A session's data goes on credit, so a session whose reader is slow stalls only itself; a Write goes to the network inline, or with the frames of other sessions when a write is in flight (Conn).

Opens and closes are OPEN and CLOSE frames, the open carrying the opener's header, and the Mux picks ids as uvarints, odd on the dialer and even on the listener, so the two sides never pick the same.

Message frames (Conn.SendMsg) are a Conn-level feature for callers that run their own receive loop; a Mux owns the Conn's receive loop and drops them. Per-peer control traffic on a Mux goes in a session whose header names its purpose.

Example

ExampleMux carries sessions over one connection without a Pool: each side wraps its Conn in a Mux, the listener serves the sessions the dialer opens, and each session is a net.Conn.

dc, lc := bind(tokenAuth("s3cret"))

go link.NewMux(lc).Serve(func(header []byte, s *link.Session) {
	defer s.Close()
	n, _ := io.Copy(io.Discard, s) // until the opener's CloseWrite
	fmt.Fprintf(s, "%s: %d bytes", header, n)
})

m := link.NewMux(dc)
go m.Serve(nil) // the dialer takes no sessions
defer m.Close()
s, err := m.Open([]byte("count"))
if err != nil {
	panic(err)
}
_, _ = io.WriteString(s, "hello, world")
_ = s.CloseWrite()
reply, _ := io.ReadAll(s)
_ = s.Close()
fmt.Println(string(reply))

Output:
count: 12 bytes

func NewMux

func NewMux(c *Conn) *Mux

NewMux returns a Mux over c. Call Serve on it: Serve is the receive loop, which a side that takes no sessions runs too (Serve(nil)), or no credit returns and its writes stall.

func (*Mux) Close

func (m *Mux) Close() error

Close closes the connection: Serve returns and every session ends.

func (*Mux) Conn

func (m *Mux) Conn() *Conn

Conn is the Mux's connection.

func (*Mux) Drain

func (m *Mux) Drain(ctx context.Context) error

Drain closes the connection gracefully: it tells the peer to open no more sessions here (GOAWAY) and stops this side opening more (Open: ErrGoAway); the open sessions run to their end (an OPEN the peer sent before it read the GOAWAY runs like the others), then the connection closes with every frame written. Drain returns once it closed, nil; at ctx's end it closes the connection at once and returns ctx's error. Serve must run (it is the receive loop).

func (*Mux) Open

func (m *Mux) Open(header []byte) (*Session, error)

Open starts a session with header, the bytes the peer's accept receives (the OPEN frame's header, at most 64 KiB). Open sends it at once and does not wait for the peer, so a peer that speaks first is reached without the caller's bytes.

On a connection that negotiated FeatureChain the open is a CHAIN-OPEN from the origin (OpenChain with a zero Chain) and the peer answers it (Answered).

func (*Mux) OpenChain

func (m *Mux) OpenChain(header []byte, c Chain) (*Session, error)

OpenChain is Open of a session on a path (Forward's Route opens the next leg with the Chain its ctx carries): a zero Trace is a new path, for which link picks a random one. On a connection without FeatureChain it is Open (the Chain is kept locally only) and the session does not answer (Answers is false).

func (*Mux) Serve

func (m *Mux) Serve(accept func(open []byte, s *Session)) error

Serve receives frames until the connection fails, then closes it and ends every session. accept, nil on a side that takes no sessions, runs in a goroutine of its own for each session the peer opens, with the header that opened it; it owns the session and closes it.

func (*Mux) Sessions

func (m *Mux) Sessions() int

Sessions is how many sessions are open.

type NopObserver

type NopObserver struct{}

NopObserver ignores every event.

func (NopObserver) ConnDown

func (NopObserver) ConnDown(Identity, error)

ConnDown ignores the event.

func (NopObserver) ConnUp

func (NopObserver) ConnUp(Identity)

ConnUp ignores the event.

func (NopObserver) CreditWait

func (NopObserver) CreditWait(time.Duration)

CreditWait ignores the event.

func (NopObserver) GoAway

func (NopObserver) GoAway(Identity)

GoAway ignores the event.

func (NopObserver) Hop

func (NopObserver) Hop(HopEvent)

Hop ignores the event.

func (NopObserver) PingRTT

func (NopObserver) PingRTT(time.Duration)

PingRTT ignores the event.

func (NopObserver) SessionClose

func (NopObserver) SessionClose(error)

SessionClose ignores the event.

func (NopObserver) SessionOpen

func (NopObserver) SessionOpen()

SessionOpen ignores the event.

type Observer

type Observer interface {
	ConnUp(Identity)          // the handshake bound the connection
	ConnDown(Identity, error) // it ended: ErrDeadPeer, the peer's or network's error, nil when this side closed it
	SessionOpen()             // a session opened by an OPEN frame
	SessionClose(error)       // one of those ended, or its connection did (the ConnDown error, before ConnDown)
	CreditWait(time.Duration) // a sender waited this long for credit
	PingRTT(time.Duration)    // a ping's round trip
	GoAway(Identity)          // the peer sent GOAWAY
	Hop(HopEvent)             // a hop of a path of joined sessions: Forward, Join (since v0.10.0)
	// contains filtered or unexported methods
}

Observer receives a connection's events, to map onto the caller's metrics. Calls come from the connection's goroutines and must not block. An implementation embeds NopObserver (an unexported method makes it), so Observer may gain methods in minor releases: an implementation then ignores the new events until it defines them.

type OwnedWriter

type OwnedWriter interface {
	WriteOwned(b []byte) error
}

OwnedWriter is a writer that may keep the slices it is given: WriteTo hands it the received slices themselves, no copy.

type Pool

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

Pool keeps connections to one peer and opens sessions on them: on the one with the fewest sessions, dialling more as Grow says. A connection the peer sends GOAWAY on takes no more sessions; its open ones run on. Drain replaces every connection make-before-break. Sessions the peer opens on the Pool's connections go to Accept (nil: refused).

func NewPool

func NewPool(cfg PoolConfig) (*Pool, error)

NewPool returns a Pool; it dials its Min connections in the background. It refuses a PoolConfig that Validate refuses.

Example

ExampleNewPool: a Pool dials connections to one peer as its sessions need them, binds each (Client) and opens sessions on them. Here each dial is an in-memory pipe whose other end a listener binds (Server) and serves, echoing each session.

package main

import (
	"context"
	"fmt"
	"io"
	"net"
	"time"

	"github.com/nfltr/link"
)

func main() {
	p, err := link.NewPool(link.PoolConfig{
		Link: link.Config{Auth: link.NameAuth("dialer")},
		Dial: func(context.Context) (net.Conn, error) {
			a, b := net.Pipe()
			go func() {
				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
				c, _, err := link.Server(ctx, b, link.Config{Auth: link.NameAuth("listener")})
				cancel()
				if err != nil {
					b.Close()
					return
				}
				_ = link.NewMux(c).Serve(func(header []byte, s *link.Session) {
					defer s.Close()
					_, _ = io.Copy(s, s) // until the opener's CloseWrite
				})
			}()
			return a, nil
		},
	})
	if err != nil {
		panic(err)
	}
	defer p.Close()

	s, err := p.Open(context.Background(), []byte("echo"))
	if err != nil {
		panic(err)
	}
	_, _ = io.WriteString(s, "hello")
	_ = s.CloseWrite()
	reply, _ := io.ReadAll(s)
	_ = s.Close()
	fmt.Println(string(reply))
}

Output:
hello

func (*Pool) Close

func (p *Pool) Close() error

Close closes every connection; Open fails from then on. A connection that carries no session closes once the frames already written on it went out (the tails and closes of the sessions that ended there reach the peer), the others at once.

func (*Pool) Drain

func (p *Pool) Drain(ctx context.Context) error

Drain replaces the Pool's connections make-before-break: it dials a replacement (Min of them, at least one) first, then sends GOAWAY on every old connection, which then takes no new session and closes once its open sessions ended, at ctx's deadline at the latest. It returns once they all closed. A replacement that cannot be dialled leaves the old connections as they were.

func (*Pool) Open

func (p *Pool) Open(ctx context.Context, header []byte) (*Session, error)

Open opens a session with header on one of the Pool's connections, with the Chain ctx carries (ContextWithChain; Mux.OpenChain) when it carries one.

func (*Pool) Stats

func (p *Pool) Stats() PoolStats

Stats is the Pool's load.

type PoolConfig

type PoolConfig struct {
	// Link is each connection's Config.
	Link Config
	// Dial opens a connection to the peer, its security done (TLS, a
	// proxy): the caller's.
	Dial func(ctx context.Context) (net.Conn, error)
	// Min connections are kept open, redialled with Backoff when one ends;
	// Max (0 is 1) is the most open at once, not counting those going away
	// (GOAWAY), which take no new session.
	Min, Max int
	// Grow, when set, says whether Open dials one more connection (below
	// Max) although one could take the session: in the background, one
	// such dial at a time. nil grows only when none can.
	Grow func(PoolStats) bool
	// GrowWait is how long the Open that started a growth dial waits for
	// it: the session goes on the new connection when it is up by then,
	// else on one that is up (the dial goes on). 0: it does not wait.
	// Sessions opened in a burst spread over the connections only when
	// the Open waits.
	GrowWait time.Duration
	// Backoff is how long to wait before redialling after attempt failed
	// dials (from 1). nil: Open fails with the dial's error, and Min is
	// redialled once a second.
	Backoff func(attempt int) time.Duration
	// Accept, when set, takes the sessions the peer opens on the Pool's
	// connections (Mux.Serve's accept: it owns the session and closes
	// it). nil: they are refused.
	Accept func(open []byte, s *Session)
}

PoolConfig is a Pool's settings. Every policy is the caller's.

func (PoolConfig) Validate

func (cfg PoolConfig) Validate() error

Validate reports the first field of cfg that NewPool refuses, naming it and what to set: Dial is required; Min and Max are not negative, and Min is at most a Max that is set; Link is valid (Config.Validate).

type PoolStats

type PoolStats struct{ Conns, Sessions int }

PoolStats is a Pool's load: open connections that take sessions, and their sessions.

type Route

type Route func(ctx context.Context, in *Session) (*Session, error)

Route opens the session in continues on (on a Mux, a Pool or anything else that gives a *Session), or returns why not: an *Abort (its Code and Msg answer in), ErrGoAway (CodeDraining) or any other error (CodeNoRoute, with its text). ctx carries in's Chain with Hops+1 (Pool.Open and Mux.OpenChain read it: ChainFromContext) and ends when in ends, when ForwardConfig's AnswerTimeout passes, or once the onward answer came. A Route may wait for the onward answer itself (Answered) to try another next hop.

type Session

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

Session is one session of a Mux. Read and WriteTo take the peer's bytes in order, then io.EOF once the peer closed the session (or its write side), or the error that cut it: ErrReset (the peer's Reset), or ErrConnClosed wrapping why the connection ended (ErrDeadPeer, the network's error). Write sends on credit; past such an end it returns io.ErrClosedPipe wrapping the same error. A read deadline ends a pending Read, and a write deadline a pending Write, with os.ErrDeadlineExceeded.

func (*Session) Abort

func (s *Session) Abort(code uint32, msg string) error

Abort resets the session with a reason, either side, any time: the peer reads an *Abort with code and msg (errors.Is ErrReset). Reads here end as after Close. On a connection without FeatureChain it is Reset.

func (*Session) AbortWith

func (s *Session) AbortWith(a *Abort) error

AbortWith is Refuse (a.Refused) or Abort with every field of a: a hop passing on an abort from the other side of a join sets Hops (the abort's own Hops plus one).

func (*Session) Accept

func (s *Session) Accept() error

Accept answers a session the peer opened: accepted. The first Write, WriteBuffers, CloseWrite or Close answers so too; a side that may wait before writing (a far end that dialled its target) calls it, since the opener's Answered waits for it. It does nothing on a session this side opened, on a connection without FeatureChain, or once answered.

func (*Session) AcceptPath

func (s *Session) AcceptPath(pathLen uint8) error

AcceptPath is Accept for a hop that joined the session to a path pathLen hops long beyond it (Forward: the onward session's PathLen plus one): the opener's PathLen.

func (*Session) Answered

func (s *Session) Answered(ctx context.Context) error

Answered waits for the peer's answer to a session this side opened: nil once it accepted (at the end of a path of Forward hops: once the far end accepted), the *Abort of its refusal (errors.Is ErrOpenRefused), or the error that ended the session before an answer (a reset's *Abort, the connection's end), or ctx's error. ErrNoAnswers on a connection without FeatureChain; nil on a session the peer opened. Bytes written before the answer are sent (no round trip per hop); toward a refused session they are discarded.

func (*Session) Answers

func (s *Session) Answers() bool

Answers reports whether the session's connection negotiated FeatureChain: the accepter answers the open (Accept, Refuse), aborts carry a reason, and the open carried a Chain.

func (*Session) Chain

func (s *Session) Chain() Chain

Chain is where the session is on its path: the CHAIN-OPEN's for a session the peer opened, OpenChain's (its Trace picked) for one this side opened. Zero on a connection without FeatureChain, but for OpenChain's own.

func (*Session) Close

func (s *Session) Close() error

Close ends the session: the peer gets a close unless it closed it; a pending Read returns.

func (*Session) CloseWrite

func (s *Session) CloseWrite() error

CloseWrite ends the session's write side: the peer reads io.EOF after what was written; this side still reads the peer's bytes. The session ends once both sides closed their write side.

func (*Session) Done

func (s *Session) Done() <-chan struct{}

Done is closed once the session ended: both sides closed it, the peer reset it, or its connection closed. Bytes already received stay readable.

func (*Session) Header

func (s *Session) Header() []byte

Header is the bytes that opened the session: the OPEN frame's header.

func (*Session) LocalAddr

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

LocalAddr is the connection's.

func (*Session) PathLen

func (s *Session) PathLen() uint8

PathLen is, once Answered returned nil, the hops beyond the peer to the session's far end (the ACCEPT's): 0 when the peer is the far end.

func (*Session) Pump

func (s *Session) Pump(w io.Writer, done func(n int64, err error))

Pump writes the peer's bytes to w as WriteTo does, without a goroutine waiting while the session is idle: one runs only while bytes are queued, started as they arrive (the receive loop never waits for it). done gets what WriteTo would return, once, from that goroutine: the session ended, or w failed (the session's further bytes stay unread: close it). Pump takes over the session's reads: no Read, WriteTo or second Pump after it.

func (*Session) Read

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

Read reads the peer's bytes in order: io.EOF once the peer ended its writes, or the error that cut the session (see Session).

func (*Session) Refuse

func (s *Session) Refuse(code uint32, msg string) error

Refuse answers a session the peer opened: no, with code (CodeApp and up are the application's) and msg (cut to MaxAbortMsg bytes of UTF-8). The opener's Answered and reads return the *Abort. The session ends. It fails once the session was answered (Abort it instead). On a connection without FeatureChain it is Reset.

func (*Session) RemoteAddr

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

RemoteAddr is the connection's.

func (*Session) Reset

func (s *Session) Reset() error

Reset ends the session as failed: the peer reads ErrReset (a peer of link v0.3.0 or later, which reads CLOSE code 2; an older one reads a close). Reads here end as after Close.

func (*Session) SetDeadline

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

SetDeadline sets the read and the write deadline.

func (*Session) SetReadDeadline

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

SetReadDeadline makes a pending or later Read return os.ErrDeadlineExceeded once t passed; zero t clears it. WriteTo ignores it.

func (*Session) SetWriteDeadline

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

SetWriteDeadline makes a pending or later Write (WriteBuffers) return os.ErrDeadlineExceeded once t passed; zero t clears it. While one is set, writes leave the network write to a goroutine (see Write).

func (*Session) Write

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

Write sends b to the peer in chunks it has credit for, waiting for credit, then writes them to the network (Flush). With a write deadline set, the frames go to the network from a goroutine, and Write returns os.ErrDeadlineExceeded, with the bytes it sent, once the deadline passed while it waited for credit or for room in the connection's write buffer.

func (*Session) WriteBuffers

func (s *Session) WriteBuffers(bufs [][]byte) (int64, error)

WriteBuffers is Write of each of bufs in turn, with one flush at the end (and before any wait for credit): a backlog goes to the network in one write.

func (*Session) WriteTo

func (s *Session) WriteTo(w io.Writer) (int64, error)

WriteTo writes the peer's bytes to w until the session ends; a backlog goes in one net.Buffers write (writev on a TCP conn), or slice by slice to an OwnedWriter.

type Trace

type Trace [16]byte

Trace identifies a path of joined sessions end to end: the origin picks it and every hop carries it unchanged. It is for correlation (logs, Observer.Hop) only: not a secret, not an authenticator.

type VersionError

type VersionError struct{ Max int }

VersionError is a listener's refusal of a hello's version: it speaks versions up to Max.

func (*VersionError) Error

func (e *VersionError) Error() string

Error says which versions the peer speaks.

func (*VersionError) Is

func (e *VersionError) Is(target error) bool

Is makes a VersionError an ErrVersion.