journal

Documentation

Overview

Package journal provides an abstraction of an append-only log with optimistic concurrency control.

Index

• Variables
• func IsEmpty[T any](ctx context.Context, j Journal[T]) (bool, error)
• func IsFresh[T any](ctx context.Context, j Journal[T]) (bool, error)
• func RunBenchmarks(b *testing.B, newStore func(b *testing.B) BinaryStore)
• func RunTests(t *testing.T, newStore func(t *testing.T) BinaryStore)
• func Scan[T, V any](ctx context.Context, j Journal[T], begin Position, scan ScanFunc[T, V]) (V, error)
• func ScanFromSearchResult[T, V any](ctx context.Context, j Journal[T], begin, end Position, cmp CompareFunc[T], ...) (V, error)
• type BinaryJournal
• type BinaryRangeFunc
• type BinaryStore
  • func WithTelemetry(s BinaryStore, p trace.TracerProvider, m metric.MeterProvider, l *slog.Logger) BinaryStore
• type CompareFunc
• type Journal
• type Position
  • 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], begin, end Position, cmp CompareFunc[T]) (pos Position, rec T, err error)
• type RangeFunc
• type ScanFunc
• type Store
  • func NewMarshalingStore[T any](s BinaryStore, m marshaler.Marshaler[T]) Store[T]

Variables

var (
	// ErrNotFound 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.
	ErrNotFound = errors.New("record not found")

	// ErrConflict is returned by [Journal.Append] if there is already a record at
	// the specified position.
	ErrConflict = errors.New("optimistic concurrency conflict")
)

Functions

func IsEmpty

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

IsEmpty returns true if j does not currently contain any records.

func IsFresh

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

IsFresh returns true if j has never contained any records.

func RunBenchmarks

func RunBenchmarks(
	b *testing.B,
	newStore func(b *testing.B) BinaryStore,
)

RunBenchmarks runs benchmarks against a BinaryStore implementation.

func RunTests

func RunTests(
	t *testing.T,
	newStore func(t *testing.T) BinaryStore,
)

RunTests runs tests that confirm a journal implementation behaves correctly.

func Scan

func Scan[T, V any](
	ctx context.Context,
	j Journal[T],
	begin Position,
	scan ScanFunc[T, V],
) (V, error)

Scan finds a value of type V within the journal by scanning all records beginning with the record at the given position.

It returns ErrNotFound if the value is not found.

This function is useful when the value being searched is not ordered, or when there are a small number of records to scan. If the records are structured in such a way that it's possible to know if the value appears before or after a specific record, use Search instead.

func ScanFromSearchResult

func ScanFromSearchResult[T, V any](
	ctx context.Context,
	j Journal[T],
	begin, end Position,
	cmp CompareFunc[T],
	scan ScanFunc[T, V],
) (V, error)

ScanFromSearchResult finds a value within the journal by scanning all records beginning with the record for which a binary search of the half-open range [begin, end) using cmp as the comparator returns true. See Scan and Search.

It returns ErrNotFound if the value is not found.

Types

type BinaryJournal

type BinaryJournal = Journal[[]byte]

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

type BinaryRangeFunc

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

type BinaryStore = Store[[]byte]

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

func WithTelemetry

func WithTelemetry(
	s BinaryStore,
	p trace.TracerProvider,
	m metric.MeterProvider,
	l *slog.Logger,
) BinaryStore

WithTelemetry returns a BinaryStore that adds telemetry to s.

type CompareFunc

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 Journal

type Journal[T any] interface {
	// Bounds returns the half-open range [begin, end) describing the positions
	// of the first and last journal records that are available for reading.
	Bounds(ctx context.Context) (begin, end Position, err error)

	// Get returns the record at the given position.
	//
	// It returns [ErrNotFound] 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 [ErrNotFound] 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.
	//
	// end must be the next "unused" position in the journal; the first position
	// is always 0.
	//
	// If there is already a record at the given position then [ErrConflict] is
	// returned, indicating an optimistic concurrency conflict.
	//
	// The behavior is undefined if end is greater than the next position.
	Append(ctx context.Context, end Position, rec T) error

	// Truncate removes journal records in the half-open range [0, end). That
	// is, it removes the oldest records up to, but not including, the record at
	// the given position.
	//
	// If it returns a non-nil error the truncation may have been partially
	// applied. That is, some of the records may have been removed but not all.
	// The implementation must guarantee that the oldest records are removed
	// first, such that there is never a "gap" between positions.
	//
	// The behavior is undefined if end is greater than the next position.
	Truncate(ctx context.Context, end 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 FirstRecord

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

FirstRecord returns the oldest record in a journal.

func LastRecord

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

LastRecord returns the newest record in a journal.

func Search

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

Search performs a binary search of j within the half-open range [begin, end) to find the position of the record for which cmp() returns zero.

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 ScanFunc

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(ctx context.Context, name string) (Journal[T], error)
}

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

func NewMarshalingStore

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

NewMarshalingStore returns a new Store that marshals/unmarshals records of type T to/from an underlying BinaryStore.