core

package
v0.6.0 Latest Latest
Warning

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

 Go to latest Published: Aug 22, 2026 License: MIT Imports: 41 Imported by: 0

Documentation

Overview

Package core 实现 MBTA 协议的核心原语（r3 重设计）：与传输无关的协议层基础组件。

主要组件：

  • 帧（frame.go）：[type][flags][varint len] wire 帧格式；type 高位为通道域。
  • 密钥日程（keyschedule.go）：ECDH + transcript 绑定 + HKDF split；密钥永不上 wire。
  • Envelope（envelope.go）：compress-then-encrypt、确定性 nonce、AD 绑定全部明文元数据。
  • 消息（message.go）：控制面消息校验（ValidateHello/HelloAck）。
  • credit（credit.go）：客户端 CreditLedger（模型 B）+ 服务端 AdmissionWindow（集合制窗口+终态环）。
  • session（session.go）：客户端 4 态 / 服务端 3 态状态机。
  • auth（auth.go）：TokenValidator（ResolveToken 必选）+ token_proof 验证。
  • cipher（cipher.go）：profile 密码组合（intl/gm）。
  • codec（codec*.go）：SignalBatch 编解码（proto 默认 / cbor）。
  • errors（errors.go）：数字错误码注册表（wire 只传数字；append-only）。
  • ulid（ulid.go）：会话 ID 生成。

Index

Constants

View Source 
const (
	HMACKeyLenIntl = 32 // HMAC-SHA-256
	HMACKeyLenGM   = 32 // HMAC-SM3
	AEADKeyLenIntl = 32 // AES-256-GCM
	AEADKeyLenGM   = 16 // SM4-GCM
)

各 profile 密钥长度（r3：profile 一次选定 ECDH+KDF+AEAD 组合）。

View Source 
const (
	DefaultCreditClampBatches = 1 << 16
	DefaultCreditClampBytes   = 1 << 32
)

DefaultCreditClamp* 是客户端对协商/追加 credit 的本地钳制上限。

View Source 
const (
	CodeOK uint32 = 0

	CodeAuthFailed            uint32 = 10 // 未认证维度唯一码（防 agent/token 枚举 oracle）
	CodeUnsupportedProfile    uint32 = 12
	CodeUnsupportedCapability uint32 = 13
	CodeBadHandshakeMessage   uint32 = 14

	CodeBatchTooLarge     uint32 = 20 // 解压后 payload 超 limits.max_batch_bytes
	CodeBatchInvalid      uint32 = 21 // 含 Envelope.seq ≠ Batch.seq
	CodeDecompressLimit   uint32 = 22
	CodeEnvelopeOpenFail  uint32 = 23
	CodeWindowExceeded    uint32 = 24 // 协议违规，配合 ABORT
	CodeStaleEpoch        uint32 = 25 // 预留（票据演进项）
	CodeLedgerDivergence  uint32 = 26 // 预留（连接级致命）
	CodeResultLevelBad    uint32 = 27 // RESULT level 非法
	CodeAckModeBad        uint32 = 28 // HelloAck.ack_mode 非法
	CodeCreditInvalid     uint32 = 29 // CREDIT 帧非法
	CodeHelloEphInvalid   uint32 = 30 // 临时公钥格式/长度非法
	CodeHelloProofInvalid uint32 = 31 // token_proof 验证失败（对外与 10 同面，仅日志区分）

	CodeServerDraining uint32 = 32
	CodeInternal       uint32 = 40 // 服务端 reaper 合成 released 时使用
	CodeUnavailable    uint32 = 41
	CodeProtocolError  uint32 = 42 // 通用协议违规（未知 flags/类型错域等）
)

错误码注册表（r3）：wire 上只传数字（Result.code / HelloNak.code / Abort.code）， append-only：只增不改不复用。分组段位：1x 握手 / 2x 数据 / 3x 会话 / 4x 服务端。

View Source 
const (
	FixedHeaderSz        = 2 // type(1) + flags(1)
	MaxFrameBytes uint64 = 16 << 20
)
View Source 
const (
	CapTraceContext = "trace_context"
	CapDatagram     = "datagram"
	CapCbor         = "cbor"
	// CapDualIntegrity：envelope 外层 HMAC（Envelope.mac = HMAC(profileHash,
	// DualMACKey, AD‖ciphertext)）。默认关闭——注册进 KnownCaps（客户端可合法
	// 宣告），但 DefaultServerPolicy.SupportedCapabilities 不含（服务端须以
	// WithPolicy 显式加入才协商启用）。
	CapDualIntegrity = "dual_integrity"
	// CapZstdDict：服务端静态 zstd 字典下发（DICT 帧 0x22，HELLO_ACK 后立即）。
	// 同样默认关闭：仅当 ServerPolicy 配置了 ZstdDict 时服务端才实际下发。
	CapZstdDict = "zstd_dict"
)

已注册 capability（append-only；x- 前缀 = experimental）。

View Source 
const (
	DefaultInitialCreditBatches = 256
	DefaultInitialCreditBytes   = 64 << 20 // 64MB
)

DefaultInitialCreditBatches / Bytes 是 HELLO_ACK 初始 credit 默认值 （服务端按 sink 吞吐调整；客户端必须 ≥ BDP/平均批大小）。

View Source 
const (
	SignalTypeLog       = "log"
	SignalTypeGauge     = "gauge"
	SignalTypeCounter   = "counter"
	SignalTypeHistogram = "histogram"
	SignalTypeSummary   = "summary"
	SignalTypeSpan      = "span"
	SignalTypeProfile   = "profile"
)

signal_type 取值（core spec §6.2 闭合枚举）。值发布后不可改（§1.4）。

View Source 
const (
	TemporalityDelta      = "delta"
	TemporalityCumulative = "cumulative"
)

temporality 取值（OTel AggregationTemporality 映射，core spec §6.2）。

View Source 
const (
	SpanKindUnspecified = ""
	SpanKindInternal    = "internal"
	SpanKindServer      = "server"
	SpanKindClient      = "client"
	SpanKindProducer    = "producer"
	SpanKindConsumer    = "consumer"
)

span kind 取值（OTel SpanKind 映射，core spec §6.2）。

View Source 
const (
	TypeHello    uint8 = 0x01 // C→S 握手（1-RTT：profile/token_proof/eph/caps）
	TypeHelloAck uint8 = 0x02 // S→C 握手应答（epoch/ack_mode/limits/credits/finished）
	TypeHelloNak uint8 = 0x03 // S→C 握手拒绝（code/retryable/retry_hint）
	TypeResult   uint8 = 0x20 // S→C 唯一反馈帧（accepted/durable/rejected/released）
	TypeCredit   uint8 = 0x21 // S→C 唯一流控发放帧
	TypeDict     uint8 = 0x22 // S→C zstd 字典下发（zstd_dict 能力协商后，HELLO_ACK 之后立即发送）
	TypePing     uint8 = 0x30 // 双向保活（仅 TCP binding）
	TypePong     uint8 = 0x31
	TypeClose    uint8 = 0x40 // 双向优雅关闭（drain 超时提示）
	TypeRedirect uint8 = 0x41 // S→C HA 重定向
	TypeAbort    uint8 = 0x50 // 双向立即拆除（写失败/协议违规/账本漂移）
	TypeDgram    uint8 = 0x90 // C→S 不可靠批次（Envelope；豁免 credit，硬限速）
	TypeBatch    uint8 = 0x91 // C→S 可靠批次（Envelope；消耗 credit）
)

帧类型注册表（r3）。type 高位为通道域：0x00-0x7F control / 0x80-0xFF data。 append-only：新类型只能追加；data 方向帧必须落在 data 域（conformance 断言）。

View Source 
const (
	StreamRoleControl = "control"
	StreamRoleData    = "data"
)

Stream roles（QUIC binding 的 stream 放置语义）。

View Source 
const DefaultDrainTimeout = 30 * time.Second

DefaultDrainTimeout 是 Close 排空的默认上限。

View Source 
const DefaultReadTimeout = 5 * time.Minute

DefaultReadTimeout 是无 ctx deadline 时 transport 读操作的默认上限。

View Source 
const DefaultReapInterval = 10 * time.Second

DefaultReapInterval 是服务端 inflight reaper 扫描间隔。

View Source 
const DefaultSessionTTL = 24 * time.Hour

DefaultSessionTTL 是会话默认有效期。

View Source 
const DefaultSinkDeadline = 60 * time.Second

DefaultSinkDeadline 是服务端 sink handler 的 done 调用超时（超时合成 released）。

View Source 
const DefaultWriteTimeout = 30 * time.Second

DefaultWriteTimeout 是无 ctx deadline 时 transport 写操作的默认上限。

View Source 
const MaxAgentIDLen = 256

MaxAgentIDLen 是 agent_id 字段的最大字节长度。agent_id 作为 ReplayCache key、 metrics label 来源、sink 调用参数及日志标识，须严格限制长度以防 DoS 与日志注入。

View Source 
const MaxDecompressedSize = 8 * 1024 * 1024

MaxDecompressedSize 限制解压后载荷上限，防压缩放大攻击。 与 limits.max_batch_bytes 对齐：解压后超过 batch 上限即为攻击或畸形。

View Source 
const MaxSignalAttrKeyLen = 256

MaxSignalAttrKeyLen 是 attribute key 的最大字节长度，抑制日志注入面。

View Source 
const MaxSignalAttrs = 128

MaxSignalAttrs 是单个 signal Record 的 attributes map 最大条目数， 防止畸形消息的内存放大 DoS（每个 entry 含 key/value + map 开销）。

View Source 
const MaxSignalFieldLen = 4096

MaxSignalFieldLen 是 SignalRecord 单个字符串字段的最大字节长度。取值保守 （远小于 maxEventBytes 256KB），用于在协议入口拒绝异常长输入，兼顾日志注入 防护与内存放大抑制。

View Source 
const MaxTraceStateFieldLen = 256

MaxTraceStateFieldLen 是 W3C tracestate 单个 key/value 的最大字符数（W3C Trace Context 规范）。tracestate 成员用于跨服务传播 trace 厂商信息，长度必须严格符合 规范以保证互操作；误用 MaxSignalFieldLen(4096) 校验会远超规范上限。

View Source 
const MaxZstdDictBytes = 1 << 20

MaxZstdDictBytes 限制服务端可配置的 zstd 字典尺寸（DICT 帧走 control 通道， 必须有界；静态训练字典典型 ~100KB，1MB 上限覆盖极端情况）。

Variables

View Source 
var (
	ErrCodeOK                 = ErrCode{CodeOK, "ok"}
	ErrCodeAuthFailed         = ErrCode{CodeAuthFailed, "auth_failed"}
	ErrCodeUnsupportedProfile = ErrCode{CodeUnsupportedProfile, "unsupported_profile"}
	ErrCodeUnsupportedCap     = ErrCode{CodeUnsupportedCapability, "unsupported_capability"}
	ErrCodeBadHandshake       = ErrCode{CodeBadHandshakeMessage, "bad_handshake_message"}
	ErrCodeBatchTooLarge      = ErrCode{CodeBatchTooLarge, "batch_too_large"}
	ErrCodeBatchInvalid       = ErrCode{CodeBatchInvalid, "batch_invalid"}
	ErrCodeDecompressLimit    = ErrCode{CodeDecompressLimit, "decompress_limit"}
	ErrCodeEnvelopeOpenFail   = ErrCode{CodeEnvelopeOpenFail, "envelope_open_failed"}
	ErrCodeWindowExceeded     = ErrCode{CodeWindowExceeded, "window_exceeded"}
	ErrCodeStaleEpoch         = ErrCode{CodeStaleEpoch, "stale_epoch"}
	ErrCodeLedgerDivergence   = ErrCode{CodeLedgerDivergence, "ledger_divergence"}
	ErrCodeResultLevelBad     = ErrCode{CodeResultLevelBad, "result_level_bad"}
	ErrCodeAckModeBad         = ErrCode{CodeAckModeBad, "ack_mode_bad"}
	ErrCodeCreditInvalid      = ErrCode{CodeCreditInvalid, "credit_invalid"}
	ErrCodeHelloEphInvalid    = ErrCode{CodeHelloEphInvalid, "hello_eph_invalid"}
	ErrCodeHelloProofInvalid  = ErrCode{CodeHelloProofInvalid, "hello_proof_invalid"}
	ErrCodeServerDraining     = ErrCode{CodeServerDraining, "server_draining"}
	ErrCodeInternal           = ErrCode{CodeInternal, "internal"}
	ErrCodeUnavailable        = ErrCode{CodeUnavailable, "unavailable"}
	ErrCodeProtocolError      = ErrCode{CodeProtocolError, "protocol_error"}
	ErrCodeUnknown            = ErrCode{^uint32(0), "unknown"}
)
View Source 
var (
	// LocalConfig: 配置无效（NewServer/NewClient 构造期 fail-fast）。
	LocalConfig = ErrCode{^uint32(0) - 1, "invalid_config"}
	// LocalSession/LocalBatch/LocalHandshake/LocalStream: 库内错误大类。
	LocalSession   = ErrCode{^uint32(0) - 2, "session_error"}
	LocalBatch     = ErrCode{^uint32(0) - 3, "batch_error"}
	LocalHandshake = ErrCode{^uint32(0) - 4, "handshake_error"}
	LocalTransport = ErrCode{^uint32(0) - 5, "transport_error"}
	LocalTLS       = ErrCode{^uint32(0) - 6, "tls_error"}
	LocalStream    = ErrCode{^uint32(0) - 7, "stream_error"}
	// LocalCredential: 凭证加载/校验失败（binding 层 TLS 配置常用）。
	LocalCredential = ErrCode{^uint32(0) - 8, "invalid_credential"}
)

本地错误码（不上 wire 的库内错误分类面；哨兵 Num 值域与 wire 注册表隔离）。 这些不是 wire 错误码的别名，是独立的本地分类——调用方按库错误来源分派处理。

View Source 
var (
	ErrCodeProtocol   = ErrCodeProtocolError
	ErrCodeStream     = LocalStream
	ErrCodeConfig     = LocalConfig
	ErrCodeSession    = LocalSession
	ErrCodeBatch      = LocalBatch
	ErrCodeHandshake  = LocalHandshake
	ErrCodeTransport  = LocalTransport
	ErrCodeTLS        = LocalTLS
	ErrCodeCredential = LocalCredential
)

旧名 → 新名迁移（破坏性重构收尾：别名最终删除，调用方全部改用 Local*）。

View Source 
var KnownCaps = map[string]bool{
	CapTraceContext:  true,
	CapDatagram:      true,
	CapCbor:          true,
	CapDualIntegrity: true,
	CapZstdDict:      true,
}

KnownCaps 是已注册 capability 集合。dual_integrity/zstd_dict 已注册（客户端可 宣告、服务端策略可加入 SupportedCapabilities），但默认关闭—— DefaultServerPolicy 不含二者（r3 裁决：按需显式开启）。

Functions

func AEADKeyLen 

func AEADKeyLen(p corepb.Profile) int

AEADKeyLen 返回 profile 的 AEAD 密钥长度。语义同 HMACKeyLen。

func BuildNonce added in v0.6.0 

func BuildNonce(prefix [4]byte, seq uint64) [12]byte

BuildNonce 构造确定性 nonce = prefix(4B) ‖ seq(8B BE)。全宽 seq：截断重用 （seq=k 与 k+2^64 同 nonce）在构造上不可表达。

func CanonicalHello added in v0.6.0 

func CanonicalHello(h *corepb.Hello) ([]byte, error)

CanonicalHello 返回 proof 置零后的 HELLO 规范编码。

func CanonicalHelloAck added in v0.6.0 

func CanonicalHelloAck(a *corepb.HelloAck) ([]byte, error)

CanonicalHelloAck 返回 finished 置零后的 HELLO_ACK 规范编码。

func CapIn added in v0.6.0 

func CapIn(caps []string, want string) bool

CapIn 报告能力列表是否含 cap。

func CodeName added in v0.6.0 

func CodeName(c uint32) string

CodeName 反查错误码名称（日志用；未注册返回 "unknown"）。

func CodeNumOf added in v0.6.0 

func CodeNumOf(err error) uint32

CodeNumOf 从 error 链提取数字错误码；非 *Error 返回 unknown 哨兵。

func CodeOf added in v0.6.0 

func CodeOf(err error) string

CodeOf 保留名称提取（内部用途）。非 *Error 返回空串。

func CodecFor added in v0.6.0 

func CodecFor(selected []string) corepb.Codec

CodecFor 从选定的能力集推导 SignalBatch 编码（cbor 能力被选中即用 cbor，否则 proto）。

func Decode 

func Decode(data []byte, m proto.Message) error

Decode proto 反序列化到 m。

func DefaultPolicyLimits added in v0.6.0 

func DefaultPolicyLimits() corepb.Limits

DefaultPolicyLimits 是帧/批次上限默认值。

func DeriveDualMACKey added in v0.6.0 

func DeriveDualMACKey(p corepb.Profile, zz, thHello, tokenProof []byte) ([]byte, error)

DeriveDualMACKey 派生 dual_integrity 的外层 HMAC 密钥（独立 Expand 段， 不改 split 布局）。

func DeriveZZ added in v0.6.0 

func DeriveZZ(p corepb.Profile, priv *EphemeralKey, peerWire []byte) ([]byte, error)

DeriveZZ 计算原始共享秘密（32B）。peerWire 严格校验长度/格式： intl 32B；gm 65B 非压缩点（0x04‖X‖Y，拒绝压缩点）。

func Encode 

func Encode(m proto.Message) ([]byte, error)

Encode proto 序列化消息。

func EphemeralPubLen added in v0.6.0 

func EphemeralPubLen(p corepb.Profile) int

EphemeralPubLen 返回 profile 临时公钥的 wire 长度（严格校验）。

func FilterUnknownCaps added in v0.6.0 

func FilterUnknownCaps(caps []string) []string

FilterUnknownCaps 返回不在注册表中的能力（客户端 NewClient fail-fast）。

func FinishedMAC added in v0.6.0 

func FinishedMAC(keys *SessionKeys, thFull []byte) []byte

FinishedMAC 计算 finished = HMAC(handshake_key, th_full)。

func HMACKeyLen 

func HMACKeyLen(p corepb.Profile) int

HMACKeyLen 返回 profile 的 HMAC 密钥长度。未知 profile 返回 -1 哨兵 （负长度强制调用方显式处理，避免静默空密钥）。

func IsControlType added in v0.6.0 

func IsControlType(t uint8) bool

IsControlType 报告 type 是否落在 control 域（type 高位 = 0）。 TCP binding 依赖此位做单连接逻辑流 demux；QUIC binding 以 stream 角色放置。

func IsDataType added in v0.6.0 

func IsDataType(t uint8) bool

IsDataType 报告 type 是否落在 data 域（type 高位 = 1）。

func MarshalSignalBatchCodec 

func MarshalSignalBatchCodec(codec corepb.Codec, sb *SignalBatch) ([]byte, error)

MarshalSignalBatchCodec 按 codec 分发编码 SignalBatch。 codec 未注册返回错误（不应发生在协商后的正常路径——协商保证双方都注册了该 codec）。

func NewAEAD 

func NewAEAD(p corepb.Profile, key []byte) (cipher.AEAD, error)

NewAEAD 按 profile 创建 AEAD（intl→AES-256-GCM / gm→SM4-GCM）。

func NewHMAC 

func NewHMAC(p corepb.Profile, key []byte) (hash.Hash, error)

NewHMAC 按 profile 创建 HMAC hasher（intl→SHA-256 / gm→SM3）。

func OpenEnvelope added in v0.6.0 

func OpenEnvelope(frameType uint8, sessionID []byte, keys *SessionKeys, dict *ZstdDictCodec, env *corepb.Envelope) ([]byte, error)

OpenEnvelope 解密 + 解压，返回解压后的 batch 编码。 frameType 必须与发送侧 Seal 时一致（AD 绑定），错配即 envelope_open_failed。

dual_integrity 语义（keys.DualMACKey 非空 = 本地协商启用）：

  • mac 非空但本地未启用（无 DualMACKey）→ envelope_open_failed；
  • mac 非空且校验失败 → envelope_open_failed；
  • 本地启用但 mac 为空（对端未按协商附加）→ envelope_open_failed（fail-closed， 双方由 selected_capabilities 对称驱动，缺 mac 即违约）。

func ProfileHash added in v0.6.0 

func ProfileHash(p corepb.Profile) func() hash.Hash

ProfileHash 返回 profile 的哈希构造器（intl=SHA-256 / gm=SM3）。

func ProfileName added in v0.6.0 

func ProfileName(p corepb.Profile) string

ProfileName 返回 profile 的注册名（transcript 与 AD 用字符串）。

func RegisterCodec 

func RegisterCodec(c SignalCodec)

RegisterCodec 注册一个 codec（覆盖同名）。线程安全，但建议在 init() 调用。

幂等：重复注册同一 Codec 值以最后一次为准。

func ResetZstdDicts added in v0.6.0 

func ResetZstdDicts()

ResetZstdDicts 清空注册表。测试隔离专用（生产静态字典部署不轮换）； 公共面保留是因为 conformance 为外部包无法用 export_test——doc 声明禁用于生产。

func SanitizeForLog 

func SanitizeForLog(s string) string

SanitizeForLog 清洗用于日志输出的网络来源字符串：截断超长值，把所有 C0 控制字符 （0x00-0x1F）和 DEL（0x7F）替换为空格。用于 slog 打印 reason 等不可信字段， 防御日志注入（换行伪造日志行、ANSI 转义终端注入、null 截断等）。

注意：\n \r 也被替换——防御换行注入。slog 自身已对结构化值做转义， 但当日志被转发到 syslog/journald/ELK 等外部系统时，中间层可能丢失转义。 SanitizeForLog 提供 defense-in-depth。

func SealEnvelope added in v0.6.0 

func SealEnvelope(p SealParams) ([]byte, error)

SealEnvelope 压缩 + 加密 batch 编码，返回 marshaled Envelope（即帧 payload）。 DualMACKey 非 nil 时追加 mac = HMAC(profileHash, DualMACKey, AD‖ciphertext)。

func TokenProof added in v0.6.0 

func TokenProof(p corepb.Profile, token string, thHello []byte) []byte

TokenProof 计算 HMAC(profileHash, key=token, msg=th_hello)。 该值既是 Hello.token_proof 字段，也是 psk_ck 的 ikm 输入（同值，无额外前缀）。

func TranscriptFull added in v0.6.0 

func TranscriptFull(p corepb.Profile, hello *corepb.Hello, ack *corepb.HelloAck) ([]byte, error)

TranscriptFull 计算 th_full = H(lp(profile) ‖ lp(canonical HELLO) ‖ lp(canonical ACK))。 HELLO_ACK 全字段入链：TLS 终止点对 limits/credits/ack_mode/epoch 的任何篡改 都会使 finished 校验失败。

func TranscriptHello added in v0.6.0 

func TranscriptHello(p corepb.Profile, hello *corepb.Hello) ([]byte, error)

TranscriptHello 计算 th_hello = H(lp(profile) ‖ lp(canonical(HELLO, proof 置零)))。

func ValidateBatchTraceContext added in v0.3.0 

func ValidateBatchTraceContext(tc *TraceContext) error

ValidateBatchTraceContext 校验 batch 级 W3C trace 上下文（spec §6.2.2，capability w3c_trace_context）。客户端发送前置与服务端解码后共用。

与 validateSignalTraceContext 的关键区别：batch 级 TraceContext 一旦显式提供， trace_id/span_id 必须非空——它是整批共享的继承点，没有「空=不参与」的退化语义 （不参与则不应携带 TraceContext）。parent_span_id 仍可选。

func ValidateHello 

func ValidateHello(m *corepb.Hello) error

ValidateHello 校验 HELLO。agent_id 必填且过长度/控制字符校验； meta 各字段可选但一旦提供须过同强度校验（继承 v1 加固）； eph 长度按 profile 严格校验；profile 必须已知。

func ValidateHelloAck 

func ValidateHelloAck(a *corepb.HelloAck) error

ValidateHelloAck 校验 HELLO_ACK（客户端侧）。 初始 credit 非正值 = 协议错误；ack_mode 零值拒绝；limits 非正值拒绝。

func ValidateHostPort added in v0.6.0 

func ValidateHostPort(addr string) error

ValidateHostPort 校验 host:port 格式的网络地址（v1 UDP / ntls TCP 共用）。 在 NewClient 构造期 fail-fast 调用，避免错误地址拖到 Connect 的 dial 层才报 一个低层 resolver 错误（错误归因不准）。接受 IPv6 字面量如 "[::1]:7400"。

func ValidateListenAddr added in v0.6.0 

func ValidateListenAddr(addr string) error

ValidateListenAddr 校验服务端监听地址（v1 QUIC/UDP 与 ntls TCP 共用）。

与 ValidateHostPort 的关键区别：监听地址允许 host 为空（":7400" = 全网卡， Go net.Listen/quic.Listen 合法）和 port=0（系统分配端口，e2e 测试常用）。 客户端 dial 地址则要求 host 非空 + port 1-65535，故二者用不同 helper。

仅校验结构性错误（缺端口、端口越界）；IP/host 合法性判断留给操作系统 listen （listen 是权威校验源，能区分格式错/地址不可 bind/权限不足/端口占用，库内重复 实现这套判断既不完整又可能与 OS 行为不一致）。在 NewServer 构造期 fail-fast 调用， 避免错误地址拖到 Start→Listen 才报低层错误。

func ValidateResultLevel added in v0.6.0 

func ValidateResultLevel(l corepb.ResultLevel) error

ValidateResultLevel 校验 RESULT level 合法（非零枚举值）。

func VerifyTokenProof added in v0.6.0 

func VerifyTokenProof(v TokenValidator, p corepb.Profile, agentID string, thHello, proof []byte) error

VerifyTokenProof 按 agentID 反查候选 token，逐个以 HMAC(profileHash, token, thHello) 与 proof 常量时间比对（hmac.Equal）。匹配返回 nil。

func Write 

func Write(w io.Writer, typ uint8, flags byte, payload []byte) error

Write 编码并写入单个帧。version/channel 域已删除（r3：ALPN 管版本、type 高位管通道）。

Types

type AbortMsg added in v0.6.0 

type AbortMsg = corepb.Abort

type AckMode 

type AckMode = corepb.AckMode

type AdmitDecision added in v0.6.0 

type AdmitDecision int

AdmitDecision 是 Admit 的判定结果。 

const (
	AdmitAccept       AdmitDecision = iota // 窗内未见：接受（已占位），返回 grant
	AdmitDuplicate                         // 在途占位中：幂等回执（查当前状态）
	AdmitTerminalHit                       // 终态环命中：回存储 outcome
	AdmitBelowUnknown                      // 环外旧 seq：released（宁重发不吞数据）
	AdmitViolation                         // seq==0：协议违规（ABORT）
)

type AdmitResult added in v0.6.0 

type AdmitResult struct {
	Decision AdmitDecision
	Grant    *SettleGrant // Decision==AdmitAccept 时非 nil
	// Duplicate/TerminalHit/BelowUnknown 的回执数据（handler 直接回 RESULT）。
	DupSlot  InflightSlot
	Terminal TerminalEntry
}

AdmitResult 是 Admit 的复合返回：判定 + grant（Accept 时非 nil）/回执数据。

type Batch added in v0.6.0 

type Batch = corepb.Batch

type BatchEvent added in v0.6.0 

type BatchEvent struct {
	Agent   string
	Meta    BatchMeta
	Decoded *SignalBatch
	Raw     []byte
}

BatchEvent 是投递给 handler 的一个 batch。 Raw/Decoded 按 SinkOptions 恰一供给（另一个为 nil）。

type BatchHandler added in v0.6.0 

type BatchHandler func(ctx context.Context, ev BatchEvent, done func(Outcome))

BatchHandler 是注册的数据投递回调。

done 契约（规范性 MUST，docs/v2-design.md §11.1）：

  • 幂等：首调生效，后续调用安全丢弃；
  • 单调：accepted 已发出后合法收缩为 Durable|Released；
  • 超时：SinkDeadline 内未调 → 协议层合成 Outcome{Released, internal}；
  • panic：per-invocation recover → 合成 released + 计数；
  • 迟调：连接已关闭后安全丢弃。

type BatchMeta added in v0.6.0 

type BatchMeta struct {
	Seq         uint64
	Correlation string
	EventsCount int
	ReceivedAt  time.Time
	Trace       *TraceContext // 可选（trace_context 能力）
}

BatchMeta 是 batch 的投递元信息。

type ChunkID 

type ChunkID [16]byte

ChunkID 是全局唯一批次标识（ULID 16 字节，core spec §5.2 / §11.2）。

全局唯一 + 时序（毫秒精度），用于去重、重试、抗重放。 wire 上传 raw 16 字节；map key / spool 文件名用文本编码（Crockford base32，26 字符）， 以最小化下游（文件系统）改动。

[16]byte 可比、可作 map key，满足 ReplayCache / spool 的键需求。

func ChunkIDFromBytes 

func ChunkIDFromBytes(b []byte) (ChunkID, error)

ChunkIDFromBytes 从 16 字节构造 ChunkID。

func NewChunkID 

func NewChunkID() ChunkID

NewChunkID 生成一个新的 ULID ChunkID（并发安全）。 MustNew panic（熵源失败/单调溢出）时降级为随机 ChunkID，保证进程不崩溃。

func (ChunkID) Bytes 

func (c ChunkID) Bytes() []byte

Bytes 返回 raw 16 字节切片（wire 传输用，core spec §5.2 chunk_id bytes 字段）。

func (ChunkID) IsZero 

func (c ChunkID) IsZero() bool

IsZero 报告 ChunkID 是否未设置（全零）。

func (ChunkID) String 

func (c ChunkID) String() string

String 返回 Crockford base32 文本（26 字符），用作 map key / spool 文件名。

type CloseMsg added in v0.6.0 

type CloseMsg = corepb.Close

type Compression added in v0.6.0 

type Compression = corepb.Compression

type CorrelationDeduper added in v0.6.0 

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

CorrelationDeduper 是进程级 LRU 关联去重器（并发安全）。

func NewCorrelationDeduper added in v0.6.0 

func NewCorrelationDeduper(capacity int, ttl time.Duration) *CorrelationDeduper

NewCorrelationDeduper 创建去重器。capacity ≤0 取 65536；ttl ≤0 表示不过期 （仅容量淘汰）。

func (*CorrelationDeduper) Dedup added in v0.6.0 

func (d *CorrelationDeduper) Dedup(key string) bool

Dedup 报告 key 是否首次出现：首次返回 true（放行）并登记；重复或已过期被 淘汰后重新出现也返回 true。空 key 恒 true（不去重语义）。

func (*CorrelationDeduper) Len added in v0.6.0 

func (d *CorrelationDeduper) Len() int

Len 返回当前登记数。

func (*CorrelationDeduper) Seen added in v0.6.0 

func (d *CorrelationDeduper) Seen(key string) bool

Seen 报告 key 是否在窗口内（不登记）。

type Counter 

type Counter interface {
	Inc()
	Add(float64)
}

Counter 是单调递增计数器的最小接口（对应 prometheus.Counter 语义）。 抽象出此接口使 protocol 层不直接依赖 prometheus 具体类型，支持注入 NoOp 实现（无 metrics 场景）或未来其他后端（如 OpenTelemetry）。

Add 必须传入非负值：prometheus.Counter.Add 对负值 panic（进程崩溃）。 若调用方无法保证入参非负，建议使用 NewGuardedCounter 包装。

type Credit added in v0.6.0 

type Credit = corepb.Credit

type CreditLedger added in v0.6.0 

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

CreditLedger 是客户端 credit 账本。

available = granted − consumed + returned（逐维度）。 consumed 在发送时递增；returned 只在收到 CREDIT 时递增（模型 B 核心： RESULT 到达不释放额度，额度返还的权威在服务端）；Rollback 在写失败时 补偿 returned（服务端从未见帧，不会为该 seq 返还——不补偿即单向泄漏）。

func NewCreditLedger added in v0.6.0 

func NewCreditLedger(initialB, initialY int64, clampB, clampY int64) *CreditLedger

NewCreditLedger 以 HELLO_ACK 初始 credit 建账。clamp ≤0 = 默认钳制。

func (*CreditLedger) Available added in v0.6.0 

func (l *CreditLedger) Available() (batches, bytes int64)

Available 返回当前可用额度。

func (*CreditLedger) Consume added in v0.6.0 

func (l *CreditLedger) Consume(seq uint64, bytes int64, correlation string, deadline time.Time) error

Consume 消耗额度并登记 slot。额度不足返回错误（不改变状态）。

func (*CreditLedger) Drained added in v0.6.0 

func (l *CreditLedger) Drained() bool

Drained 报告全部已发 seq 是否移出 pending（drain 判定的客户端侧）。

func (*CreditLedger) OnCredit added in v0.6.0 

func (l *CreditLedger) OnCredit(dB, dY int64)

OnCredit 处理 CREDIT 帧增量：双零 delta 忽略；负值（uint64 回绕）违规丢弃； 超出已消耗未返还量的部分钳制并计违规。

func (*CreditLedger) Pending added in v0.6.0 

func (l *CreditLedger) Pending() int

Pending 返回在途 slot 数。

func (*CreditLedger) ReapExpired added in v0.6.0 

func (l *CreditLedger) ReapExpired(now time.Time) []*OutSlot

ReapExpired 回收 deadline 已过的 slot（标记失败并移出 pending；不回补 credit）。

func (*CreditLedger) Reset added in v0.6.0 

func (l *CreditLedger) Reset()

Reset 清空全部状态（close 时）。

func (*CreditLedger) Rollback added in v0.6.0 

func (l *CreditLedger) Rollback(seq uint64) *OutSlot

Rollback 写失败路径专用：移出 slot 并把额度补偿回 returned。 服务端从未收到该帧、不会为它返还 CREDIT——不补偿即单向泄漏。 若服务端实际已收到帧并随后 CREDIT，OnCredit 的 maxRet 钳制会吸收多补部分。

func (*CreditLedger) Terminal added in v0.6.0 

func (l *CreditLedger) Terminal(seq uint64) *OutSlot

Terminal 处理某 seq 的 RESULT：移出 pending slot（不触碰 credit——返还权威在 服务端 CREDIT）。返回 slot 供回调；未知 seq（已 reap/rollback）返回 nil。

type CreditManager added in v0.6.0 

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

CreditManager 是服务端准入/去重/credit 守恒的单一真相（单锁）。

去重语义（乱序安全）：

  • low = 连续已接受前缀 + 1：[1, low) 全部接受过（在途或已终态）；
  • accepted：已接受（含已终态）未被前缀覆盖的 seq——乱序终态（3 先于 1、2） 停留于此，重放查终态环（不依赖前缀推进这一时序副产物）；
  • seen：在途（Admit 占位起，Settle/Release 止）——占位使「至多一个 goroutine 拿到 Accept」成为构造保证（消除 Admit→Register 解密间隙的并发双接受）；
  • terminal：最近 K 条终态（LRU 淘汰）。

func NewCreditManager added in v0.6.0 

func NewCreditManager(initialB, initialY int64, ringCap int) *CreditManager

NewCreditManager 创建管理器（initial 为 HELLO_ACK 初始 credit）。ringCap ≤0 取 256。

func (*CreditManager) Admit added in v0.6.0 

func (m *CreditManager) Admit(seq uint64) AdmitResult

Admit 判定 seq 准入。Accept 即占位（并发同 seq 的后来者得 Duplicate）。 无 cap 判定：超发防线是 Consume 的额度检查（seq 空间含 DGRAM， 按 batches 计的 cap 会误杀混发连接——已在审查中确认为缺陷并移除）。

func (*CreditManager) ForceSettle added in v0.6.0 

func (m *CreditManager) ForceSettle(send func(*corepb.Result), seq uint64, level corepb.ResultLevel, code uint32, retryMs uint32, frameBytes int64)

ForceSettle 对仍在途（占位）的 seq 做恢复性终态（服务端 reaper 与 Handle 收尾）：grant 对象可能已随 panic 丢失，此处绕过 grant 直接处置。与 grant.Settle 共享防双回收守卫（terminal 环命中即跳过额度，仅重发 RESULT）。

func (*CreditManager) OldestBefore added in v0.6.0 

func (m *CreditManager) OldestBefore(cutoff time.Time) (uint64, InflightSlot, bool)

OldestBefore 返回 AddedAt 早于 cutoff 的最旧在途条目（服务端 reaper 用）。

func (*CreditManager) Pending added in v0.6.0 

func (m *CreditManager) Pending() int

Pending 返回在途占位数。

func (*CreditManager) String added in v0.6.0 

func (m *CreditManager) String() string

String 状态快照（日志）。

func (*CreditManager) TakeRecovered added in v0.6.0 

func (m *CreditManager) TakeRecovered() (batches, bytes int64)

TakeRecovered 取出攒批缓冲的回收量（发放 CREDIT 前调用；移入 outstanding）。 无回收返回 (0,0)。

type Credits added in v0.6.0 

type Credits = corepb.Credits

type DeadlineReader added in v0.6.0 

type DeadlineReader interface {
	io.Reader
	SetReadDeadline(t time.Time) error
}

DeadlineReader 是支持读超时的 io.Reader（*quic.Stream 和 net.Conn 均满足）。

type Envelope added in v0.6.0 

type Envelope = corepb.Envelope

type EphemeralKey added in v0.6.0 

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

EphemeralKey 是连接级临时密钥对（具体曲线由 profile 决定，调用方不感知库差异）。

func GenerateEphemeral added in v0.6.0 

func GenerateEphemeral(p corepb.Profile) (*EphemeralKey, []byte, error)

GenerateEphemeral 生成 profile 对应的临时密钥对。

func (*EphemeralKey) PubBytes added in v0.6.0 

func (k *EphemeralKey) PubBytes() ([]byte, error)

PubBytes 返回本方临时公钥的 wire 字节。

type ErrCode added in v0.6.0 

type ErrCode struct {
	Num  uint32
	Name string
}

ErrCode 是标准错误的错误码（Num=注册表数字；Name 仅供日志，不上 wire）。

type Error 

type Error struct {
	Code    ErrCode
	Message string
	Detail  any
	Err     error
}

Error 是 MBTA 库的标准错误类型。

func NewError 

func NewError(c ErrCode, msg string) *Error

NewError / WrapError 构造标准错误。

func WrapError 

func WrapError(c ErrCode, msg string, err error) *Error

func (*Error) Error 

func (e *Error) Error() string

func (*Error) Unwrap 

func (e *Error) Unwrap() error

type ExponentialHistogram added in v0.2.0 

type ExponentialHistogram = corepb.ExponentialHistogram

ExponentialHistogram 是 histogram signal 的 exponential bucket 表达（core spec §6.2）。 等价 corepb.ExponentialHistogram。

type Frame 

type Frame struct {
	Header  Header
	Payload []byte
}

func Read 

func Read(r io.Reader, lim FrameLimits) (Frame, error)

Read 读取并校验单个帧。

接收端 MUST（继承 v1 加固资产）：

  • 长度上限校验先于分配（防声明式 varint 长度直接 OOM）；
  • varint 非最短编码拒绝；
  • payload 短读时按实际 n drain 剩余字节维持帧边界。

func ReadFrameCtx added in v0.6.0 

func ReadFrameCtx(ctx context.Context, r DeadlineReader) (Frame, error)

ReadFrameCtx 是 ctx 感知的帧读取：watcher 把 ctx.Done 翻译成已到期 deadline 立即唤醒阻塞读（join 保证 watcher 退出后才恢复 deadline，避免竞态污染下一次读）。

func ReadFrameCtxLimit added in v0.6.0 

func ReadFrameCtxLimit(ctx context.Context, r DeadlineReader, lim FrameLimits) (Frame, error)

ReadFrameCtxLimit 是带帧上限的 ReadFrameCtx：协商/策略的 max_frame_bytes 经此接入客户端控制流读取路径（服务端 data 循环直接用 core.Read + FrameLimits）。

type FrameLimits added in v0.6.0 

type FrameLimits struct {
	MaxFrameBytes uint64
}

func DefaultLimits 

func DefaultLimits() FrameLimits

DefaultLimits 返回接收端默认上限。

type Gauge 

type Gauge interface {
	Set(float64)
	Inc()
	Dec()
	Add(float64)
}

Gauge 是可增可减的瞬时值接口（对应 prometheus.Gauge 语义）。 

type Header struct {
	Flags  byte
	Type   uint8
	Length uint64
}

Header 表示帧头。帧格式：[type(1)][flags(1)][varint Length][Payload]。 无帧层 CRC——完整性由传输层 AEAD（TLS 1.3 / TLCP）+ 应用层 envelope AEAD 承担。

type Hello added in v0.6.0 

type Hello = corepb.Hello

type HelloAck added in v0.6.0 

type HelloAck = corepb.HelloAck

type HelloNak added in v0.6.0 

type HelloNak = corepb.HelloNak

type Histogram 

type Histogram interface {
	Observe(float64)
}

Histogram 是观测值分布接口（对应 prometheus.Histogram 语义）。

type InflightSlot added in v0.6.0 

type InflightSlot struct {
	Seq         uint64
	Bytes       int64
	Correlation string
	AddedAt     time.Time
}

InflightSlot 是在途 batch 状态（占位/回填后均存在，Settle 时移除）。

type Limits 

type Limits = corepb.Limits

type MBTAMetrics 

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

MBTAMetrics 是 Metrics 的 prometheus 后端实现。

所有字段为非导出（小写）：外部必须通过 Metrics 接口的 getter 方法访问， Counter 类方法返回 guardedCounter 包装，防止负值 Add 触发 prometheus panic。 New(reg) 返回的 *MBTAMetrics 可直接赋值给 Metrics 类型的配置字段。

func New 

func New(reg prometheus.Registerer) *MBTAMetrics

New creates and registers all MBTA metrics with the given registerer. Pass prometheus.DefaultRegisterer for global registration, or a *prometheus.Registry for isolated registration (tests, multi-tenant).

func (*MBTAMetrics) AuthFailure 

func (m *MBTAMetrics) AuthFailure() Counter

func (*MBTAMetrics) AuthSuccess 

func (m *MBTAMetrics) AuthSuccess() Counter

func (*MBTAMetrics) BatchLatency 

func (m *MBTAMetrics) BatchLatency() Histogram

func (*MBTAMetrics) BatchSizeBytes 

func (m *MBTAMetrics) BatchSizeBytes() Histogram

func (*MBTAMetrics) BatchSizeEvents 

func (m *MBTAMetrics) BatchSizeEvents() Histogram

func (*MBTAMetrics) BatchesSent 

func (m *MBTAMetrics) BatchesSent() Counter

func (*MBTAMetrics) ConnectionDuration 

func (m *MBTAMetrics) ConnectionDuration() Histogram

func (*MBTAMetrics) ConnectionsActive 

func (m *MBTAMetrics) ConnectionsActive() Gauge

func (*MBTAMetrics) CreditsOutstanding added in v0.6.0 

func (m *MBTAMetrics) CreditsOutstanding() Gauge

func (*MBTAMetrics) DecryptFailures 

func (m *MBTAMetrics) DecryptFailures() Counter

func (*MBTAMetrics) HMACFailures 

func (m *MBTAMetrics) HMACFailures() Counter

func (*MBTAMetrics) SinkFailures added in v0.6.0 

func (m *MBTAMetrics) SinkFailures() Counter

func (*MBTAMetrics) SinkQueueDepth added in v0.6.0 

func (m *MBTAMetrics) SinkQueueDepth() Gauge

type Metrics 

type Metrics interface {
	// 连接与认证
	ConnectionsActive() Gauge
	AuthSuccess() Counter
	AuthFailure() Counter

	// 批次投递
	BatchesSent() Counter

	// 安全失败
	HMACFailures() Counter
	DecryptFailures() Counter

	// sink 投递失败（BatchHandler 返回错误或 panic 时递增）。
	// ACK_MODE_ACCEPTED 已发但 sink 持久化失败=潜在数据丢失。
	SinkFailures() Counter

	// 流控窗口
	CreditsOutstanding() Gauge
	SinkQueueDepth() Gauge

	// 关键 SLI 直方图
	BatchLatency() Histogram    // SendBatch → ACK latency
	BatchSizeEvents() Histogram // events per batch distribution
	BatchSizeBytes() Histogram  // bytes per batch distribution
	ConnectionDuration() Histogram
}

Metrics 是 MBTA 协议层的可观测性抽象。

protocol 层通过此接口记录指标，与具体后端解耦：

  • 默认实现 MBTAMetrics（prometheus 后端，见 New）；
  • NoOpMetrics 用于不需要指标的场景（测试、轻量嵌入）；
  • 未来可添加 OpenTelemetry 等后端实现。

方法式 API（如 AuthFailure() 而非字段 AuthFailureTotal）让接口定义自包含， 且各实现可惰性构造返回值。HandlerConfig.Metrics 为 nil 时 handler 视作 NoOp。

type NoOpMetrics 

type NoOpMetrics struct{}

NoOpMetrics 是 Metrics 的空实现，所有方法返回 no-op 指标（零开销）。 用于不需要指标的场景（测试、轻量嵌入），或作为 HandlerConfig.Metrics 为 nil 时的回退。

func (NoOpMetrics) AuthFailure 

func (NoOpMetrics) AuthFailure() Counter

func (NoOpMetrics) AuthSuccess 

func (NoOpMetrics) AuthSuccess() Counter

func (NoOpMetrics) BatchLatency 

func (NoOpMetrics) BatchLatency() Histogram

func (NoOpMetrics) BatchSizeBytes 

func (NoOpMetrics) BatchSizeBytes() Histogram

func (NoOpMetrics) BatchSizeEvents 

func (NoOpMetrics) BatchSizeEvents() Histogram

func (NoOpMetrics) BatchesSent 

func (NoOpMetrics) BatchesSent() Counter

func (NoOpMetrics) ConnectionDuration 

func (NoOpMetrics) ConnectionDuration() Histogram

func (NoOpMetrics) ConnectionsActive 

func (NoOpMetrics) ConnectionsActive() Gauge

func (NoOpMetrics) CreditsOutstanding added in v0.6.0 

func (NoOpMetrics) CreditsOutstanding() Gauge

func (NoOpMetrics) DecryptFailures 

func (NoOpMetrics) DecryptFailures() Counter

func (NoOpMetrics) HMACFailures 

func (NoOpMetrics) HMACFailures() Counter

func (NoOpMetrics) ResultsByLevel added in v0.6.0 

func (NoOpMetrics) ResultsByLevel(string) Counter

func (NoOpMetrics) SinkFailures added in v0.6.0 

func (NoOpMetrics) SinkFailures() Counter

func (NoOpMetrics) SinkQueueDepth added in v0.6.0 

func (NoOpMetrics) SinkQueueDepth() Gauge

type OnSession added in v0.6.0 

type OnSession func(ev SessionEvent)

OnSession 是可选的会话生命周期回调。nil 表示不观测连接生命周期。

调用时序约定：

  • CONNECTED：在 HELLO_ACK 发送、状态机进入 READY 之后触发（早于 RedirectChecker 判定， 也早于任何 BATCH 处理）。被重定向（follower→leader）的连接也会先触发 CONNECTED 再 触发 DISCONNECTED，保证 CONNECTED/DISCONNECTED 严格成对。
  • DISCONNECTED：在连接结束（正常 CLOSE、传输错误、ctx 取消）时触发，仅当该连接此前 已认证成功（agentID 非空）。

实现必须是线程安全的、非阻塞的：回调在连接处理 goroutine 内同步调用，阻塞会拖死该连接。 重型持久化/网络操作应异步派发。

type OutSlot added in v0.6.0 

type OutSlot struct {
	Seq         uint64
	Bytes       int64
	Correlation string
	SentAt      time.Time
	Deadline    time.Time
	// Failed 由客户端 reaper 标记（RESULT 迟到/丢失）；不回补 credit，
	// 由服务端 reaper 保证最终 RESULT(released)+CREDIT。
	Failed bool
}

OutSlot 记录一个在途 batch（RESULT 关联/回调/超时观测用）。

type Outcome added in v0.6.0 

type Outcome struct {
	Level      OutcomeLevel
	Code       uint32        // Rejected 时必填（错误码注册表）
	RetryAfter time.Duration // Released 时可选
}

Outcome 是 handler 通过 done 报告的投递结果。

type OutcomeLevel added in v0.6.0 

type OutcomeLevel int8

OutcomeLevel 是投递结果级别（wire 语义见 ResultLevel）。 

const (
	OutcomeAccepted OutcomeLevel = iota
	OutcomeDurable
	OutcomeRejected
	OutcomeReleased
)

type PingMsg added in v0.6.0 

type PingMsg = corepb.Ping

type PongMsg added in v0.6.0 

type PongMsg = corepb.Pong

type Profile added in v0.6.0 

type Profile = corepb.Profile

type ProfilePayload added in v0.2.0 

type ProfilePayload = corepb.ProfilePayload

ProfilePayload 是 profile signal 载荷 + 跨信号双向关联（core spec §6.2）。 等价 corepb.ProfilePayload。

type Redirect added in v0.6.0 

type Redirect = corepb.Redirect

type RedirectChecker added in v0.2.0 

type RedirectChecker func(ctx context.Context) (RedirectInfo, bool)

RedirectChecker 在握手完成后调用；返回 ok=true 时服务端发送 REDIRECT 帧并 关闭连接，把客户端导向 leader。nil = 不参与 HA。

type RedirectInfo added in v0.2.0 

type RedirectInfo struct {
	LeaderAddr string
	LeaderID   string
}

RedirectInfo 携带 leader 位置（HA 重定向）。

type Resource 

type Resource struct {
	Attributes map[string]any `json:"attributes,omitempty"`
}

Resource 产生信号的实体属性。

type Result added in v0.6.0 

type Result = corepb.Result

type ResultLevel added in v0.6.0 

type ResultLevel = corepb.ResultLevel

type Scope 

type Scope struct {
	Name        string `json:"name,omitempty"`
	Version     string `json:"version,omitempty"`
	CollectorID string `json:"collector_id,omitempty"`
}

Scope 采集器或插件信息。

type SealParams added in v0.6.0 

type SealParams struct {
	FrameType   uint8  // 帧类型（TypeBatch/TypeDgram，入 AD）
	SessionID   []byte // 会话 ID（入 AD）
	Seq         uint64 // 连接级序号（nonce 与 AD 双绑定）
	KeyEpoch    uint32 // 首版恒 0
	Compression corepb.Compression
	Keys        *SessionKeys
	Dict        *ZstdDictCodec // ZSTD_DICT 时的字典 codec（nil=fail-closed）
	BatchWire   []byte         // Batch 消息编码（未压缩）
	// DualMACKey 是 dual_integrity 能力协商启用时的外层 HMAC 密钥，通常传
	// Keys.DualMACKey。nil = 跳过，Envelope.mac 置空。
	DualMACKey []byte
}

SealParams 是 Envelope 构建参数。

type SendConfig added in v0.3.0 

type SendConfig struct {
	// TraceContext 是 batch 级 W3C trace 上下文（capability w3c_trace_context，
	// spec §6.2.2）。非 nil 时由 CoreClient 设到 BatchMessage.TraceContext（field 7），
	// 不进入 SignalBatch 的 marshal 字节。nil 表示该 batch 不携带 batch 级 trace
	// （混装多 trace 场景应保持 nil，改用 SignalRecord 的 per-signal trace 字段）。
	TraceContext *TraceContext
}

SendConfig 承载 per-call 发送选项，是变参 SendOption 的聚合目标。 仅含可选字段，零值即「不启用任何选项」，与不传 opts 的旧行为完全一致。

func ApplySendOptions added in v0.3.0 

func ApplySendOptions(opts []SendOption) SendConfig

ApplySendOptions 聚合变参为 SendConfig，供各发送层统一解析。 显式导出以便 internal/protocol 层跨包调用。

type SendOption added in v0.3.0 

type SendOption func(*SendConfig)

SendOption 是发送类 API 的 per-call 选项，参照 ClientOption 的函数式选项范式。 变参形态保证旧调用（不传 opts）零改动兼容。

func WithTraceContext added in v0.3.0 

func WithTraceContext(tc *TraceContext) SendOption

WithTraceContext 携带 batch 级 W3C trace 上下文（整批共享一个 trace 的优化承载）。

适用前提：整个 batch 属于同一 trace（如 trace 聚合器收集单 trace 的全部 span）。 若 batch 混装多个不同 trace_id 的 signal（FIFO 攒批的常见形态），不应使用本选项—— 应改用每个 SignalRecord 的 per-signal trace 字段（TraceID/SpanID/ParentSpanID/ TraceFlags/TraceState），下游按 trace_id 重组。batch 级 TraceContext 与 per-signal trace 是协同设计：前者是整批基线，后者是偏离覆盖。

需对端握手协商 w3c_trace_context，否则发送端在门控阶段显式报错（不静默丢弃）。

type SeqGenerator 

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

SeqGenerator 生成连接级单调 seq（BATCH/DGRAM 共享单一计数器——分设会跨通道 复用 nonce，GCM 灾难）。从 1 起；0 为非法哨兵。

func NewSeqGenerator 

func NewSeqGenerator() *SeqGenerator

func (*SeqGenerator) Next 

func (g *SeqGenerator) Next() uint64

type ServerMachine 

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

ServerMachine 是线程安全的服务端状态机。

func NewServerMachine 

func NewServerMachine() *ServerMachine

func (*ServerMachine) State 

func (sm *ServerMachine) State() ServerState

func (*ServerMachine) Transition 

func (sm *ServerMachine) Transition(next ServerState) error

type ServerPolicy added in v0.6.0 

type ServerPolicy struct {
	Profile               corepb.Profile
	SupportedCapabilities []string
	AckMode               corepb.AckMode
	InitialCreditBatches  uint64
	InitialCreditBytes    uint64
	MaxFrameBytes         uint64
	MaxBatchBytes         uint64
	// DatagramRateBytes 是 DGRAM 令牌桶速率（bytes/s）。0 = 不限（仅并发槽防护）。
	DatagramRateBytes int64
	// ZstdDict/ZstdDictID 是 zstd_dict 能力的服务端静态字典（可选）。
	// 配置后：客户端协商 zstd_dict 时 HELLO_ACK 后立即下发 DICT 帧，且服务端
	// 解压 COMPRESSION_ZSTD_DICT envelope 时使用该字典。未配置时服务端不会
	// 选中 zstd_dict（即使 SupportedCapabilities 含它）。
	ZstdDict   []byte
	ZstdDictID uint32
}

ServerPolicy 是服务端策略声明。

func DefaultServerPolicy added in v0.6.0 

func DefaultServerPolicy(p corepb.Profile) ServerPolicy

DefaultServerPolicy 返回 profile 对应的默认策略。

func (ServerPolicy) IsZero added in v0.6.0 

func (p ServerPolicy) IsZero() bool

IsZero 报告策略是否为全零值（调用方据此回退 DefaultServerPolicy）。 含 slice 字段故不可用 == 比较，需逐字段判定。

func (ServerPolicy) Negotiate added in v0.6.0 

func (p ServerPolicy) Negotiate(clientCaps []string) []string

Negotiate 计算能力交集（保序：按服务端策略顺序）。

func (ServerPolicy) Validate added in v0.6.0 

func (p ServerPolicy) Validate() error

Validate 校验策略完备性（NewServer fail-fast）。

type ServerState 

type ServerState int
const (
	// ServerStateHandshaking 等待/处理 HELLO。
	ServerStateHandshaking ServerState = iota
	// ServerStateReady 已完成握手。
	ServerStateReady
	// ServerStateDraining 排空中。
	ServerStateDraining
)

func (ServerState) String 

func (s ServerState) String() string

type SessionEvent added in v0.6.0 

type SessionEvent struct {
	Type       SessionEventType
	AgentID    string
	SessionID  string
	RemoteAddr net.Addr
}

SessionEvent 在 agent 连接生命周期关键节点（认证成功、连接断开）投递给 OnSession 回调。

与 RedirectChecker（仅返回 leader 地址）不同，SessionEvent 携带连接身份： agentID/sessionID 来自握手状态机，RemoteAddr 来自传输层。这让上层（如 forwarder 的 agentregistry）能在 connect 时刻登记一个 agent，而不必等到首条 BATCH 才 lazy 注册。应用层扩展元数据（hostname/os/version 等）经 HELLO.meta 自由携带，不在本结构重复。

type SessionEventType added in v0.6.0 

type SessionEventType int

SessionEventType 标识会话生命周期事件的类型。 

const (
	// SessionConnected 表示一个 agent 完成认证（HELLO_ACK 发送、进入 READY）。
	SessionConnected SessionEventType = iota
	// SessionDisconnected 表示一个已认证 agent 的连接结束（正常关闭或异常断开）。
	// 仅对认证成功过的连接触发；未通过 HELLO 的连接不会产生 DISCONNECTED 事件。
	SessionDisconnected
)

type SessionKeys 

type SessionKeys struct {
	Profile      corepb.Profile
	C2SKey       []byte // 数据加密（c→s envelope AEAD）
	S2CKey       []byte // 预留（首版无 s→c 数据加密帧）
	HandshakeKey []byte // finished MAC 专用（密钥用途分隔）
	NoncePrefixC [4]byte
	NoncePrefixS [4]byte
	// DualMACKey 是 dual_integrity 能力协商启用时追加派生的外层 HMAC 密钥，
	// 即 HMAC(profileHash, DualMACKey, AD‖ciphertext)，见 envelope.go。
	// nil = 未启用（该能力默认关闭，SelectedCapabilities 不含即不派生）。
	// 由协商驱动：握手双方都在 selected_capabilities 含 dual_integrity 时
	// 调 DeriveDualMACKey 填充——仅单侧派生会使 Envelope mac 字段判定不对称。
	DualMACKey []byte
}

SessionKeys 是派生出的会话密钥组（密钥永不上 wire）。

func DeriveSessionKeys added in v0.6.0 

func DeriveSessionKeys(p corepb.Profile, zz, thHello, tokenProof []byte) (*SessionKeys, error)

DeriveSessionKeys 执行完整密钥日程（两侧调用相同函数）。

func DeriveSessionKeysWithCaps added in v0.6.0 

func DeriveSessionKeysWithCaps(p corepb.Profile, zz, thHello, tokenProof []byte, selectedCaps []string) (*SessionKeys, error)

DeriveSessionKeysWithCaps 是密钥日程的完整单点（双端共用）： 基础 DeriveSessionKeys + selected 含 dual_integrity 时追加 DualMACKey。 消除 client/handler 两侧逐行复制的漂移面（一侧能力名/info 串笔误 → mac 判定不对称、全部 envelope 打开失败）。

type SettleGrant added in v0.6.0 

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

SettleGrant 是一次准入的独占处置权。

生命周期：Admit 占位 → Consume 扣额度（或直拒路径跳过）→ Backfill 回填元数据 → Accept（可选中间态）→ Settle 终态 / Release 异常释放。 Accept 与 Settle 各自单次生效；Settle 后一切调用为 no-op——RESULT 双发与 额度漏偿在构造上不可表达。

func (*SettleGrant) Accept added in v0.6.0 

func (g *SettleGrant) Accept(send func(*corepb.Result))

Accept 发送中间态 accepted RESULT（durable 模式）；单次生效。 不触碰终态/额度（durable 才是终态）。

func (*SettleGrant) Backfill added in v0.6.0 

func (g *SettleGrant) Backfill(correlation string)

Backfill 回填明文侧元数据（correlation）——解密后调用。

func (*SettleGrant) Consume added in v0.6.0 

func (g *SettleGrant) Consume(frameBytes int64) error

Consume 扣减额度并登记帧字节数。额度不足返回错误并自动释放占位 （调用方应 ABORT：客户端在无额度下发送属协议违规）。

func (*SettleGrant) Release added in v0.6.0 

func (g *SettleGrant) Release()

Release 无 wire 动作的占位释放（Consume 前的异常路径：解码失败且无法定夺 seq 的场景。生产主路径的解码失败发生在 Consume 后，走 Settle(REJECTED)）。

func (*SettleGrant) Settle added in v0.6.0 

func (g *SettleGrant) Settle(send func(*corepb.Result), level corepb.ResultLevel, code uint32, retryMs uint32, frameBytes int64)

Settle 终态处置：发送 RESULT + 终态环记录 + 额度回收。单次生效（幂等）。 未 Consume 的 grant（sem-full 直拒等路径）也允许 Settle——按 frameBytes 补回收， 保证客户端侧已扣额度最终必有对应返还（守恒闭环）。

type SignalBatch 

type SignalBatch struct {
	SchemaURL string          `json:"schema_url"`
	Resource  Resource        `json:"resource"`
	Scope     Scope           `json:"scope"`
	Signals   []*SignalRecord `json:"signals"`
}

SignalBatch 是 BATCH payload 的规范结构，对齐协议文档 §6。

func UnmarshalSignalBatchCodec 

func UnmarshalSignalBatchCodec(codec corepb.Codec, data []byte) (*SignalBatch, error)

UnmarshalSignalBatchCodec 按 codec 分发解码 SignalBatch。

func (*SignalBatch) Validate 

func (b *SignalBatch) Validate() error

Validate 校验 SignalBatch 必填字段。

type SignalCodec 

type SignalCodec interface {
	// Codec 返回该实现对应的 wire enum 值。
	Codec() corepb.Codec
	// Marshal 将 SignalBatch 编码为 bytes。
	Marshal(sb *SignalBatch) ([]byte, error)
	// Unmarshal 从 bytes 解码出 SignalBatch。
	Unmarshal(data []byte) (*SignalBatch, error)
}

SignalCodec 是 SignalBatch 的编码/解码契约（core spec §6.3）。

每种 wire Codec（PROTO/CBOR）实现一个，通过 RegisterCodec 注册到包级注册表。 协议核心（internal/protocol）通过 MarshalSignalBatchCodec / UnmarshalSignalBatchCodec 按 HELLO 协商结果分发，不直接依赖具体 codec 实现——新增 codec 只需 RegisterCodec， 不改动分发逻辑。

与 cipher.go / envelope.go 的 switch 分发不同：codec 允许第三方注册（私有互通场景）， 而 cipher/compression 是固定内置集合，故前者用注册表、后者用 switch。

func LookupCodec 

func LookupCodec(codec corepb.Codec) SignalCodec

LookupCodec 返回 codec 对应实现；未注册返回 nil。

type SignalRecord 

type SignalRecord struct {
	SignalType     string         `json:"signal_type"` // 必填，禁止空字符串
	EventID        string         `json:"event_id,omitempty"`
	TimeUnixMs     int64          `json:"time_unix_ms"`
	ObservedTimeMs int64          `json:"observed_time_unix_ms,omitempty"`
	TraceID        string         `json:"trace_id,omitempty"`
	SpanID         string         `json:"span_id,omitempty"`
	ParentSpanID   string         `json:"parent_span_id,omitempty"`
	Attributes     map[string]any `json:"attributes,omitempty"`
	Body           any            `json:"body,omitempty"`
	SeverityNumber int            `json:"severity_number,omitempty"`
	SeverityText   string         `json:"severity_text,omitempty"`
	// 指标字段
	MetricName   string             `json:"metric_name,omitempty"`
	MetricFields map[string]float64 `json:"metric_fields,omitempty"`
	Unit         string             `json:"unit,omitempty"`
	Temporality  string             `json:"temporality,omitempty"`
	IsMonotonic  bool               `json:"is_monotonic,omitempty"`
	// Span 字段
	Name            string `json:"name,omitempty"`
	Kind            string `json:"kind,omitempty"`
	StartTimeUnixMs int64  `json:"start_time_unix_ms,omitempty"`
	EndTimeUnixMs   int64  `json:"end_time_unix_ms,omitempty"`
	StatusCode      string `json:"status_code,omitempty"`
	StatusMessage   string `json:"status_message,omitempty"`
	// W3C Trace Context 字段（capability w3c_trace_context，core spec §6.2.2）。
	// trace_flags 承载 traceparent 的采样位等（低 8 位有效）；
	// trace_state 承载 W3C tracestate（有序键值对）。
	// 这两个字段与 trace_id/span_id/parent_span_id 一起构成完整的 W3C traceparent 语义，
	// 使外部请求携带的 traceparent 能在协议层无损承载，而非退化塞入 attributes。
	TraceFlags uint32             `json:"trace_flags,omitempty"`
	TraceState []*TraceStateEntry `json:"trace_state,omitempty"`
	// Histogram / Profile 载荷（core spec §6.2）。
	// exp_histogram 用于 signal_type=histogram 且 aggregation=exponential；
	// profile 用于 signal_type=profile（OTLP Profiles 映射，附录 B）。
	ExpHistogram *ExponentialHistogram `json:"exp_histogram,omitempty"`
	Profile      *ProfilePayload       `json:"profile,omitempty"`
}

SignalRecord 统一信号记录外壳。

type SinkOptions added in v0.6.0 

type SinkOptions struct {
	Raw     bool // 供给原始编码字节（跳过逐事件解码）；Raw=true 时 BatchEvent.Raw 非 nil
	Durable bool // handler 会调用 done 表达持久化结果（服务端 ack_mode=durable 的前置）
	Async   bool // done 允许在 handler 返回后异步调用（协议层保证幂等与超时合成）
}

SinkOptions 是 sink handler 注册时声明的供给偏好。

type State 

type State int
const (
	// StateIdle 初始态（含握手进行中——握手子步骤不再是 wire 可见状态）。
	StateIdle State = iota
	// StateReady 就绪，可发送。
	StateReady
	// StateDraining 排空中（等待全部已发 seq 终态）。
	StateDraining
	// StateClosed 终态。任意状态可达（close 收敛保证）。
	StateClosed
)

func (State) String 

func (s State) String() string

type StateMachine 

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

StateMachine 是线程安全的客户端状态机。

func NewStateMachine 

func NewStateMachine() *StateMachine

func (*StateMachine) ForceClosed added in v0.6.0 

func (sm *StateMachine) ForceClosed()

ForceClosed 从任意状态收敛到 Closed（close 专用；幂等）。

func (*StateMachine) State 

func (sm *StateMachine) State() State

func (*StateMachine) Transition 

func (sm *StateMachine) Transition(next State) error

type StaticTokenValidator 

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

StaticTokenValidator 是基于内存 map 的静态 token 表（测试/简单部署）。

func NewStaticTokenValidator 

func NewStaticTokenValidator(tokens map[string]string) *StaticTokenValidator

NewStaticTokenValidator 构造静态 token 表。同 agent 多 token 时按字典序 注册（确定性），输入 map 防御性拷贝（validator 跨连接共享）。

func (*StaticTokenValidator) ResolveToken 

func (v *StaticTokenValidator) ResolveToken(agentID string) ([]string, error)

func (*StaticTokenValidator) Validate 

func (v *StaticTokenValidator) Validate(token string) (string, error)

type TerminalEntry added in v0.6.0 

type TerminalEntry struct {
	Level uint32 // corepb.ResultLevel 数值
	Code  uint32
}

TerminalEntry 是终态环条目。

type TokenValidator 

type TokenValidator interface {
	ResolveToken(agentID string) ([]string, error)
	Validate(token string) (agentID string, err error)
}

TokenValidator 是服务端 token 认证的必选接口。

  • ResolveToken：按 agentID 返回候选 token（确定性顺序）；空列表 = 无已知 token。
  • Validate：直接校验 token（保留给需要明文校验的部署面）。

type TraceContext added in v0.2.0 

type TraceContext = corepb.TraceContext

TraceContext 是 batch/stream 级 W3C trace 上下文继承点（core spec §6.2.2， capability w3c_trace_context）。等价 corepb.TraceContext。

type TraceStateEntry added in v0.2.0 

type TraceStateEntry = corepb.TraceStateEntry

TraceStateEntry 是 W3C tracestate 的有序键值对成员（core spec §6.2.2 / W3C Trace Context）。 等价 corepb.TraceStateEntry。

type ZstdDict added in v0.6.0 

type ZstdDict = corepb.ZstdDict

type ZstdDictCodec added in v0.6.0 

type ZstdDictCodec struct {
	ID uint32
	// contains filtered or unexported fields
}

--- zstd 字典（capability: zstd_dict，服务端 DICT 帧下发）---

进程级单例（简单优先，r3 裁决）：COMPRESSION_ZSTD_DICT 隐含"当前字典" （dict_id 由 DICT 帧带外声明，Envelope 不重复携带）。服务端在配置字典时 ZstdDictCodec 是字典绑定的编解码器：池化 encoder（逐调用实例，单并发最优） + 共享 decoder。共享的是池（性能——dict-bound encoder 构造昂贵）， 隔离的是选择权（正确性——连接持有指针引用，进程单例的覆盖污染不再可能）。

func RegisterZstdDict added in v0.6.0 

func RegisterZstdDict(id uint32, dict []byte) (*ZstdDictCodec, error)

RegisterZstdDict 注册/获取字典 codec（幂等：同 (id, 内容) 返回同一实例）。 超出注册表上界时淘汰最久未使用条目。

Source Files

View all Source files

Jump to

Keyboard shortcuts

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