observe

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

Documentation

Overview

Package observe connects a CRDT to an application-owned reactive view.

A Store serializes mutations made through it and publishes an immutable application projection after a successful mutation. It is deliberately not a CRDT protocol, operation log, persistence layer, or transport. In particular, its Version values are process-local UI revisions and must not be sent to a peer or used as a replication acknowledgement.

The projection passed to New must be safe to retain after the function returns. For maps, slices, pointers, and byte slices that normally means returning an owned copy. Store shares one projection with all subscribers; subscribers must treat Event.Value as immutable.

Store invokes callbacks asynchronously and never while it holds the Store lock. A slow subscriber retains only its newest undelivered event. This bounds memory and lets a UI render the latest state, but it means observers must use Event.Coalesced and Version gaps when every intermediate mutation is significant. Durable replication belongs in replica and durable, not in this package.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNilStore reports an operation on a nil Store.
	ErrNilStore = errors.New("observe: nil store")
	// ErrNilView reports a Store constructed without an immutable view function.
	ErrNilView = errors.New("observe: nil view function")
	// ErrNilMutation reports a nil mutation function.
	ErrNilMutation = errors.New("observe: nil mutation function")
	// ErrNilCallback reports a nil subscriber callback.
	ErrNilCallback = errors.New("observe: nil callback")
	// ErrInvalidOrigin reports an origin that cannot describe a mutation.
	ErrInvalidOrigin = errors.New("observe: invalid mutation origin")
	// ErrClosed reports a mutation or subscription after Store.Close.
	ErrClosed = errors.New("observe: store is closed")
)

Functions

This section is empty.

Types

type Callback 

type Callback[V any] func(Event[V])

Callback receives one Event on a subscriber-owned goroutine. It must not retain mutable aliases from Event.Value or block indefinitely. A panic is contained, recorded on the Subscription, and unsubscribes that callback.

type Event 

type Event[V any] struct {
	Version   uint64
	Origin    Origin
	Value     V
	State     crdt.StateSnapshot
	Coalesced uint64
}

Event is one application-visible Store revision. Value is produced by the application-supplied view function while the Store serializes a mutation. It must be treated as immutable by every callback.

Coalesced reports how many older, not-yet-delivered events this event replaced for this particular subscriber. Event versions are monotonic per Store, but a subscriber may observe gaps when it is slower than mutations.

type GCounterObserver added in v1.0.36 

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

GCounterObserver owns one G-Counter mutation and observation boundary. Local increments return a delta for the authenticated application transport; ApplyDelta accepts an already-decoded remote delta. It does not establish a network connection, authenticate peers, or persist state.

func NewGCounterObserver added in v1.0.36 

func NewGCounterObserver(replicaID string) (*GCounterObserver, error)

NewGCounterObserver creates a distributed G-Counter observation boundary.

func NewGCounterObserverWithOptions added in v1.0.36 

func NewGCounterObserverWithOptions(replicaID string, options Options) (*GCounterObserver, error)

NewGCounterObserverWithOptions creates a G-Counter observer with callback panic diagnostics selected by options.

func (*GCounterObserver) ApplyDelta added in v1.0.36 

func (o *GCounterObserver) ApplyDelta(delta counter.GCounterDelta) (bool, error)

ApplyDelta joins one decoded remote G-Counter delta. It returns false for a duplicate or subsumed delta and deliberately publishes no Remote event then.

func (*GCounterObserver) Close added in v1.0.36 

func (o *GCounterObserver) Close()

Close stops this observer and its active subscriptions.

func (*GCounterObserver) Increment added in v1.0.36 

func (o *GCounterObserver) Increment(amount uint64) (counter.GCounterDelta, error)

Increment changes this replica's component and returns its joinable delta. The caller owns framing, authenticated transport, durable acknowledgement, and retry; it must not send Event.Version as a wire or causal value.

func (*GCounterObserver) Snapshot added in v1.0.36 

func (o *GCounterObserver) Snapshot() (Event[GCounterView], error)

Snapshot returns the current G-Counter projection and local UI revision.

func (*GCounterObserver) Subscribe added in v1.0.36 

func (o *GCounterObserver) Subscribe(callback Callback[GCounterView]) (*Subscription[GCounterView], error)

Subscribe atomically queues the current G-Counter projection before later counter changes may publish.

func (*GCounterObserver) SubscribeFromNow added in v1.0.36 

func (o *GCounterObserver) SubscribeFromNow(callback Callback[GCounterView]) (*Subscription[GCounterView], error)

SubscribeFromNow registers a G-Counter callback without an initial view.

type GCounterView added in v1.0.36 

type GCounterView struct {
	Value    uint64
	Overflow bool
}

GCounterView is the immutable application projection of a G-Counter. An accepted distributed state can have more valid uint64 components than fit in one uint64 aggregate; in that case Overflow is true and Value is zero.

type Options 

type Options struct {
	OnPanic func(Panic)
}

Options controls diagnostic handling for a Store. OnPanic runs after a callback panic, on the failing callback's goroutine. It must return quickly; its own panic is contained. Callback delivery never depends on this hook.

type Origin 

type Origin uint8

Origin identifies why an application-visible state update was committed. It is local process metadata, not a CRDT conflict-resolution input and not a wire-protocol field. 

const (
	// Initial describes the current state delivered immediately after Subscribe.
	Initial Origin = iota
	// Local describes a successful local user or application mutation.
	Local
	// Remote describes a successfully applied remote delta or state update.
	Remote
	// Merge describes a successful CRDT state merge.
	Merge
	// Restore describes successful installation of recovered local state.
	Restore
	// Maintenance describes a successful local maintenance operation, such as
	// an authority-approved tombstone compaction.
	Maintenance
)

func (Origin) String 

func (o Origin) String() string

String returns a stable diagnostic name for o.

type PNCounterObserver added in v1.0.36 

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

PNCounterObserver owns one PN-Counter mutation and observation boundary. Local changes return deltas for an application-owned authenticated transport; remote deltas are observed only when they extend either component map.

func NewPNCounterObserver added in v1.0.36 

func NewPNCounterObserver(replicaID string) (*PNCounterObserver, error)

NewPNCounterObserver creates a distributed PN-Counter observation boundary.

func NewPNCounterObserverWithOptions added in v1.0.36 

func NewPNCounterObserverWithOptions(replicaID string, options Options) (*PNCounterObserver, error)

NewPNCounterObserverWithOptions creates a PN-Counter observer with callback panic diagnostics selected by options.

func (*PNCounterObserver) ApplyDelta added in v1.0.36 

func (o *PNCounterObserver) ApplyDelta(delta counter.PNCounterDelta) (bool, error)

ApplyDelta joins one decoded remote PN-Counter delta. Duplicate or subsumed delivery returns changed == false and produces no Remote event.

func (*PNCounterObserver) Close added in v1.0.36 

func (o *PNCounterObserver) Close()

Close stops this observer and its active subscriptions.

func (*PNCounterObserver) Decrement added in v1.0.36 

func (o *PNCounterObserver) Decrement(amount uint64) (counter.PNCounterDelta, error)

Decrement changes this replica's negative component and returns its delta.

func (*PNCounterObserver) Increment added in v1.0.36 

func (o *PNCounterObserver) Increment(amount uint64) (counter.PNCounterDelta, error)

Increment changes this replica's positive component and returns its delta.

func (*PNCounterObserver) Snapshot added in v1.0.36 

func (o *PNCounterObserver) Snapshot() (Event[PNCounterView], error)

Snapshot returns the current PN-Counter projection and local UI revision.

func (*PNCounterObserver) Subscribe added in v1.0.36 

func (o *PNCounterObserver) Subscribe(callback Callback[PNCounterView]) (*Subscription[PNCounterView], error)

Subscribe atomically queues the current PN-Counter projection before later counter changes may publish.

func (*PNCounterObserver) SubscribeFromNow added in v1.0.36 

func (o *PNCounterObserver) SubscribeFromNow(callback Callback[PNCounterView]) (*Subscription[PNCounterView], error)

SubscribeFromNow registers a PN-Counter callback without an initial view.

type PNCounterView added in v1.0.36 

type PNCounterView struct {
	Value string
}

PNCounterView is the immutable application projection of a PN-Counter. The decimal representation preserves the full signed range of its uint64 components without exposing a mutable big.Int to callbacks.

type Panic 

type Panic struct {
	Value        any
	EventVersion uint64
	Origin       Origin
}

Panic describes a callback panic captured by the observer dispatcher. Value is the recovered panic value and must be treated as diagnostic data.

type Store 

type Store[T crdt.StateReporter, V any] struct {
	// contains filtered or unexported fields
}

Store owns one application-facing mutation path around a CRDT or another StateReporter. T should not be mutated outside Mutate while observers are active, otherwise Store cannot preserve version/event ordering.

V is an application view such as a counter value, immutable document text, or copied list of set elements. Store does not use V for CRDT merge or wire semantics.

Example 
package main

import (
	"fmt"

	"github.com/DarkInno/crdt/counter"
	"github.com/DarkInno/crdt/observe"
)

func main() {
	value, err := counter.NewGCounter("browser-tab")
	if err != nil {
		panic(err)
	}
	store, err := observe.New(value, func(current *counter.GCounter) uint64 {
		total, err := current.Value()
		if err != nil {
			panic(err)
		}
		return total
	})
	if err != nil {
		panic(err)
	}

	rendered := make(chan uint64, 2)
	subscription, err := store.Subscribe(func(event observe.Event[uint64]) {
		rendered <- event.Value
	})
	if err != nil {
		panic(err)
	}
	<-rendered // Initial view; a UI can render it immediately.
	if err := store.Mutate(observe.Local, func(current *counter.GCounter) error {
		_, err := current.Increment(4)
		return err
	}); err != nil {
		panic(err)
	}

	fmt.Println(<-rendered)
	subscription.Unsubscribe()
	<-subscription.Done()
}

Output:
4

func New 

func New[T crdt.StateReporter, V any](value T, view func(T) V) (*Store[T, V], error)

New creates a Store with an application-owned view function. The view runs only to serve a subscription or publish to at least one subscriber, so a Store without subscribers does not pay projection/copy costs on mutations.

func NewWithOptions 

func NewWithOptions[T crdt.StateReporter, V any](value T, view func(T) V, options Options) (*Store[T, V], error)

NewWithOptions creates a Store with diagnostic callback-panic handling.

func (*Store[T, V]) Close 

func (s *Store[T, V]) Close()

Close prevents future mutations and subscriptions, and cancels all current subscriptions. It does not wait for a callback already in progress; use a Subscription's Done channel when a caller needs to wait for quiescence.

func (*Store[T, V]) Mutate 

func (s *Store[T, V]) Mutate(origin Origin, mutation func(T) error) error

Mutate runs mutation under Store's serialization gate. A nil or failed mutation does not advance Version or notify subscribers. Mutate assumes a successful mutation changed the application-visible state; callers that can determine otherwise should use MutateIf. The wrapped CRDT operation must retain its documented all-or-nothing-on-error behavior.

mutation is intentionally not a notification callback: it may change T and must not recursively call methods on the same Store. Subscribers run only after this method releases the Store lock, so they may safely call Mutate.

func (*Store[T, V]) MutateIf added in v1.0.36 

func (s *Store[T, V]) MutateIf(origin Origin, mutation func(T) (changed bool, err error)) (bool, error)

MutateIf runs mutation under Store's serialization gate and publishes a revision only when mutation reports changed. A nil, failed, or unchanged mutation does not advance Version or notify subscribers. It is suitable for idempotent remote CRDT joins whose caller can determine whether a duplicate delivery extended retained state.

Like Mutate, mutation is not a notification callback and must not recursively call methods on the same Store. Subscribers run only after this method releases the Store lock, so they may safely start a later mutation.

func (*Store[T, V]) Snapshot 

func (s *Store[T, V]) Snapshot() (Event[V], error)

Snapshot returns the current reactive view. Its Origin is Initial because it is a point-in-time read rather than a mutation notification.

func (*Store[T, V]) Subscribe 

func (s *Store[T, V]) Subscribe(callback Callback[V]) (*Subscription[V], error)

Subscribe atomically registers callback and queues the Store's current view before later updates may publish. This prevents a UI from missing a change between reading state and starting observation. If the callback is slow, that initial event may be coalesced into a newer event before delivery.

func (*Store[T, V]) SubscribeFromNow 

func (s *Store[T, V]) SubscribeFromNow(callback Callback[V]) (*Subscription[V], error)

SubscribeFromNow registers callback without an initial state event. It is for consumers that have already obtained a coherent Snapshot themselves.

type Subscription 

type Subscription[V any] struct {
	// contains filtered or unexported fields
}

Subscription owns one callback registration. Unsubscribe is idempotent. Done closes after the subscription goroutine has stopped, including any callback that was already executing at the time of Unsubscribe.

func (*Subscription[V]) Done 

func (s *Subscription[V]) Done() <-chan struct{}

Done is closed after all delivery work for this subscription stops.

func (*Subscription[V]) Panic 

func (s *Subscription[V]) Panic() (Panic, bool)

Panic returns the callback panic, if one caused this subscription to stop.

func (*Subscription[V]) Unsubscribe 

func (s *Subscription[V]) Unsubscribe()

Unsubscribe cancels delivery to this subscription without waiting for an in-progress callback. It is safe to call from the callback itself.

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.