durable

package module
v1.0.36 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Overview

Package durable provides a single-writer WebSocket relay reference with a persistent operation log, bounded replay, and reconnect support.

It is intentionally separate from extensions: extensions is a bounded live relay, while durable owns a bbolt-backed transport log. It remains a reference for one active process and one persistent volume, not a clustered replication service. Applications must still supply TLS, authentication, authorization, concrete CRDT state/frontier checkpoints, durable outboxes, membership, and tombstone-GC policy.

Index

Constants

View Source 
const MerkleSubprotocol = "crdt-durable-v3"

MerkleSubprotocol identifies the optional HLC/Merkle anti-entropy protocol. It does not send a replica state vector. Each committed durable event has a relay-generated HLC identity, and a Merkle root commits to its canonical event contents. Roots only detect divergence; they do not authenticate a peer, acknowledge persistence, or authorize tombstone compaction.

View Source 
const StateVectorSubprotocol = "crdt-durable-v2"

StateVectorSubprotocol identifies the optional state-vector catch-up protocol. A v2 peer proves only its own installed Dot prefixes; the vector is neither authentication, a receipt, nor permission to compact tombstones.

View Source 
const Subprotocol = "crdt-durable-v1"

Subprotocol identifies the stable cursor-resume durable relay protocol. It is independent from CRDT frame versions and extensions.Subprotocol.

Variables

View Source 
var (
	// ErrInvalidConfig reports a missing or unsafe durable relay configuration.
	ErrInvalidConfig = errors.New("crdt durable: invalid configuration")
	// ErrUnauthorized reports authentication or authorization failure.
	ErrUnauthorized = errors.New("crdt durable: unauthorized")
	// ErrConflictingDot reports a retry that reuses an existing Dot with a
	// different canonical payload. The existing binding remains authoritative.
	ErrConflictingDot = errors.New("crdt durable: conflicting dot")
	// ErrStoreFull reports that retaining another event would exceed the
	// configured operation-log budget. The server does not evict history.
	ErrStoreFull = errors.New("crdt durable: operation log limit reached")
	// ErrReplayUnavailable reports an invalid cursor or a replay that exceeds
	// the configured bounded replay window. Callers must bootstrap from a
	// validated application checkpoint instead of accepting a partial replay.
	ErrReplayUnavailable = errors.New("crdt durable: replay unavailable")
	// ErrStateVectorUnavailable reports that a storage implementation cannot
	// calculate a complete bounded suffix for a durable delivery frontier.
	// Callers must fall back to a valid cursor replay or bootstrap from a
	// validated application checkpoint; accepting a partial catch-up is unsafe.
	ErrStateVectorUnavailable = errors.New("crdt durable: state-vector catch-up unavailable")
	// ErrAntiEntropyUnavailable reports that a storage implementation cannot
	// produce a complete bounded HLC/Merkle inventory. The caller must
	// bootstrap from a validated application checkpoint rather than accepting a
	// partial repair.
	ErrAntiEntropyUnavailable = errors.New("crdt durable: HLC/Merkle anti-entropy unavailable")
	// ErrMerkleDiverged reports a local inventory that contains a leaf absent
	// from the authoritative relay view or the same HLC identity with a
	// different digest. Repair must bootstrap or be investigated; it must not
	// silently drop either history.
	ErrMerkleDiverged = errors.New("crdt durable: HLC/Merkle histories diverged")
	// ErrCorruptStore reports damaged or internally inconsistent durable data.
	// The relay fails closed rather than guessing which operation to omit.
	ErrCorruptStore = errors.New("crdt durable: corrupt store")
	// ErrClosed reports use of a closed client or store.
	ErrClosed = errors.New("crdt durable: closed")
	// ErrQueueFull reports a bounded peer or client queue that cannot accept
	// another message without unbounded memory growth.
	ErrQueueFull = errors.New("crdt durable: queue full")
)

Functions

func DecodeChange 

func DecodeChange(data []byte, maxMessageBytes, maxActorBytes int) (replica.Dot, []byte, error)

DecodeChange decodes a bounded durable-log envelope. Storage providers must construct a replica.Change with the expected manifest and policy before returning this data to a relay.

func EncodeChange 

func EncodeChange(change replica.Change) ([]byte, error)

EncodeChange produces the canonical durable-log envelope for one validated CRDT change. The envelope is not authentication; callers still bind it to an authenticated manifest and actor at their transport boundary.

Types

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 

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

NewGroup validates one immutable-by-convention manifest and requires a state-independent concrete CRDT validator.

func (*Group) Manifest 

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

Manifest returns the group's immutable-by-convention 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 

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

NewHandler validates a complete, bounded durable-relay configuration.

func (*Handler) ServeHTTP 

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

ServeHTTP exposes a single durable WebSocket endpoint at /ws.

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 added in v1.0.34 

func NewMerkleIndex() *MerkleIndex

NewMerkleIndex constructs an empty HLC/Merkle inventory.

func (*MerkleIndex) Put added in v1.0.34 

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

Put records one durably accepted v3 event. Call it only after the same application transaction has persisted the concrete CRDT state and event identity; the helper itself is not a persistence layer.

func (*MerkleIndex) Reconcile added in v1.0.34 

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

Reconcile compares a complete remote inventory with this local inventory and returns the sorted relay HLC identities missing locally. A local-only or different-digest leaf is a fail-closed divergence, not a deletion request.

func (*MerkleIndex) Root added in v1.0.34 

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

Root returns the current canonical inventory root. A nil index returns the canonical empty root so callers can use it for an empty durable checkpoint.

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 

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

NewReconnectClient validates configuration without making a network call.

func (*ReconnectClient) Cursor 

func (client *ReconnectClient) Cursor() uint64

Cursor reports the highest event whose OnEvent callback succeeded during the current process. On restart, supply the cursor loaded from the same durable application transaction as the CRDT state/frontier.

func (*ReconnectClient) Err 

func (client *ReconnectClient) Err() error

Err returns the last transient session error observed by Run. A successful handshake clears it; callers still own logging and operational policy.

func (*ReconnectClient) Publish 

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

Publish validates and queues one change for the next connected session. It only confirms bounded in-memory acceptance. The caller must retain its own durable outbox until it observes the echoed committed Event in OnEvent.

func (*ReconnectClient) Run 

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

Run maintains sessions until ctx is cancelled. It returns ErrReplayUnavailable without retrying because accepting a partial replay is unsafe; the caller must bootstrap a validated checkpoint first.

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 

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

OpenStore opens or creates a durable operation log at path with mode 0600. The parent directory must already be owned and protected by the host.

func (*Store) Append 

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

Append transactionally binds a Dot to its canonical envelope and allocates the next group-local sequence for new data. The caller must validate the concrete CRDT delta before invoking Append.

func (*Store) CatchUp 

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

CatchUp returns the complete bounded set of events not covered by vector. The vector represents only contiguous locally installed Dot prefixes. A vector ahead of retained log data, or a required suffix beyond the caller's explicit limits, fails closed rather than silently declaring a client synced.

func (*Store) Close 

func (store *Store) Close() error

Close releases the database lock. Calls after Close fail with ErrClosed.

func (*Store) Closed 

func (store *Store) Closed() bool

Closed reports whether Close has completed or is in progress. It allows a Handler to fail closed without taking ownership of the store's lifetime.

func (*Store) MerkleEnabled added in v1.0.34 

func (store *Store) MerkleEnabled() bool

MerkleEnabled reports whether this store was explicitly configured to persist a relay HLC with each event. A disabled store remains a valid v1/v2 Log but must not advertise the v3 anti-entropy protocol.

func (*Store) MerkleEvents added in v1.0.34 

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

MerkleEvents returns the complete requested event set in canonical HLC order. A request for an unknown HLC identity fails closed: partial repair would let a client attest to a root it cannot actually reconstruct.

func (*Store) MerkleSnapshot added in v1.0.34 

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

MerkleSnapshot returns a complete bounded HLC/Merkle view of one durable group. It verifies every retained event before it contributes to the root; an absent HLC tag or index is unsafe because a receiver could otherwise accept a root that does not name all replayable events.

func (*Store) Replay 

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

Replay atomically reads every event after after in sequence order. It fails rather than returning a prefix when the caller's explicit replay budget cannot cover the entire missed suffix.

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.

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.