journal

type AdaptiveProbeFunc ¶ added in v0.16.0

type AdaptiveProbeFunc[T any] func(
	ctx context.Context,
	i Interval,
	pos Position,
	rec T,
) (next Position, found bool, err error)

AdaptiveProbeFunc is a function that determines the next probe position for an AdaptiveSearch.

i is the current search bracket — the half-open interval of journal positions still under consideration.

pos and rec are the position and record of the most recent probe.

If found is true, the record at pos is the target and next is ignored.

When found is false, next must be within i and must differ from pos. If either condition is violated, AdaptiveSearch returns a ValueNotFoundError.

type BinaryInterceptor ¶ added in v0.13.0

type BinaryInterceptor = Interceptor[[]byte]

BinaryInterceptor is an Interceptor that can be used to intercept operations on a BinaryJournal.

type BinaryJournal ¶ added in v0.8.0

type BinaryJournal = Journal[[]byte]

A BinaryJournal is an append-only log of binary records.

type BinaryRangeFunc ¶ added in v0.8.0

type BinaryRangeFunc = RangeFunc[[]byte]

A BinaryRangeFunc is a function used to range over the records in a BinaryJournal.

If err is non-nil, ranging stops and err is propagated up the stack. Otherwise, if ok is false, ranging stops without any error being propagated.

type BinaryStore ¶ added in v0.8.0

type BinaryStore = Store[[]byte]

BinaryStore is a Store of journals that contain opaque binary records.

func WithTelemetry(
	s BinaryStore,
	t trace.TracerProvider,
	m metric.MeterProvider,
	l log.LoggerProvider,
) BinaryStore

type CompareFunc ¶ added in v0.3.0

type CompareFunc[T any] func(context.Context, Position, T) (cmp int, err error)

CompareFunc is a function that compares a record to some datum.

If the record is less than the datum, cmp is negative. If the record is greater than the datum, cmp is positive. Otherwise, the record is considered equal to the datum.

type ConflictError ¶ added in v0.10.0

type ConflictError struct {
	Journal  string
	Position Position
}

ConflictError is returned by [Journal.Append] if there is already a record at the specified position.

func (e ConflictError) Error() string

type Interceptor ¶ added in v0.13.0

type Interceptor[T any] struct {
	// contains filtered or unexported fields
}

Interceptor defines functions that are invoked around journal operations.

func (i *Interceptor[T]) AfterAppend(fn func(journal string, rec T) error)

func (i *Interceptor[T]) BeforeAppend(fn func(journal string, rec T) error)

func (i *Interceptor[T]) BeforeOpen(fn func(name string) error)

type Interval ¶ added in v0.10.0

type Interval struct {
	// Begin is the position of the first record in the interval.
	Begin Position

	// End is the position immediately after the last record in the interval.
	End Position
}

Interval describes a half-open interval of positions in a Journal.

func (i Interval) Contains(pos Position) bool

func (i Interval) IsEmpty() bool

func (i Interval) Len() int

func (i Interval) Positions() iter.Seq2[int, Position]

func (i Interval) String() string

type Journal ¶

type Journal[T any] interface {
	// Name returns the name of the journal.
	Name() string

	// Bounds returns the half-open interval [begin, end) describing the
	// positions of the first and last records in the journal.
	Bounds(ctx context.Context) (Interval, error)

	// Get returns the record at the given position.
	//
	// It returns a [RecordNotFoundError] if there is no record at the given
	// position.
	Get(ctx context.Context, pos Position) (rec T, err error)

	// Range invokes fn for each record in the journal, in order, starting with
	// the record at the given position.
	//
	// It returns a [RecordNotFoundError] if there is no record at the given
	// position.
	Range(ctx context.Context, pos Position, fn RangeFunc[T]) error

	// Append adds a record to the journal as the given position.
	//
	// The record is stored at the given position and the end of the journal
	// becomes pos + 1.
	//
	// pos must be the end of the journal, as returned by [Bounds]. If pos < end
	// then a [ConflictError] is returned, indicating that there is already a
	// record at the given position. The behavior is undefined if pos > end.
	Append(ctx context.Context, pos Position, rec T) error

	// Truncate removes journal records in the half-open interval [begin, pos),
	// such that pos becomes the new beginning of the journal.
	//
	// If it returns an error the truncation may have been partially applied.
	//
	// The behavior is undefined if pos > end.
	Truncate(ctx context.Context, pos Position) error

	// Close closes the journal.
	Close() error
}

A Journal is an append-only log containing records of type T.

type Position ¶

type Position uint64

Position is the index of a record within a Journal. The first record is always at position 0.

func AdaptiveSearch[T any](
	ctx context.Context,
	j Journal[T],
	i Interval,
	probe Position,
	fn AdaptiveProbeFunc[T],
) (Position, T, error)

func AppendWithConflictResolution[T any](
	ctx context.Context,
	j Journal[T],
	end Position,
	rec T,
	fn func(context.Context, Position) (Position, error),
) (Position, error)

func FirstRecord[T any](ctx context.Context, j Journal[T]) (Position, T, bool, error)

func LastRecord[T any](ctx context.Context, j Journal[T]) (Position, T, bool, error)

func Search[T any](
	ctx context.Context,
	j Journal[T],
	i Interval,
	cmp CompareFunc[T],
) (pos Position, rec T, err error)

type RangeFunc ¶

type RangeFunc[T any] func(context.Context, Position, T) (ok bool, err error)

A RangeFunc is a function used to range over the records in a Journal.

If err is non-nil, ranging stops and err is propagated up the stack. Otherwise, if ok is false, ranging stops without any error being propagated.

type RecordNotFoundError ¶ added in v0.10.0

type RecordNotFoundError struct {
	Journal  string
	Position Position
}

RecordNotFoundError is returned by [Journal.Get] and [Journal.Range] if the requested record does not exist, either because it has been truncated or because the given position has not been written yet.

func (e RecordNotFoundError) Error() string

type ScanFunc ¶ added in v0.8.0

type ScanFunc[T, V any] func(ctx context.Context, pos Position, rec T) (v V, ok bool, err error)

ScanFunc is a predicate function that produces a value of type V from a record of type T.

If the record cannot be used to produce a value of type V, ok is false.

type Store ¶

type Store[T any] interface {
	// Open returns the journal with the given name.
	Open(ctx context.Context, name string) (Journal[T], error)

	// Provision creates the infrastructure used by the store if it does not
	// already exist.
	Provision(ctx context.Context) error
}

Store is a collection of journals containing records of type T.

func NewMarshalingStore[T any](
	s BinaryStore,
	m marshaler.Marshaler[T],
) Store[T]

func WithInterceptor[T any](s Store[T], in *Interceptor[T]) Store[T]

func WithNameTransform[T any](
	s Store[T],
	x func(string) string,
) Store[T]

type ValueNotFoundError ¶ added in v0.10.0

type ValueNotFoundError struct{}

ValueNotFoundError is returned by search and scan operations if the target value is not found.

func (e ValueNotFoundError) Error() string