replication

package
v1.0.5 Latest Latest
Warning

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

 Go to latest Published: Aug 13, 2026 License: MIT Imports: 18 Imported by: 0

Documentation

Overview

Package replication contains reusable transport-side building blocks for cube-core/replication. Protocol state and delta algorithms stay in core; queueing and concrete network adaptation belong in kit.

Index

Constants

View Source 
const DefaultQUICALPN = "cube-replication-v1"
View Source 
const DefaultUDPMaxPacketBytes = 1232 // IPv6 minimum MTU minus IPv6 + UDP headers

Variables

View Source 
var (
	ErrTransportRequired     = errors.New("replication transport: downstream transport is required")
	ErrTransportClosed       = errors.New("replication transport: transport is closed")
	ErrSessionNotRegistered  = errors.New("replication transport: session is not registered")
	ErrSessionAlreadyExists  = errors.New("replication transport: session is already registered")
	ErrSessionFailed         = errors.New("replication transport: session transport has failed")
	ErrSessionLimit          = errors.New("replication transport: session limit exceeded")
	ErrReliableBackpressure  = errors.New("replication transport: reliable queue is full")
	ErrInvalidDatagramBatch  = errors.New("replication transport: invalid datagram batch")
	ErrReliableMessageTooBig = errors.New("replication transport: reliable message is too large")
	ErrRouteNotBound         = errors.New("replication transport: session network route is not bound")
	ErrProtocolConfig        = errors.New("replication transport: invalid protocol configuration")
	ErrPayloadTooLarge       = errors.New("replication transport: protocol payload is too large")
	ErrAuthentication        = errors.New("replication transport: packet authentication failed")
)

Functions

func DialKCP 

func DialKCP(address string, block kcp.BlockCrypt, dataShards, parityShards int) (*kcp.UDPSession, error)

func DialQUIC 

func DialQUIC(ctx context.Context, address string, tlsConfig *tls.Config, config *quic.Config) (*quic.Conn, error)

func KCPRemoteAddress 

func KCPRemoteAddress(session *kcp.UDPSession) net.Addr

func ListenKCP 

func ListenKCP(address string, block kcp.BlockCrypt, dataShards, parityShards int) (*kcp.Listener, error)

func ListenQUIC 

func ListenQUIC(address string, tlsConfig *tls.Config, config *quic.Config) (*quic.Listener, error)

func NewKCPAESGCM 

func NewKCPAESGCM(key []byte) (kcp.BlockCrypt, error)

func QUICConfig 

func QUICConfig(config *quic.Config) *quic.Config

func RandomNonceSalt 

func RandomNonceSalt() ([4]byte, error)

func ValidateQUICTLS 

func ValidateQUICTLS(config *tls.Config) error

Types

type AEADSessionProtector 

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

AEADSessionProtector authenticates the UDP routing header, encrypts the replication packet, and rejects duplicate or stale packets with a 64-packet replay window. SendSalt and ReceiveSalt must be exchanged by an authenticated handshake and must differ for the two directions.

func NewAESGCMProtector 

func NewAESGCMProtector(key []byte, sendSalt, receiveSalt [4]byte) (*AEADSessionProtector, error)

func (*AEADSessionProtector) Open 

func (protector *AEADSessionProtector) Open(expected core.SessionID, packet []byte) ([]byte, error)

func (*AEADSessionProtector) Overhead 

func (protector *AEADSessionProtector) Overhead() int

func (*AEADSessionProtector) Seal 

func (protector *AEADSessionProtector) Seal(session core.SessionID, payload []byte) ([]byte, error)

type AsyncTransport 

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

func NewAsyncTransport 

func NewAsyncTransport(downstream core.Transport, config AsyncTransportConfig) (*AsyncTransport, error)

func (*AsyncTransport) Close 

func (transport *AsyncTransport) Close(ctx context.Context) error

Close rejects new work, drains each session's currently admitted work, and then stops. If ctx expires, in-flight sends are cancelled and ctx.Err is returned. Downstream transports must honor the provided send context.

func (*AsyncTransport) RegisterSession 

func (transport *AsyncTransport) RegisterSession(info core.SessionInfo) error

RegisterSession starts two independent workers: latest-only datagrams and bounded reliable messages. This avoids reliable stream stalls blocking state.

func (*AsyncTransport) RemoveSession 

func (transport *AsyncTransport) RemoveSession(id core.SessionID) bool

RemoveSession cancels queued work immediately. Room shutdown should use Close when reliable messages need a bounded graceful drain.

func (*AsyncTransport) SendDatagram 

func (transport *AsyncTransport) SendDatagram(ctx context.Context, id core.SessionID, payload []byte) error

func (*AsyncTransport) SendDatagramBatch 

func (transport *AsyncTransport) SendDatagramBatch(ctx context.Context, id core.SessionID, packets [][]byte) error

SendDatagramBatch atomically replaces the pending state frame. It copies all packets before admission, so callers may safely reuse their buffers.

func (*AsyncTransport) SendReliable 

func (transport *AsyncTransport) SendReliable(ctx context.Context, id core.SessionID, payload []byte) error

func (*AsyncTransport) Stats 

func (transport *AsyncTransport) Stats() AsyncTransportStats

type AsyncTransportConfig 

type AsyncTransportConfig struct {
	MaxSessions          int
	ReliableQueueSize    int
	MaxDatagramsPerFrame int
	MaxDatagramBytes     int
	MaxReliableBytes     int
	SendTimeout          time.Duration
	// AllowOpaqueDatagrams disables replication header, checksum, and complete
	// frame-batch validation. Keep false unless a trusted upstream has already
	// performed equivalent validation.
	AllowOpaqueDatagrams bool
	OnError              ErrorHandler
}

func DefaultAsyncTransportConfig 

func DefaultAsyncTransportConfig() AsyncTransportConfig

type AsyncTransportStats 

type AsyncTransportStats struct {
	ActiveSessions        int
	DrainingSessions      int
	PendingDatagramFrames int
	PendingReliable       int
	DatagramSendsInFlight int
	ReliableSendsInFlight int
	DatagramFramesQueued  uint64
	DatagramFramesSent    uint64
	DatagramFramesDropped uint64
	DatagramBytesSent     uint64
	ReliableQueued        uint64
	ReliableSent          uint64
	ReliableBytesSent     uint64
	ReliableBackpressure  uint64
	ReliableAbandoned     uint64
	SendErrors            uint64
	ErrorHandlerPanics    uint64
}

type Channel 

type Channel uint8
const (
	ChannelDatagram Channel = iota + 1
	ChannelReliable
)

type CompositeTransport 

type CompositeTransport struct {
	Datagrams DatagramSender
	Reliable  ReliableSender
}

CompositeTransport joins independent unreliable and reliable network lanes. A QUIC adapter, for example, can supply a DATAGRAM sender and a stream sender.

func (CompositeTransport) SendDatagram 

func (transport CompositeTransport) SendDatagram(ctx context.Context, session core.SessionID, payload []byte) error

func (CompositeTransport) SendDatagramBatch 

func (transport CompositeTransport) SendDatagramBatch(ctx context.Context, session core.SessionID, packets [][]byte) error

func (CompositeTransport) SendReliable 

func (transport CompositeTransport) SendReliable(ctx context.Context, session core.SessionID, payload []byte) error

type DatagramBatchSender 

type DatagramBatchSender interface {
	SendDatagramBatch(context.Context, core.SessionID, [][]byte) error
}

type DatagramSender 

type DatagramSender interface {
	SendDatagram(context.Context, core.SessionID, []byte) error
}

type ErrorHandler 

type ErrorHandler func(SendError)

ErrorHandler must return promptly. Panics are contained and counted, but a blocking handler still blocks the affected session lane.

type KCPDatagramHandler 

type KCPDatagramHandler func(core.SessionID, []byte)

type KCPTransport 

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

func NewKCPTransport 

func NewKCPTransport(config KCPTransportConfig) (*KCPTransport, error)

func (*KCPTransport) BindDatagramHandler 

func (transport *KCPTransport) BindDatagramHandler(sessionID core.SessionID, handler KCPDatagramHandler) error

func (*KCPTransport) BindSession 

func (transport *KCPTransport) BindSession(sessionID core.SessionID, session *kcp.UDPSession) error

BindSession configures an authenticated KCP session. The KCP listener/dialer must enable FEC; kcp-go's OOB channel uses that framing to carry unreliable state datagrams while the normal KCP path carries reliable messages.

func (*KCPTransport) Close 

func (transport *KCPTransport) Close() error

func (*KCPTransport) ReceiveReliable 

func (transport *KCPTransport) ReceiveReliable(ctx context.Context, sessionID core.SessionID) ([]byte, error)

func (*KCPTransport) RegisterSession 

func (transport *KCPTransport) RegisterSession(info core.SessionInfo) error

func (*KCPTransport) RemoveSession 

func (transport *KCPTransport) RemoveSession(sessionID core.SessionID) bool

func (*KCPTransport) SendDatagram 

func (transport *KCPTransport) SendDatagram(ctx context.Context, sessionID core.SessionID, payload []byte) error

func (*KCPTransport) SendDatagramBatch 

func (transport *KCPTransport) SendDatagramBatch(ctx context.Context, sessionID core.SessionID, packets [][]byte) error

func (*KCPTransport) SendReliable 

func (transport *KCPTransport) SendReliable(ctx context.Context, sessionID core.SessionID, payload []byte) error

func (*KCPTransport) Stats 

func (transport *KCPTransport) Stats() KCPTransportStats

type KCPTransportConfig 

type KCPTransportConfig struct {
	MaxDatagramBytes  int
	MaxReliableBytes  int
	MTU               int
	SendWindow        int
	ReceiveWindow     int
	NoDelay           int
	Interval          int
	FastResend        int
	DisableCongestion int
	ACKNoDelay        bool
	WriteDelay        bool
	DSCP              int
	RateLimitBytes    uint32
	PreserveSessions  bool
}

func DefaultKCPTransportConfig 

func DefaultKCPTransportConfig() KCPTransportConfig

type KCPTransportStats 

type KCPTransportStats struct {
	ActiveRoutes             int
	DatagramsSent            uint64
	DatagramBytesSent        uint64
	ReliableMessagesSent     uint64
	ReliableBytesSent        uint64
	DatagramsReceived        uint64
	ReliableMessagesReceived uint64
	SendErrors               uint64
	ReceiveErrors            uint64
}

type QUICTransport 

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

func NewQUICTransport 

func NewQUICTransport(config QUICTransportConfig) *QUICTransport

func (*QUICTransport) BindSession 

func (transport *QUICTransport) BindSession(session core.SessionID, connection *quic.Conn) error

BindSession associates an authenticated QUIC connection with a replication session. QUIC DATAGRAM support must have been negotiated by both peers.

func (*QUICTransport) Close 

func (transport *QUICTransport) Close() error

func (*QUICTransport) ReceiveDatagram 

func (transport *QUICTransport) ReceiveDatagram(ctx context.Context, session core.SessionID) ([]byte, error)

func (*QUICTransport) ReceiveReliable 

func (transport *QUICTransport) ReceiveReliable(ctx context.Context, session core.SessionID) ([]byte, error)

func (*QUICTransport) RegisterSession 

func (transport *QUICTransport) RegisterSession(info core.SessionInfo) error

func (*QUICTransport) RemoveSession 

func (transport *QUICTransport) RemoveSession(session core.SessionID) bool

func (*QUICTransport) SendDatagram 

func (transport *QUICTransport) SendDatagram(ctx context.Context, session core.SessionID, payload []byte) error

func (*QUICTransport) SendDatagramBatch 

func (transport *QUICTransport) SendDatagramBatch(ctx context.Context, session core.SessionID, packets [][]byte) error

func (*QUICTransport) SendReliable 

func (transport *QUICTransport) SendReliable(ctx context.Context, session core.SessionID, payload []byte) error

func (*QUICTransport) Stats 

func (transport *QUICTransport) Stats() QUICTransportStats

type QUICTransportConfig 

type QUICTransportConfig struct {
	MaxDatagramBytes    int
	MaxReliableBytes    int
	PreserveConnections bool
	CloseErrorCode      quic.ApplicationErrorCode
}

type QUICTransportStats 

type QUICTransportStats struct {
	ActiveRoutes             int
	DatagramsSent            uint64
	DatagramBytesSent        uint64
	ReliableMessagesSent     uint64
	ReliableBytesSent        uint64
	DatagramsReceived        uint64
	ReliableMessagesReceived uint64
	SendErrors               uint64
	ReceiveErrors            uint64
}

type ReliableSender 

type ReliableSender interface {
	SendReliable(context.Context, core.SessionID, []byte) error
}

type SendError 

type SendError struct {
	Session core.SessionID
	Channel Channel
	Err     error
}

func (SendError) Error 

func (sendError SendError) Error() string

func (SendError) Unwrap 

func (sendError SendError) Unwrap() error

type UDPReceiveHandler 

type UDPReceiveHandler func(context.Context, core.SessionID, []byte, net.Addr) error

type UDPTransport 

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

func NewUDPTransport 

func NewUDPTransport(config UDPTransportConfig) (*UDPTransport, error)

func (*UDPTransport) BindSession 

func (transport *UDPTransport) BindSession(session core.SessionID, address net.Addr, protector *AEADSessionProtector) error

BindSession installs the authenticated endpoint and directional AEAD state. It may be called before RegisterSession, which is useful during handshake.

func (*UDPTransport) Close 

func (transport *UDPTransport) Close() error

func (*UDPTransport) RegisterSession 

func (transport *UDPTransport) RegisterSession(info core.SessionInfo) error

func (*UDPTransport) RemoveSession 

func (transport *UDPTransport) RemoveSession(session core.SessionID) bool

func (*UDPTransport) SendDatagram 

func (transport *UDPTransport) SendDatagram(ctx context.Context, session core.SessionID, payload []byte) error

func (*UDPTransport) SendDatagramBatch 

func (transport *UDPTransport) SendDatagramBatch(ctx context.Context, session core.SessionID, packets [][]byte) error

func (*UDPTransport) SendReliable 

func (*UDPTransport) SendReliable(context.Context, core.SessionID, []byte) error

func (*UDPTransport) Serve 

func (transport *UDPTransport) Serve(ctx context.Context, handler UDPReceiveHandler) error

func (*UDPTransport) Stats 

func (transport *UDPTransport) Stats() UDPTransportStats

type UDPTransportConfig 

type UDPTransportConfig struct {
	PacketConn            net.PacketConn
	MaxPacketBytes        int
	AllowAddressMigration bool
	OwnPacketConn         bool
	ReadPollInterval      time.Duration
	OnReceiveError        func(error, net.Addr)
}

type UDPTransportStats 

type UDPTransportStats struct {
	ActiveRoutes      int
	PacketsSent       uint64
	BytesSent         uint64
	PacketsReceived   uint64
	BytesReceived     uint64
	AuthFailures      uint64
	UnknownSessions   uint64
	AddressMigrations uint64
	SendErrors        uint64
	ReceiveErrors     uint64
}

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.