durable

type AppendResult ¶ added in v1.0.24

type AppendResult struct {
	Event     Event
	Duplicate bool
}

AppendResult records the outcome of one idempotent log append. A duplicate Dot is safe only when the store verified that its canonical payload is identical to the existing binding.

type Authenticate ¶

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

Authenticate authenticates a request before the WebSocket upgrade.

type Authorize ¶

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

Authorize binds a proposed CRDT change to the authenticated peer and exact manifest. At minimum it must prevent a peer from publishing another actor.

type AuthorizeSubscription ¶

type AuthorizeSubscription func(Peer, replica.Manifest) error

AuthorizeSubscription controls replay/live-event access independently from write authorization.

type ClientConfig ¶

type ClientConfig struct {
	Header                http.Header
	HTTPClient            *http.Client
	Policy                crdt.ProtocolPolicy
	MaxMessageBytes       int
	MaxActorBytes         int
	MaxQueuedChanges      int
	MaxStateVectorEntries int
	HandshakeTimeout      time.Duration
	WriteTimeout          time.Duration
	PingInterval          time.Duration
	PingTimeout           time.Duration
	MinReconnectBackoff   time.Duration
	MaxReconnectBackoff   time.Duration
	Cursor                uint64
	// StateVector returns the contiguous frontier from the same durable
	// application checkpoint as the CRDT state. When set, the client requests
	// the v2 bounded missing-Dot catch-up protocol on every reconnect.
	StateVector func() replica.Frontier
	// OnCatchUp persists the state/frontier after all v2 catch-up events and
	// records highWater as the durable cursor in the same transaction. It is
	// required with StateVector because a skipped log event is never proof that
	// its payload was installed.
	OnCatchUp func(highWater uint64) error
	OnEvent   func(Event) error
}

ClientConfig configures a reconnecting durable WebSocket client. OnEvent must durably install the concrete CRDT state and delivery frontier before it returns nil. Its transaction must also record event.Sequence as the resume cursor and settle any matching application outbox row.

type Config ¶

type Config struct {
	Store                 Log
	Groups                []*Group
	Authenticate          Authenticate
	Authorize             Authorize
	AuthorizeSubscription AuthorizeSubscription
	// RevalidateSubscription is optional. When provided, it runs before every
	// heartbeat so the host can close revoked or expired long-lived sessions.
	RevalidateSubscription RevalidateSubscription
	OriginPatterns         []string
	MaxMessageBytes        int
	MaxActorBytes          int
	MaxQueuedEvents        int
	MaxQueuedBytes         int
	MaxReplayEvents        int
	MaxReplayBytes         int
	MaxStateVectorEntries  int
	HandshakeTimeout       time.Duration
	WriteTimeout           time.Duration
	PingInterval           time.Duration
	PingTimeout            time.Duration
	// Telemetry receives bounded, payload-free operational events for
	// handshake, replay, and append outcomes. A nil Reporter is the default
	// and adds no reporting work to relay paths.
	Telemetry *telemetry.Reporter
}

Config configures an authenticated durable WebSocket relay. Store and all authorization callbacks are required; the handler never starts a listener.

type Event ¶

type Event struct {
	Sequence uint64
	Change   replica.Change
}

Event is one committed transport-log entry. Sequence is strictly increasing within its group and is the only valid durable replay cursor.

type Group ¶

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

Group owns the manifest, validation boundary, and live subscribers for one durable operation log. It does not own concrete application CRDT state.

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

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

type GroupConfig ¶

type GroupConfig struct {
	Manifest replica.Manifest
	Policy   crdt.ProtocolPolicy
	Validate Validate
}

GroupConfig defines one manifest-bound durable transport group.

type Handler ¶

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

Handler is safe to mount into an application-owned HTTP server. It exposes only GET /ws and requires Subprotocol on every accepted connection.

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

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

type Log ¶ added in v1.0.24

type Log interface {
	Append(groupID string, change replica.Change) (AppendResult, error)
	Replay(groupID string, after, maxEvents, maxBytes uint64, manifest replica.Manifest, policy crdt.ProtocolPolicy, maxMessageBytes, maxActorBytes int) ([]Event, uint64, error)
	Closed() bool
}

Log is the durable-relay storage contract. Implementations must make Append atomic: a new event, its group-local sequence, the Dot-to-canonical-payload binding, and capacity accounting either all become durable or none do.

Replay must return a contiguous complete suffix or ErrReplayUnavailable; returning a prefix silently would let a receiver advance its cursor past missing CRDT changes. Implementations validate stored bytes against the manifest and policy supplied by the relay before returning them.

The relay never closes a Log. Its owner controls connection lifetime so a shared PostgreSQL or Redis client pool can serve more than one handler.

type Peer ¶

type Peer struct {
	ID string
}

Peer is an authenticated application identity. ID must be stable and must never be copied from a client-controlled CRDT actor identifier.

type ReconnectClient ¶

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

ReconnectClient reconnects with an application-provided durable cursor. Its in-memory queue is deliberately bounded and is not a replacement for an application outbox; persist an outgoing change before calling Publish.

func NewReconnectClient(endpoint string, manifest replica.Manifest, config ClientConfig) (*ReconnectClient, error)

func (client *ReconnectClient) Cursor() uint64

func (client *ReconnectClient) Err() error

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

func (client *ReconnectClient) Run(ctx context.Context) error

type RevalidateSubscription ¶ added in v1.0.31

type RevalidateSubscription func(Peer, replica.Manifest) error

RevalidateSubscription is invoked periodically for an established subscription. Hosts use it to apply session expiry or revocation policy to long-lived connections; a failure closes only that connection.

type StateVectorLog ¶ added in v1.0.31

type StateVectorLog interface {
	Log
	CatchUp(groupID string, vector replica.Frontier, maxEvents, maxBytes uint64, manifest replica.Manifest, policy crdt.ProtocolPolicy, maxMessageBytes, maxActorBytes int) ([]Event, uint64, error)
}

StateVectorLog is an optional Log capability for bounded recovery from a replica.Frontier. CatchUp must return every event whose Dot is not covered by vector, ordered by durable sequence, or return an error. It must never return a convenient partial result.

The base Log contract intentionally remains cursor-based so existing Redis, PostgreSQL, and host implementations continue to work with v1 clients.

type Store ¶

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

Store is a bbolt-backed, single-writer operation log. bbolt enforces an exclusive file lock; deployments must still schedule only one active relay process for a data file.

func OpenStore(path string, config StoreConfig) (*Store, error)

func (store *Store) Append(groupID string, change replica.Change) (AppendResult, error)

func (store *Store) CatchUp(groupID string, vector replica.Frontier, maxEvents, maxBytes uint64, manifest replica.Manifest, policy crdt.ProtocolPolicy, maxMessageBytes, maxActorBytes int) ([]Event, uint64, error)

func (store *Store) Close() error

func (store *Store) Closed() bool

func (store *Store) Replay(groupID string, after, maxEvents, maxBytes uint64, manifest replica.Manifest, policy crdt.ProtocolPolicy, maxMessageBytes, maxActorBytes int) ([]Event, uint64, error)

type StoreConfig ¶

type StoreConfig struct {
	MaxEvents   uint64
	MaxBytes    uint64
	OpenTimeout time.Duration
}

StoreConfig bounds retained canonical event data per replication group. Both limits are required: durable replay must apply an explicit overload policy rather than retaining unbounded history.

type Validate ¶

type Validate func([]byte) error

Validate checks a concrete CRDT delta before it is persisted or relayed. It must use application-selected bounds, make no application-state change, and return an error for an invalid payload. A frame checksum alone is not a sufficient semantic validator.