link

README

link

github.com/nfltr/link is the nfltr endpoint-to-endpoint link layer: many sessions, each a byte stream with its own flow control, multiplexed over one connection the caller dials and secures (TLS, a proxy, a pipe). A session whose reader is slow stalls only itself; one write to the network carries every frame queued at the time.

• Standard library only (TestModuleBoundary).
• Apache-2.0 (LICENSE, NOTICE; every source file carries an SPDX header).
• Wire format version 1 is specified in doc.go; testdata/v1_session.golden holds a recorded session every release is tested against, byte for byte.

Packages

package	what
link	Hello/ReadHello/Reply (the version-1 handshake), Conn (frames, per-session credit, write coalescing), Mux and Session (sessions as net.Conns; the caller's Codec encodes opens and closes), Liveness and its presets
link/bytequeue	a FIFO of byte slices bounded by total bytes

Use

window, err := link.Hello(nc, token, 16<<20)  // dialer
c := link.NewConn(nc, 16<<20, window)
m := link.NewMux(c, codec)
go m.Serve(nil)
s, err := m.Open(func(sid string) ([]byte, error) { return codec.Open(sid, header) })

Checks

make check: build, vet, tests with -race, fuzz-short (every decoder, FUZZTIME each, default 10s), gosec, semgrep (docker), govulncheck.

Status

Phase 1 of the link package (version-1 wire, extracted unchanged from the nfltr's TCP data plane). Planned: wire version 2 (OPEN, CLOSE, PING, GOAWAY frames, link-chosen session ids), credit on every path, Pool (several connections per peer, make-before-break drain), Authenticator, Observer.

Documentation

Overview

Package link is the nfltr endpoint-to-endpoint link layer: many sessions, each a byte stream with its own flow control, multiplexed over one connection the caller dials and secures (TLS, a proxy, a pipe). A session whose reader is slow stalls only itself; one write to the network carries every frame queued at the time.

It imports only the Go standard library.

Wire format, version 1

hello  dialer -> listener: "NFDP" version(1) token(32 bytes) window(4)
reply  listener -> dialer: status(1), 0 = bound; window(4) when bound
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)
  type 2, window:  sid_len(1) sid increment(4)

Integers are big endian. A session id (sid) is at most 255 bytes, chosen by the side that opens the session. A window is how many bytes of one session the side that announces it buffers (see Conn.Acquire). In version 1 the frames that open and close a session are message frames in the caller's encoding (Codec); session ids that begin with '_' name control frames, which Mux leaves to the caller.

testdata/v1_session.golden holds a recorded version-1 session; the tests hold every release to those bytes.

Index

• Constants
• Variables
• func CheckWindow(w uint32) (int, error)
• func Hello(nc net.Conn, token string, window int) (int, error)
• func ReadHello(r io.Reader) (string, int, error)
• func Reply(w io.Writer, status byte, window int) error
• type Codec
• type Conn
  • func NewConn(nc net.Conn, window, peerWindow int) *Conn
  • func (c *Conn) Acquire(sid string, n int) bool
  • func (c *Conn) Close() error
  • func (c *Conn) Consumed(sid string, n int)
  • func (c *Conn) Flush() error
  • func (c *Conn) Forget(sid string)
  • func (c *Conn) Open(sid string)
  • func (c *Conn) PeerWindow() int
  • func (c *Conn) Received(sid string, n int) error
  • func (c *Conn) Recv() (Frame, error)
  • func (c *Conn) SendData(sid string, data []byte) error
  • func (c *Conn) SendMsg(b []byte) error
  • func (c *Conn) Window() int
• type Frame
• 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 Message
• type Mux
  • func NewMux(c *Conn, codec Codec) *Mux
  • func (m *Mux) Close() error
  • func (m *Mux) Open(open func(sid string) ([]byte, error)) (*Session, error)
  • func (m *Mux) Serve(accept func(open []byte, s *Session)) error
• type OwnedWriter
• type Session
  • func (s *Session) Close() error
  • func (s *Session) LocalAddr() net.Addr
  • func (s *Session) Read(b []byte) (int, error)
  • func (s *Session) RemoteAddr() net.Addr
  • func (s *Session) SetDeadline(time.Time) error
  • func (s *Session) SetReadDeadline(time.Time) error
  • func (s *Session) SetWriteDeadline(time.Time) error
  • func (s *Session) Write(b []byte) (int, error)
  • func (s *Session) WriteTo(w io.Writer) (int64, error)

Constants

const (
	StatusBound   = 0
	StatusRefused = 1
)

Status bytes of the listener's reply to a hello.

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

Window bounds: a side announces a window in [MinWindow, MaxWindow]. Credit returns in steps of half a window, which must hold a MaxChunk.

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 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 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: Open refuses more (ErrFull), Serve closes a session the peer opens past it.

const TokenLen = 32

TokenLen is the length of a hello's token: the hex of 16 random bytes.

const Version = 1

Version is the hello's protocol version.

Variables

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 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 ErrFrame = errors.New("link: malformed frame")

ErrFrame reports bytes that break the frame format.

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

ErrFull is Open's refusal of a session past MaxSessions.

var ErrRefused = errors.New("link: peer refused the connection")

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

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

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

Functions

func CheckWindow

func CheckWindow(w uint32) (int, error)

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

func Hello

func Hello(nc net.Conn, token string, window int) (int, error)

Hello binds a new connection with token, announcing window, and returns the listener's window from its reply.

func ReadHello

func ReadHello(r io.Reader) (string, int, error)

ReadHello reads the rest of a hello, after its Magic, and returns the token and the dialer's window.

func Reply

func Reply(w io.Writer, status byte, window int) error

Reply answers a hello; the reply of a bound connection announces window.

Types

type Codec

type Codec interface {
	// Close returns the message that closes session sid.
	Close(sid string) ([]byte, error)
	// Parse reads a received message: the session it names, whether it
	// opens or closes it, and data it carries for the session.
	Parse(msg []byte) (Message, error)
}

Codec reads and writes the messages that open and close sessions: in wire version 1 their encoding is the caller's.

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 are safe from any.

Reads go straight to the connection: over TLS, tls.Conn's record buffer serves the small header reads, and an idle connection pins no buffer. ponytail: a plaintext connection (loopback, dev) reads the socket for each header; give it a bufio.Reader if one carries real traffic.

func NewConn

func NewConn(nc net.Conn, window, peerWindow int) *Conn

NewConn returns a Conn over nc, whose hello is done: window is this side's window, peerWindow the other side's.

func (*Conn) Acquire

func (c *Conn) 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 connection closed or failed.

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 (*Conn) Close

func (c *Conn) Close() error

Close closes the connection; senders waiting for credit give up.

func (*Conn) Consumed

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

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

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) Forget

func (c *Conn) Forget(sid string)

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

func (*Conn) Open

func (c *Conn) 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.

func (*Conn) PeerWindow

func (c *Conn) PeerWindow() int

PeerWindow is the other side's window. allow:wrapper

func (*Conn) Received

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

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

func (*Conn) Recv

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

Recv returns the next data or message frame; it applies window frames 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.

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 (Acquire).

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.

func (*Conn) Window

func (c *Conn) Window() int

Window is this side's window. allow:wrapper

type Frame

type Frame struct {
	SID  string
	Data []byte
	Msg  []byte
}

Frame is a received frame: a session's data (SID, Data), or a message (Msg, the caller's bytes). Window frames never surface: Recv applies them.

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.

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 Message

type Message struct {
	SID         string
	Open, Close bool
	Data        []byte
}

Message is a parsed message frame (Codec.Parse).

type Mux

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

Mux carries sessions over a Conn, each a net.Conn. One 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).

func NewMux

func NewMux(c *Conn, codec Codec) *Mux

NewMux returns a Mux over c whose open and close messages codec reads and writes.

func (*Mux) Close

func (m *Mux) Close() error

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

func (*Mux) Open

func (m *Mux) Open(open func(sid string) ([]byte, error)) (*Session, error)

Open starts a session: open returns the message that opens it, for the session id the Mux chose. Open sends it at once, so a peer that speaks first is reached without the caller's bytes.

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 the side that opens sessions, runs in a goroutine of its own for each session the peer opens, with the message that opened it; it owns the session and closes it.

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 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; Write sends on credit. Deadlines are not supported: Close ends a session.

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) LocalAddr

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

LocalAddr is the connection's. allow:wrapper

func (*Session) Read

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

func (*Session) RemoteAddr

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

RemoteAddr is the connection's. allow:wrapper

func (*Session) SetDeadline

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

SetDeadline and its siblings are no-ops: Close ends a session.

func (*Session) SetReadDeadline

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

func (*Session) SetWriteDeadline

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

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).

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.