v1

Documentation

Overview

Package v1 实现 MBTA v1（ALPN mbta/1）over QUIC + TLS 1.3 的传输 binding。

v1 是薄 transport binding：协议逻辑（1-RTT 握手/Envelope/RESULT+CREDIT/drain） 全部委托给 internal/protocol.CoreClient/CoreHandler，本包仅实现 QUIC 传输适配—— QUIC 多流（control stream + 单 data stream）+ QUIC DATAGRAM 不可靠通道。 帧路由按 type 高位通道域（core.IsControlType/IsDataType）。

默认 profile 为国际密码组合（X25519+HKDF-SHA256+AES-256-GCM）。 详见 docs/v2-design.md。

Index

• Constants
• type Client
  • func NewClient(cfg ClientConfig) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Connect(ctx context.Context) error
  • func (c *Client) SelectedCapabilities() []string
  • func (c *Client) SendBatch(ctx context.Context, signalBatch *core.SignalBatch, opts ...core.SendOption) (uint64, error)
  • func (c *Client) SendDatagram(ctx context.Context, signalBatch *core.SignalBatch, opts ...core.SendOption) (uint64, error)
  • func (c *Client) SessionID() string
  • func (c *Client) SetRedirectHandler(h func(context.Context, []byte))
  • func (c *Client) SetResultHandler(h func(seq uint64, level string, code uint32, correlation string))
  • func (c *Client) State() core.State
• type ClientConfig
• type ClientCredentials
• type Conn
  • func Dial(ctx context.Context, cfg QUICClientConfig) (*Conn, error)
  • func (c *Conn) AcceptStream(ctx context.Context) (*quic.Stream, string, error)
  • func (c *Conn) Close() error
  • func (c *Conn) CloseWithError(code quic.ApplicationErrorCode, reason string) error
  • func (c *Conn) OpenControlStream(ctx context.Context) (*quic.Stream, error)
  • func (c *Conn) OpenDataStream(ctx context.Context) (*quic.Stream, error)
  • func (c *Conn) SetAuthed(authed bool)
• type DataStream
• type Listener
  • func Listen(ctx context.Context, cfg QUICServerConfig) (*Listener, error)
  • func (l *Listener) Accept(ctx context.Context) (*Conn, error)
  • func (l *Listener) Addr() net.Addr
  • func (l *Listener) Close() error
• type QUICClientConfig
• type QUICServerConfig
• type Server
  • func NewServer(cfg ServerConfig) (*Server, error)
  • func (s *Server) Addr() string
  • func (s *Server) Close() error
  • func (s *Server) Config() ServerConfig
  • func (s *Server) Start(ctx context.Context) error
• type ServerConfig
• type ServerCredentials
• type StreamPicker
  • func NewSingleStream(ds DataStream) StreamPicker

Constants

const ALPNProtocol = "mbta/1"

ALPNProtocol is the Application-Layer Protocol Negotiation identifier for MBTA v1 over QUIC.

Types

type Client

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

Client is an MBTA agent that connects to a server and sends event batches.

Client 是薄包装：协议逻辑（状态机/握手/RESULT/drain/heartbeat）全部委托给 protocol.CoreClient；传输帧 I/O 下沉到 v1ClientTransport（QUIC 多流适配）。 自身只保留 QUIC 连接引用，用于「已连接守卫」与 Connect 失败兜底关闭。

实例单次使用：底层 CoreClient 的 lifecycle 由 closeOnce 保证一次性终结， Close 后或握手失败清理后不可复用——重连须新建 Client 实例。 例外：Connect 在 dial 阶段失败（网络不可达等）时状态机回退 Idle， 同实例重试 Connect 可用。

可靠投递语义：credit 账本仅在内存追踪已发送未终态的 batch。进程崩溃/重连后 未终态的 batch 会丢失——持久化与重发由调用方负责。

func NewClient

func NewClient(cfg ClientConfig) (*Client, error)

NewClient creates a new MBTA client.

func (*Client) Close

func (c *Client) Close() error

Close sends a CLOSE frame, drains pending results, and shuts down.

func (*Client) Connect

func (c *Client) Connect(ctx context.Context) error

Connect dials the server and completes the HELLO handshake (1-RTT).

ctx 仅控制握手阶段（Dial、开 stream、HELLO）的超时与取消。握手成功后， 后台 goroutine 运行在独立的 lifecycle ctx 上，不随 ctx 取消而退出；client 生命周期 由 Close() 终结。

func (*Client) SelectedCapabilities

func (c *Client) SelectedCapabilities() []string

SelectedCapabilities 返回握手协商选定的能力集（握手前为 nil）。

func (*Client) SendBatch

func (c *Client) SendBatch(ctx context.Context, signalBatch *core.SignalBatch, opts ...core.SendOption) (uint64, error)

SendBatch sends a SignalBatch through the MBTA protocol, 返回分配的 seq。

opts 携带 per-call 发送选项（如 core.WithTraceContext），透传给协议核心。

func (*Client) SendDatagram

func (c *Client) SendDatagram(ctx context.Context, signalBatch *core.SignalBatch, opts ...core.SendOption) (uint64, error)

SendDatagram 以 QUIC DATAGRAM 不可靠通道发送（at-most-once，豁免 credit）。 协商未含 datagram 能力或连接未启用 DATAGRAM 扩展时，降级为可靠 SendBatch 路径（消耗 credit、有 RESULT 反馈）——语义与直接调用 SendBatch 一致。

func (*Client) SessionID

func (c *Client) SessionID() string

func (*Client) SetRedirectHandler

func (c *Client) SetRedirectHandler(h func(context.Context, []byte))

SetRedirectHandler registers a callback invoked when the server sends a REDIRECT frame (S→C cluster redirect to the HA leader).

func (*Client) SetResultHandler

func (c *Client) SetResultHandler(h func(seq uint64, level string, code uint32, correlation string))

SetResultHandler 注册结果回调：服务端 RESULT（accepted/durable/rejected/released） 终态时调用，参数为 (seq, level, code, correlation)。

func (*Client) State

func (c *Client) State() core.State

type ClientConfig

type ClientConfig struct {
	Transport    QUICClientConfig  // QUIC 连接配置
	AgentID      string            // agent ID
	Token        string            // 认证 token（不上 wire，仅参与 token_proof HMAC）
	Meta         map[string]string // HELLO 元数据（hostname/os/agent_version 等，可选）
	Capabilities []string          // 客户端能力（与服务端 policy 协商取交集）
	// Profile 是应用层密码组合。0 = 默认 PROFILE_INTL（v1 binding 即国际套件）。
	Profile corepb.Profile

	Metrics      core.Metrics  // 可选：客户端可观测性指标（nil=NoOp）
	DrainTimeout time.Duration // Close 时等待未终态 batch 的上限。0=默认(30s)
}

ClientConfig holds configuration for an MBTA client.

type ClientCredentials

type ClientCredentials struct {
	CAFile             string // PEM-encoded CA certificate for server verification
	CertFile           string // PEM-encoded client certificate (optional, for mTLS)
	KeyFile            string // PEM-encoded client private key (optional, for mTLS)
	ServerName         string // expected server hostname for SNI
	InsecureSkipVerify bool   // skip TLS verification (dev only, never use in production)
}

ClientCredentials holds client-side TLS credentials.

type Conn

type Conn struct {
	QC         *quic.Conn
	RemoteAddr net.Addr
	// contains filtered or unexported fields
}

Conn wraps a QUIC connection with MBTA stream role tracking.

func Dial

func Dial(ctx context.Context, cfg QUICClientConfig) (*Conn, error)

Dial establishes a QUIC connection to an MBTA server.

func (*Conn) AcceptStream

func (c *Conn) AcceptStream(ctx context.Context) (*quic.Stream, string, error)

AcceptStream accepts a stream from the remote peer and assigns its role.

func (*Conn) Close

func (c *Conn) Close() error

Close 关闭连接：QUIC 层优雅关闭后补关本地 UDP socket（幂等）。

func (*Conn) CloseWithError

func (c *Conn) CloseWithError(code quic.ApplicationErrorCode, reason string) error

CloseWithError closes the QUIC connection with an error code.

func (*Conn) OpenControlStream

func (c *Conn) OpenControlStream(ctx context.Context) (*quic.Stream, error)

OpenControlStream opens the control stream. Must be called before OpenDataStream.

持锁检查 controlClaimed 并在解锁前置位，再 OpenStreamSync：消除检查与置位之间的 TOCTOU 窗口（控制流仅开一次，锁内调用无热路径竞争）。

func (*Conn) OpenDataStream

func (c *Conn) OpenDataStream(ctx context.Context) (*quic.Stream, error)

OpenDataStream opens a new data stream. Requires auth to be completed.

func (*Conn) SetAuthed

func (c *Conn) SetAuthed(authed bool)

SetAuthed marks the connection as authenticated.

type DataStream

type DataStream interface {
	io.Writer
	Index() int
}

DataStream represents an opened QUIC data stream with an integer index.

type Listener

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

Listener accepts MBTA QUIC connections.

func Listen

func Listen(ctx context.Context, cfg QUICServerConfig) (*Listener, error)

Listen creates a QUIC listener.

func (*Listener) Accept

func (l *Listener) Accept(ctx context.Context) (*Conn, error)

Accept waits for and returns the next QUIC connection.

func (*Listener) Addr

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

Addr returns the listener address.

func (*Listener) Close

func (l *Listener) Close() error

Close shuts down the listener.

type QUICClientConfig

type QUICClientConfig struct {
	Server      string             // server address (e.g. "localhost:7400")
	Credentials *ClientCredentials // TLS credentials (nil skips client cert)
	IdleTimeout time.Duration      // connection idle timeout
}

QUICClientConfig holds client QUIC configuration.

type QUICServerConfig

type QUICServerConfig struct {
	Address            string             // listen address (e.g. "0.0.0.0:7400")
	Credentials        *ServerCredentials // TLS credentials (nil uses default config)
	MaxIncomingStreams int64              // maximum concurrent QUIC streams
	IdleTimeout        time.Duration      // connection idle timeout
}

QUICServerConfig holds server QUIC configuration.

type Server

type Server struct {
	*binding.Server[*Listener, *Conn]
	// contains filtered or unexported fields
}

Server accepts and handles MBTA agent connections. 内嵌 binding.Server（消除与 ntls 的服务端外壳重复）， 自身仅保留 QUIC 专属的 Start/Addr/Close 入口。

func NewServer

func NewServer(cfg ServerConfig) (*Server, error)

NewServer creates a new MBTA server.

func (*Server) Addr

func (s *Server) Addr() string

Addr 返回服务器监听地址（Start 完成监听后有效；启动前返回空串）。

func (*Server) Close

func (s *Server) Close() error

Close shuts down the server.

func (*Server) Config

func (s *Server) Config() ServerConfig

Config 返回服务端配置（facade 测试与诊断用；返回副本）。

func (*Server) Start

func (s *Server) Start(ctx context.Context) error

Start begins listening for QUIC connections and runs the accept loop. Blocks until the context is cancelled. Each accepted connection is handled in its own goroutine via CoreHandler.Handle.

type ServerConfig

type ServerConfig struct {
	Transport QUICServerConfig
	Auth      core.TokenValidator // token 认证（必填，ResolveToken+Validate）
	Policy    core.ServerPolicy   // 服务端策略（零值 = DefaultServerPolicy(intl)）
	ServerID  string              // 服务端标识；空则 ResolveHandlerConfig 生成 ULID
	Metrics   core.Metrics        // 可选可观测性指标（nil=NoOp）

	RedirectChecker core.RedirectChecker // HA：握手完成后检查角色，非 leader 发 REDIRECT（可选）
	OnSession       core.OnSession       // 会话生命周期回调（可选，nil=禁用）

	// Sink 三件套：注册式投递面（替代旧 6 接口 sink 矩阵）。
	SinkOpts     core.SinkOptions // 供给偏好（Raw/Durable/Async）
	SinkHandler  core.BatchHandler
	SinkDeadline time.Duration // handler done 超时（0 = core.DefaultSinkDeadline）

	MaxConcurrentConns int   // 并发连接上限，0 = 使用 binding.DefaultMaxConcurrentConns
	SinkQueueSize      int   // sink 队列容量（0=默认 256，负=同步调用）
	SinkQueueMaxBytes  int64 // sink 队列字节上限（0=默认 32MB，负=不限）

	// Epochs 可选：连接纪元分配器（facade 双监听时共享一份，保证同 agent
	// 跨 v1/ntls 的纪元单调；nil = 本 server 独立分配）。
	Epochs *protocol.EpochAllocator
}

ServerConfig holds configuration for an MBTA server.

Policy 零值时 NewServer 回填 core.DefaultServerPolicy(PROFILE_INTL)——v1 binding 即国际密码组合（X25519+AES-GCM），默认策略与 binding 语义一致；显式配置则原样 经 binding.ResolveHandlerConfig 校验（不回填，避免半配置静默取默认）。

type ServerCredentials

type ServerCredentials struct {
	CertFile   string // PEM-encoded server certificate
	KeyFile    string // PEM-encoded server private key
	CAFile     string // PEM-encoded CA certificate for client verification (optional)
	ClientAuth string // Client certificate mode: "none", "request", "require-and-verify"
}

ServerCredentials holds server-side TLS credentials. Follows Go naming convention (similar to tls.Certificate).

type StreamPicker

type StreamPicker interface {
	Pick() (DataStream, error)
}

StreamPicker selects which data stream a batch should be sent on.

func NewSingleStream

func NewSingleStream(ds DataStream) StreamPicker

NewSingleStream 返回总是路由到给定单流的 StreamPicker。 传入 nil 会导致直接 panic；调用方应确保 ds 非 nil。