durable

type AppendResult ¶

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
	MaxMerkleLeaves       int
	MaxMerkleBytes        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
	// MerkleRoot returns the root reconstructed from the same durable local
	// event inventory as the concrete CRDT checkpoint. When all three Merkle
	// callbacks are supplied, the client requests v3 without sending a state
	// vector.
	MerkleRoot func() [32]byte
	// ReconcileMerkle compares the complete, bounded remote inventory against
	// that durable local inventory and returns the sorted HLC identities absent
	// locally. It must reject an unexpected local-only or differently-digested
	// leaf instead of silently accepting a divergent history.
	ReconcileMerkle func([]MerkleLeaf) ([]crdt.Tag, error)
	// OnMerkleCatchUp atomically records the completed checkpoint boundary
	// after all requested events have been installed and MerkleRoot equals the
	// remote root. It is required for a v3 client.
	OnMerkleCatchUp func(MerkleBoundary) 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
	MaxMerkleLeaves        int
	MaxMerkleBytes         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
	// HLC is the relay-persisted HLC identity used by MerkleSubprotocol. It is
	// zero for v1/v2 stores and events. It is independent from tags contained
	// inside an application CRDT delta.
	HLC    crdt.Tag
	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 ¶

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, MySQL, SQL Server, SQLite, or Redis client pool can serve more than one handler.

type MerkleBoundary ¶ added in v1.0.34

type MerkleBoundary struct {
	Root      [sha256.Size]byte
	HighWater uint64
	HLC       crdt.Tag
}

MerkleBoundary is the durable checkpoint boundary emitted after a v3 repair. The application persists it with its concrete CRDT state, HLC state, local Merkle inventory, and cursor before the client accepts live traffic.

type MerkleIndex ¶ added in v1.0.34

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

MerkleIndex is a small in-process helper for an application's durable event inventory. It mirrors only HLC identities and canonical event digests; it never stores CRDT payloads. Applications must persist the equivalent index with their concrete CRDT checkpoint before reporting Root to a v3 client.

Put is idempotent for the same immutable event and rejects a same-HLC, different-digest conflict. Reconcile refuses local-only history instead of turning a Merkle mismatch into accidental data loss.

func NewMerkleIndex() *MerkleIndex

func (index *MerkleIndex) Put(event Event) error

func (index *MerkleIndex) Reconcile(remote []MerkleLeaf) ([]crdt.Tag, error)

func (index *MerkleIndex) Root() [sha256.Size]byte

type MerkleLeaf ¶ added in v1.0.34

type MerkleLeaf struct {
	HLC    crdt.Tag
	Digest [sha256.Size]byte
}

MerkleLeaf identifies one immutable event in a Merkle anti-entropy inventory. Digest is SHA-256 over the canonical HLC plus durable change envelope; HLC is the leaf key and is allocated atomically with that event. Neither field is an authorization credential.

type MerkleLog ¶ added in v1.0.34

type MerkleLog interface {
	Log
	MerkleEnabled() bool
	MerkleSnapshot(groupID string, maxLeaves, maxBytes uint64, manifest replica.Manifest, policy crdt.ProtocolPolicy, maxMessageBytes, maxActorBytes int) (MerkleSnapshot, error)
	MerkleEvents(groupID string, identities []crdt.Tag, maxEvents, maxBytes uint64, manifest replica.Manifest, policy crdt.ProtocolPolicy, maxMessageBytes, maxActorBytes int) ([]Event, error)
}

MerkleLog is an optional Log capability for bounded no-state-vector anti-entropy. A caller first compares a MerkleSnapshot root, then, only on divergence, compares its complete bounded leaf inventory and requests the HLC identities it lacks. Snapshot and Events must fail closed when either requested set cannot be complete within the supplied limits.

MerkleEnabled permits a storage implementation to keep v1/v2 support while requiring explicit HLC persistence before it advertises v3.

type MerkleSnapshot ¶ added in v1.0.34

type MerkleSnapshot struct {
	Root      [sha256.Size]byte
	HighWater uint64
	HLC       crdt.Tag
	Leaves    []MerkleLeaf
}

MerkleSnapshot is one immutable relay-log view. HighWater and HLC form the replay-to-live boundary; Leaves is a complete bounded inventory matching Root. Empty logs have a zero HLC and the canonical empty Merkle root.

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 ¶

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 ¶

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, MySQL, SQL Server, SQLite, 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) MerkleEnabled() bool

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

func (store *Store) MerkleSnapshot(groupID string, maxLeaves, maxBytes uint64, manifest replica.Manifest, policy crdt.ProtocolPolicy, maxMessageBytes, maxActorBytes int) (MerkleSnapshot, error)

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
	// HLCReplicaID enables the no-state-vector HLC/Merkle anti-entropy
	// capability. The store persists this relay-local clock in the same bbolt
	// transaction as every newly committed event. Empty preserves the legacy
	// cursor/state-vector-only behavior.
	HLCReplicaID string
}

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.