replica

package
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: 10 Imported by: 0

Documentation

Overview

Package replica defines the transport-independent boundary around one framed CRDT replication group.

It deliberately does not open connections, authenticate peers, retain an operation log, or compact CRDT tombstones. Its job is narrower: make the agreement required before a transport exchanges frames explicit, keep a contiguous per-actor delivery frontier, and prevent an acknowledgement from being emitted before a checkpoint has been durably installed.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	ErrInvalidManifest   = errors.New("replica: invalid manifest")
	ErrManifestMismatch  = errors.New("replica: manifest mismatch")
	ErrProtocolMismatch  = errors.New("replica: protocol mismatch")
	ErrInvalidDot        = errors.New("replica: invalid dot")
	ErrFrontierGap       = errors.New("replica: non-contiguous frontier advance")
	ErrInvalidCheckpoint = errors.New("replica: invalid checkpoint")
	ErrNilValidator      = errors.New("replica: nil state validator")
	ErrNilStore          = errors.New("replica: nil checkpoint store")
	ErrNotInstalled      = errors.New("replica: checkpoint is not installed")
	ErrCheckpointChanged = errors.New("replica: a different checkpoint is already installed")
	ErrInvalidChange     = errors.New("replica: invalid change")
	ErrDotConflict       = errors.New("replica: conflicting payload for one dot")
	ErrPendingLimit      = errors.New("replica: pending change limit exceeded")
	ErrNilApply          = errors.New("replica: nil delta apply function")
)

Functions

This section is empty.

Types

type Acknowledgement 

type Acknowledgement struct {
	GroupID      string
	Epoch        uint64
	CheckpointID [sha256.Size]byte
	Frontier     Frontier
}

Acknowledgement proves that one installed checkpoint was durably recorded. It is intentionally not a tombstone-GC acknowledgement: eligibility to compact a tombstone remains type-specific, especially for RGA and trees.

type ApplyDelta 

type ApplyDelta func([]byte) error

ApplyDelta applies one already validated canonical delta frame. It must leave the concrete CRDT unchanged on an error. Inbox advances its frontier only after this function succeeds.

type Change 

type Change struct {
	Dot Dot
	// contains filtered or unexported fields
}

Change binds exactly one canonical delta frame to a persistently assigned dot. The dot is a delivery/accounting identity, not a replacement for the CRDT's own mutation tags inside the delta payload.

func NewChange 

func NewChange(manifest Manifest, dot Dot, delta []byte) (Change, error)

NewChange validates that delta belongs to manifest's exact delta protocol and returns a copy. It does not decode the CRDT-specific payload; that work remains the concrete ApplyDelta implementation's responsibility.

func NewChangeWithPolicy added in v1.0.6 

func NewChangeWithPolicy(manifest Manifest, dot Dot, delta []byte, policy crdt.ProtocolPolicy) (Change, error)

NewChangeWithPolicy validates one change under policy. The policy parameter remains for source compatibility and future policy extensions.

func (Change) Delta 

func (c Change) Delta() []byte

Delta returns an owned copy of the canonical delta frame.

type Checkpoint 

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

Checkpoint is a validated, immutable recovery boundary for one replication group. HLC-backed CRDTs include their local clock state so the local replica cannot reuse an earlier tag after restoring the checkpoint.

func NewCheckpoint 

func NewCheckpoint(manifest Manifest, state []byte, frontier Frontier, clockState clock.State, validator StateValidator) (Checkpoint, error)

NewCheckpoint validates the manifest/frame agreement and invokes validator before a checkpoint may be persisted. For HLC-backed protocols, clockState must be valid; for non-HLC protocols it must be the zero value.

func NewCheckpointWithPolicy added in v1.0.6 

func NewCheckpointWithPolicy(manifest Manifest, state []byte, frontier Frontier, clockState clock.State, validator StateValidator, policy crdt.ProtocolPolicy) (Checkpoint, error)

NewCheckpointWithPolicy validates a checkpoint under policy. The policy parameter remains for source compatibility and future policy extensions.

func (Checkpoint) ClockState 

func (c Checkpoint) ClockState() (clock.State, bool)

ClockState reports the HLC state included in c.

func (Checkpoint) Frontier 

func (c Checkpoint) Frontier() Frontier

Frontier returns a copy of the durable delivery frontier.

func (Checkpoint) ID 

func (c Checkpoint) ID() [sha256.Size]byte

ID is a stable digest over all checkpoint meaning, including epoch, protocol semantics, state bytes, frontier, and HLC state.

func (Checkpoint) Manifest 

func (c Checkpoint) Manifest() Manifest

Manifest returns the checkpoint's manifest by value.

func (Checkpoint) State 

func (c Checkpoint) State() []byte

State returns a copy of the canonical concrete state frame.

type CheckpointStore 

type CheckpointStore interface {
	SaveCheckpoint(Checkpoint) error
}

CheckpointStore atomically records the state frame, HLC state (when any), frontier, checkpoint ID, and epoch. Its success result is the durability boundary used by Session before it emits an acknowledgement.

type Delivery 

type Delivery struct {
	Buffered  bool
	Duplicate bool
	Applied   []Dot
}

Delivery describes the result of receiving one change. Buffered reports an out-of-order change retained for its missing per-actor prefix; Applied lists every dot installed by this call, including any now-unblocked buffered changes. Duplicate reports that this call did not retain or install its change because the same dot was already known.

func (Delivery) Accepted added in v1.0.18 

func (d Delivery) Accepted() bool

Accepted reports whether this call added a change to the receiver's pending queue or installed at least one dot. Relays should forward only accepted changes: once a dot is already installed, an Inbox no longer retains its payload bytes and therefore cannot prove a later same-dot payload is identical. Durable relays must additionally bind actor/counter to payload identity in their application-owned operation store.

type Dot 

type Dot struct {
	Actor   string
	Counter uint64
}

Dot is a persistently assigned, per-group operation sequence number. It is deliberately not an HLC tag: an HLC timestamp is ordered, but is not proof that every earlier timestamped mutation was received. A Frontier advances only over contiguous dots and is therefore safe for missing-update queries.

type Frontier 

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

Frontier records the greatest contiguous dot durably installed for each actor. Its internals are private so callers cannot create a false prefix by mutating a returned map.

func NewFrontier 

func NewFrontier(entries map[string]uint64) (Frontier, error)

NewFrontier returns an immutable-by-convention frontier from contiguous sequence values. Zero entries are rejected because absence already denotes counter zero.

func (Frontier) Advance 

func (f Frontier) Advance(dot Dot) (Frontier, error)

Advance returns a frontier that includes dot. Duplicates are idempotent; receiving a future dot before its predecessor is rejected instead of making the frontier claim knowledge it cannot prove.

func (Frontier) Counter 

func (f Frontier) Counter(actor string) uint64

Counter returns the contiguous prefix installed for actor.

func (Frontier) Covers 

func (f Frontier) Covers(dot Dot) bool

Covers reports whether dot is included in f's installed prefix.

func (Frontier) Entries 

func (f Frontier) Entries() map[string]uint64

Entries returns a copy of the complete frontier.

type Inbox 

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

Inbox is a bounded, transport-independent receiver for one manifest. It accepts duplicate and out-of-order deliveries, but its frontier advances only across contiguous persisted actor counters. Applications must persist the concrete CRDT state and the resulting frontier atomically before using it as a recovery/checkpoint boundary.

func NewInbox 

func NewInbox(manifest Manifest, frontier Frontier, maxPending, maxBytes int, apply ApplyDelta) (*Inbox, error)

NewInbox creates a bounded receiver. maxPending and maxBytes cover only deferred out-of-order frames; immediately applicable frames are bounded by encoding.DefaultLimits and the concrete CRDT decoder.

func NewInboxWithPolicy added in v1.0.6 

func NewInboxWithPolicy(manifest Manifest, frontier Frontier, maxPending, maxBytes int, apply ApplyDelta, policy crdt.ProtocolPolicy) (*Inbox, error)

NewInboxWithPolicy creates a bounded receiver under policy. The policy parameter remains for source compatibility and future policy extensions.

func (*Inbox) Frontier 

func (i *Inbox) Frontier() Frontier

Frontier returns a copy of the installed contiguous delivery frontier.

func (*Inbox) Pending 

func (i *Inbox) Pending() (changes, bytes int)

Pending reports the current bounded deferred queue size.

func (*Inbox) Receive 

func (i *Inbox) Receive(change Change) (Delivery, error)

Receive applies change immediately when it is the next expected dot for its actor, otherwise it buffers it. A conflicting duplicate dot is rejected; accepting either payload would make recovery depend on arrival order.

type Manifest 

type Manifest struct {
	GroupID  string
	SchemaID string
	Epoch    uint64
	Protocol Protocol
}

Manifest is the authenticated agreement for one CRDT replication group. GroupID and SchemaID are application-defined stable names. A group carries one concrete CRDT protocol; applications that replicate multiple objects create multiple manifests rather than treating unrelated frames as one atomic document.

func NewManifest 

func NewManifest(groupID, schemaID string, epoch uint64, protocol Protocol, policy crdt.ProtocolPolicy) (Manifest, error)

NewManifest validates an immutable-by-convention manifest. policy is checked here so a caller cannot accidentally construct a group for a reserved or unknown frame type.

Example

ExampleNewManifest binds one replication group to one explicitly admitted protocol. A manifest describes an agreement; the transport must still authenticate the peer before accepting its frames. 

policy := crdt.ProtocolPolicy{AllowExperimental: true}
manifest, err := NewManifest("notes/42", "example.com/note/v1", 1, Protocol{
	StateID:          crdt.TypeIDRGAState,
	DeltaID:          crdt.TypeIDRGADelta,
	SemanticsVersion: 1,
}, policy)
if err != nil {
	panic(err)
}

fmt.Println(manifest.GroupID, manifest.Epoch)

Output:
notes/42 1

func NewManifestForFrameType added in v1.0.30 

func NewManifestForFrameType(groupID, schemaID string, epoch uint64, frameType crdt.FrameType, codecID string, policy crdt.ProtocolPolicy) (Manifest, error)

NewManifestForFrameType builds a manifest for one canonical registered frame type. It is equivalent to ProtocolFromFrameType followed by NewManifest, and retains all manifest validation and policy checks.

func (Manifest) Compatible 

func (m Manifest) Compatible(remote Manifest) error

Compatible reports whether local and remote describe the same replication group and exact CRDT semantics. It intentionally rejects a semantic-version mismatch rather than guessing that two versions can read one another.

type Protocol 

type Protocol struct {
	StateID           uint64
	DeltaID           uint64
	CodecID           string
	SemanticsVersion  uint64
	WireFormatVersion uint64
}

Protocol identifies exactly one framed CRDT protocol in a replication group. SemanticsVersion is independent of encoding.FormatVersion: changing conflict semantics, snapshot meaning, or tombstone lifecycle requires a new semantic version (and, when frame compatibility is broken, new type IDs).

CodecID is the schema/element codec identifier carried by every frame in this group. It may be empty for CRDTs whose canonical frames have no codec. WireFormatVersion selects the outer frame representation. Zero retains the v1 default for source and JSON compatibility; v2 is an explicit capability that must match across the authenticated manifest.

func ProtocolFromFrameType added in v1.0.30 

func ProtocolFromFrameType(frameType crdt.FrameType, codecID string) (Protocol, error)

ProtocolFromFrameType converts one canonical registered frame type into the protocol fields required by a manifest. It exists to keep applications from copying state IDs, delta IDs, and semantics versions by hand.

frameType must exactly match a type returned by crdt.FrameTypeForState, crdt.RegisteredFrameTypes, or a ReplicationProfile. This helper does not choose a codec, authenticate a manifest, authorize a peer, or select input limits.

func (Protocol) FrameFormatVersion added in v1.0.24 

func (p Protocol) FrameFormatVersion() uint64

FrameFormatVersion returns the negotiated outer encoding version. A zero field is the legacy spelling of encoding.FormatVersion.

type Session 

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

Session binds a manifest to an optional checkpoint. It is safe for concurrent callers. A session permits only one checkpoint ID: a rebase must create a new epoch/manifest instead of silently replacing a recovery base.

func NewSession 

func NewSession(manifest Manifest) (*Session, error)

NewSession creates a checkpoint session for one implemented-protocol manifest.

func NewSessionWithPolicy added in v1.0.6 

func NewSessionWithPolicy(manifest Manifest, policy crdt.ProtocolPolicy) (*Session, error)

NewSessionWithPolicy creates a checkpoint session under policy. The policy parameter remains for source compatibility and future policy extensions.

func (*Session) Acknowledge 

func (s *Session) Acknowledge() (Acknowledgement, error)

Acknowledge returns a proof of durable checkpoint installation. The returned frontier is a copy and cannot mutate Session state.

func (*Session) Install 

func (s *Session) Install(checkpoint Checkpoint, store CheckpointStore) error

Install persists checkpoint before publishing it to the session. If storage fails, the session remains unable to acknowledge the checkpoint.

type SessionBuilder added in v1.0.19 

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

SessionBuilder binds one validated manifest to the local protocol policy selected during authenticated group negotiation. It centralizes construction of the replication objects that must share that same policy.

A builder is not a handshake: callers must still authenticate one exact manifest before creating it. The policy parameter is retained for source compatibility and future policy extensions.

func NewSessionBuilder added in v1.0.19 

func NewSessionBuilder(groupID, schemaID string, epoch uint64, protocol Protocol, policy crdt.ProtocolPolicy) (SessionBuilder, error)

NewSessionBuilder creates a manifest and binds it to policy. Use this after authenticating the group, schema, epoch, and exact protocol agreement.

func NewSessionBuilderForFrameType added in v1.0.30 

func NewSessionBuilderForFrameType(groupID, schemaID string, epoch uint64, frameType crdt.FrameType, codecID string, policy crdt.ProtocolPolicy) (SessionBuilder, error)

NewSessionBuilderForFrameType creates a builder from one canonical registered frame type, without making callers repeat its state ID, delta ID, and semantics version. Applications must still authenticate the resulting exact manifest before a transport accepts frames.

func NewSessionBuilderFromManifest added in v1.0.19 

func NewSessionBuilderFromManifest(manifest Manifest, policy crdt.ProtocolPolicy) (SessionBuilder, error)

NewSessionBuilderFromManifest binds an already authenticated manifest to policy. It rejects a manifest that is structurally invalid or names an unknown protocol pair.

func (SessionBuilder) Manifest added in v1.0.19 

func (b SessionBuilder) Manifest() Manifest

Manifest returns the immutable-by-convention agreement used by b.

func (SessionBuilder) NewChange added in v1.0.19 

func (b SessionBuilder) NewChange(dot Dot, delta []byte) (Change, error)

NewChange validates one canonical delta frame under b's bound policy.

func (SessionBuilder) NewCheckpoint added in v1.0.19 

func (b SessionBuilder) NewCheckpoint(state []byte, frontier Frontier, clockState clock.State, validator StateValidator) (Checkpoint, error)

NewCheckpoint validates one durable recovery boundary under b's bound policy. HLC-backed protocols require their persisted clock state.

func (SessionBuilder) NewInbox added in v1.0.19 

func (b SessionBuilder) NewInbox(frontier Frontier, maxPending, maxBytes int, apply ApplyDelta) (*Inbox, error)

NewInbox creates a bounded receiver under b's bound policy.

func (SessionBuilder) NewSession added in v1.0.19 

func (b SessionBuilder) NewSession() (*Session, error)

NewSession creates a checkpoint session under b's bound policy.

type StateValidator 

type StateValidator func([]byte) error

StateValidator must perform concrete CRDT validation of one complete state frame. A frame-envelope checksum is not sufficient evidence that a snapshot is recoverable. Validators receive owned bytes, and a panic is treated as a rejected checkpoint.

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.