extensions

package
v1.0.23-beta.3 Latest Latest
Warning

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

Go to latest
Published: Jul 30, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Overview

Package extensions provides opt-in, bounded live transport adapters for CRDT replication groups.

Its zero-value feature set exposes no endpoint and never starts an HTTP listener. Applications remain responsible for TLS, identity lifecycle, durable state and outbox transactions, bootstrap/anti-entropy, membership, rate limits, and operations. The adapters are deliberately not a production replication service.

Index

Constants

View Source
const (
	// Subprotocol identifies this package's control and binary change envelope.
	// It is independent from CRDT frame-format versions.
	Subprotocol = "crdt-sync-v1"
	// BatchSubprotocol carries a bounded list of complete v1 envelopes. It is
	// opt-in and does not change the v1 envelope or CRDT frame contract.
	BatchSubprotocol = "crdt-sync-v2"
)

Variables

View Source
var (
	// ErrInvalidConfig reports a missing or unsafe extensions configuration.
	ErrInvalidConfig = errors.New("crdt extensions: invalid configuration")
	// ErrUnauthorized reports authentication or authorization failure.
	ErrUnauthorized = errors.New("crdt extensions: unauthorized")
	// ErrClosed reports use of a closed transport client.
	ErrClosed = errors.New("crdt extensions: client is closed")
	// ErrBatchUnsupported reports an unavailable batch subprotocol.
	ErrBatchUnsupported = errors.New("crdt extensions: websocket batch subprotocol is not enabled")
	// ErrBatchLimit reports a local batch that exceeds a configured boundary.
	ErrBatchLimit = errors.New("crdt extensions: websocket batch limit exceeded")
)

Functions

This section is empty.

Types

type Authenticate

type Authenticate func(*http.Request) (Peer, error)

Authenticate authenticates one transport request before it is upgraded or its body is read. Returning an error rejects the request.

type Authorize

type Authorize func(Peer, replica.Manifest, replica.Dot) error

Authorize binds one proposed CRDT change to an authenticated peer and its negotiated manifest. At minimum it should prevent a peer from publishing as another logical replica actor.

type AuthorizeSubscription

type AuthorizeSubscription func(Peer, replica.Manifest) error

AuthorizeSubscription decides whether peer may receive live changes for the manifest. Read permission is deliberately separate from write authorization.

type ClientConfig

type ClientConfig struct {
	Header          http.Header
	HTTPClient      *http.Client
	Policy          crdt.ProtocolPolicy
	MaxMessageBytes int
	MaxActorBytes   int
	// EnableBatches offers BatchSubprotocol in addition to Subprotocol. It
	// remains false by default so an existing client keeps its v1 contract.
	EnableBatches bool
	// MaxBatchChanges is a local publication and receive bound when
	// EnableBatches is true. It must not exceed the relay's configured bound;
	// both sides default to 16. It is ignored for a v1-only connection.
	MaxBatchChanges  int
	HandshakeTimeout time.Duration
	WriteTimeout     time.Duration
	OnChange         func(replica.Change) error
}

ClientConfig configures either optional transport client. OnChange must pass received changes to an application-owned, manifest-compatible replica.Inbox or an equivalently durable boundary. HTTPClient is used for HTTP/SSE and, if non-nil, as the underlying client for WebSocket dialing.

type Config

type Config struct {
	Features              Feature
	Groups                []*Group
	Authenticate          Authenticate
	Authorize             Authorize
	AuthorizeSubscription AuthorizeSubscription
	// OriginPatterns lists case-insensitive host patterns permitted to make
	// browser-originated cross-origin requests, for example "app.example" or
	// "*.example.internal". It uses path.Match syntax, so "*" is rejected.
	// The request host is always permitted. These are host patterns rather than
	// URL strings so HTTP and WebSocket apply the same rule.
	OriginPatterns    []string
	MaxMessageBytes   int
	MaxActorBytes     int
	MaxQueuedMessages int
	MaxQueuedBytes    int
	// MaxBatchChanges bounds independently identified changes in one
	// crdt-sync-v2 WebSocket message. It is relevant only when
	// FeatureWebSocketBatch is enabled and cannot exceed MaxQueuedMessages,
	// so v1 and SSE peers can be queued atomically or disconnected.
	MaxBatchChanges  int
	HandshakeTimeout time.Duration
	WriteTimeout     time.Duration
}

Config configures one optional transport handler. Features is default-off; no endpoint is active unless FeatureWebSocket or FeatureHTTP is selected. Authentication and separate read/write authorization are required whenever a feature is active.

type Feature

type Feature uint8

Feature selects an optional transport surface. The zero value intentionally enables no transport.

const (
	// FeatureWebSocket enables the manifest-bound WebSocket endpoint.
	FeatureWebSocket Feature = 1 << iota
	// FeatureHTTP enables the HTTP publication and SSE live-event endpoints.
	FeatureHTTP
	// FeatureWebSocketBatch enables the opt-in crdt-sync-v2 WebSocket
	// subprotocol. It requires FeatureWebSocket and leaves HTTP/SSE on their
	// single-change v1 envelope.
	FeatureWebSocketBatch
)

func (Feature) Enabled

func (f Feature) Enabled(feature Feature) bool

Enabled reports whether f includes feature.

type Group

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

Group owns a manifest-bound replica inbox and its currently connected live peers. It has no operation log, snapshot store, or replay history.

func NewGroup

func NewGroup(config GroupConfig) (*Group, error)

NewGroup creates one manifest-bound in-memory receiver. Apply runs while delivery ordering is held, so it must use the concrete CRDT decoder with application limits, leave the CRDT unchanged on an error, and not re-enter Group or block on a transport callback.

func (*Group) Frontier

func (g *Group) Frontier() replica.Frontier

Frontier returns a copy of g's installed contiguous delivery frontier.

func (*Group) Manifest

func (g *Group) Manifest() replica.Manifest

Manifest returns the immutable-by-convention manifest negotiated by g.

func (*Group) Pending

func (g *Group) Pending() (changes, bytes int)

Pending reports the number and bytes of out-of-order changes retained by g.

type GroupConfig

type GroupConfig struct {
	Manifest          replica.Manifest
	Frontier          replica.Frontier
	Policy            crdt.ProtocolPolicy
	MaxPendingChanges int
	MaxPendingBytes   int
	Apply             replica.ApplyDelta
}

GroupConfig describes one bounded, manifest-bound receiver. Frontier must come from the same durable transaction as the application CRDT state when a production application restores a group.

type HTTPClient

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

HTTPClient maintains one HTTP/SSE live subscription for a manifest. It publishes changes with POST and receives change events over SSE. It has no replay log and does not reconnect automatically.

func ConnectHTTP

func ConnectHTTP(ctx context.Context, endpoint string, manifest replica.Manifest, config ClientConfig) (*HTTPClient, error)

ConnectHTTP starts an authenticated SSE subscription, verifies the exact manifest supplied by the relay, and then receives live changes. A successful return means the relay registered the subscription before it sent the stream manifest. endpoint is the base mount URL, for example "https://sync.example.com/crdt". The configured HTTPClient must not have a global Timeout because SSE is a long-lived response; use request contexts and WriteTimeout instead.

func (*HTTPClient) Close

func (client *HTTPClient) Close() error

Close stops the SSE stream without claiming durable delivery.

func (*HTTPClient) Done

func (client *HTTPClient) Done() <-chan struct{}

Done closes when the SSE receive loop stops.

func (*HTTPClient) Err

func (client *HTTPClient) Err() error

Err returns the first non-local stream or callback error.

func (*HTTPClient) Publish

func (client *HTTPClient) Publish(ctx context.Context, change replica.Change) error

Publish validates and posts a canonical change envelope. A POST failure does not close the independent SSE subscription; callers decide whether and how to retry from their durable outbox.

type Handler

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

Handler exposes enabled optional endpoints. It is safe to mount into an application-owned http.ServeMux and never starts an HTTP listener itself.

func NewHandler

func NewHandler(config Config) (*Handler, error)

NewHandler validates config and constructs a disabled handler for the zero feature set. A disabled handler returns 404 for every request and does not invoke authentication or start background work.

func (*Handler) Mount

func (h *Handler) Mount(mux *http.ServeMux, prefix string) (err error)

Mount mounts h below prefix and strips that prefix before routing its endpoints. For example, Mount(mux, "/crdt/") exposes "/crdt/ws" and the HTTP endpoints under "/crdt/http/".

func (*Handler) ServeHTTP

func (h *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request)

ServeHTTP routes only enabled transport endpoints. It does not serve an index page or start a listener.

type Peer

type Peer struct {
	ID string
}

Peer is an authenticated application identity. ID must be stable and must not be taken from a client-supplied CRDT actor identifier.

type WebSocketClient

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

WebSocketClient maintains one manifest-bound WebSocket live connection. It has no persistent outbox and does not reconnect automatically.

func DialWebSocket

func DialWebSocket(ctx context.Context, endpoint string, manifest replica.Manifest, config ClientConfig) (*WebSocketClient, error)

DialWebSocket authenticates the WebSocket handshake through config.Header, verifies the exact manifest returned by the relay, then receives live binary changes. A successful return means the relay registered the live subscription before it sent the confirmation. Use wss after the host application has configured TLS.

func (*WebSocketClient) Close

func (client *WebSocketClient) Close() error

Close terminates the live connection. It does not make a durability claim; persist an application checkpoint before a graceful recovery boundary.

func (*WebSocketClient) Done

func (client *WebSocketClient) Done() <-chan struct{}

Done closes when the receive loop stops.

func (*WebSocketClient) Err

func (client *WebSocketClient) Err() error

Err returns the first non-local connection or callback failure.

func (*WebSocketClient) Publish

func (client *WebSocketClient) Publish(ctx context.Context, change replica.Change) error

Publish validates change against the connection manifest, then transmits its canonical envelope. A caller may retry after an ambiguous network result; durable outbox and recovery behavior remain application-owned.

func (*WebSocketClient) PublishBatch added in v1.0.19

func (client *WebSocketClient) PublishBatch(ctx context.Context, changes []replica.Change) error

PublishBatch sends independently identified changes in one explicitly negotiated WebSocket message. It is a transport coalescing operation, not an atomic application transaction. The relay can accept an earlier item and reject a later one, so callers must retain every original change in their durable outbox and retry them independently after any ambiguous connection result.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL