statedb

Documentation

Overview

The statedb package provides an extendable in-memory database built on the go-memdb library which uses immutable radix trees (https://en.wikipedia.org/wiki/Radix_tree) that supports any number of readers without locking but only a single writer at a time.

As this is built around an immutable data structure, any objects stored must never be mutated and a copy must be made prior to modifications.

See pkg/statedb/example for an example how to construct an application that uses this library.

Index

• Variables
• func Collect[Obj any](iter Iterator[Obj]) []Obj
• func Length[Obj any, It Iterator[Obj]](iter It) (n int)
• func NewPrivateTableCell[Obj ObjectConstraints[Obj]](schema *memdb.TableSchema) cell.Cell
• func NewTableCell[Obj ObjectConstraints[Obj]](schema *memdb.TableSchema) cell.Cell
• func ProcessEach[Obj any, It Iterator[Obj]](iter It, fn func(Obj) error) (err error)
• type DB
  • func New(p params) (DB, error)
• type Event
• type IPIndexer
  • func (ii *IPIndexer) FromArgs(args ...interface{}) ([]byte, error)
  • func (ii *IPIndexer) FromObject(obj interface{}) (bool, []byte, error)
  • func (ii *IPIndexer) PrefixFromArgs(args ...interface{}) ([]byte, error)
• type Index
• type Iterator
  • func Filter[Obj any](it Iterator[Obj], keep func(obj Obj) bool) Iterator[Obj]
• type ObjectConstraints
• type Query
  • func ByID(id any) Query
  • func ByRevision(rev uint64) Query
  • func ByUUID(uuid UUID) Query
• type ReadTransaction
• type StringerSliceFieldIndex
  • func (s *StringerSliceFieldIndex) FromArgs(args ...interface{}) ([]byte, error)
  • func (s *StringerSliceFieldIndex) FromObject(obj interface{}) (bool, [][]byte, error)
  • func (s *StringerSliceFieldIndex) PrefixFromArgs(args ...interface{}) ([]byte, error)
• type Table
• type TableName
• type TableReader
• type TableReaderWriter
• type UUID
  • func NewUUID() UUID
• type WatchableIterator
• type WriteTransaction

Variables

var (
	UUIDIndex       = Index("id")
	UUIDIndexSchema = &memdb.IndexSchema{
		Name:         string(UUIDIndex),
		AllowMissing: false,
		Unique:       true,
		Indexer:      &memdb.UUIDFieldIndex{Field: "UUID"},
	}

	RevisionIndex       = Index("revision")
	RevisionIndexSchema = &memdb.IndexSchema{
		Name:         string(RevisionIndex),
		AllowMissing: false,
		Unique:       false,
		Indexer:      &memdb.UintFieldIndex{Field: "Revision"},
	}

	IPIndex  = Index("ip")
	IPSchema = &memdb.IndexSchema{
		Name:         string(IPIndex),
		AllowMissing: false,
		Unique:       false,
		Indexer:      &IPIndexer{Field: "IP"},
	}
)

Common index schemas

var All = Query{"id", []any{}}

All is a query that returns all objects. Order is based on the "id" field index.

var Cell = cell.Module(
	"statedb",
	"In-memory database",

	cell.Provide(New),
)

This module provides an in-memory database built on top of immutable radix trees (courtesy of github.com/hashicorp/go-memdb). It adapts the go-memdb library for use with Hive by taking the table schemas as a group values from hive and provides typed API (Table[Obj]) for manipulating tables. As the database is based on an immutable data structure, all objects inserted into the database MUST NOT be mutated!

For example use see pkg/statedb/example.

Functions

func Collect

func Collect[Obj any](iter Iterator[Obj]) []Obj

Collect collects the object returned by the iterator into a slice.

func Length

func Length[Obj any, It Iterator[Obj]](iter It) (n int)

Length consumes the iterator and returns the number of items consumed.

func NewPrivateTableCell

func NewPrivateTableCell[Obj ObjectConstraints[Obj]](schema *memdb.TableSchema) cell.Cell

NewPrivateTableCell is like NewTableCell, but provides Table[Obj] privately, e.g. only to the module defining it.

func NewTableCell

func NewTableCell[Obj ObjectConstraints[Obj]](schema *memdb.TableSchema) cell.Cell

NewTableCell constructs a new hive cell for a table. Provides Table[Obj] to the application and registers the table's schema with the database.

Example usage:

var beeTableSchema = &memdb.TableSchema{...}
cell.Module(
  "bee-table",
  "Bees!",

  statedb.NewTableCell[*Bee](beeTableSchema), // Provides statedb.Table[*Bee] and register the schema.
  cell.Provide(New)
)
type Bee inteface {
  // some nicer accessors to Table[*Bee]
}
func New(bees state.Table[*Bee]) Bee { ... }

func ProcessEach

func ProcessEach[Obj any, It Iterator[Obj]](iter It, fn func(Obj) error) (err error)

ProcessEach invokes the given function for each object provided by the iterator.

Types

type DB

type DB interface {
	// Observable for observing when tables in the state are changed.
	// It is preferred to use more fine-grained notifications via
	// WatchableIterator when possible.
	//
	// The events may not follow a strict ordering, e.g. if write transactions
	// are performed to table A and then to table B, the observer may see event
	// for table B before table A. Thus these events should only be used as
	// (throttled) triggers to schedule reconciling work.
	stream.Observable[Event]

	// WriteJSON marshals out the whole database as JSON into the given writer.
	WriteJSON(io.Writer) error

	// ReadTxn constructs a new read transaction that can be used to read tables.
	// Reads occur against a snapshot of the database at the time of the call and
	// do not block other readers or writers. A new read transaction is needed to observe
	// new changes to the database.
	ReadTxn() ReadTransaction

	// WriteTxn constructs a new write transaction that can be used
	// to modify tables. Caller must call Commit() or Abort() to release
	// the database write lock.
	WriteTxn() WriteTransaction
}

func New

func New(p params) (DB, error)

type Event

type Event struct {
	Table TableName // The name of the table that changed
}

type IPIndexer

type IPIndexer struct {
	Field string
}

func (*IPIndexer) FromArgs

func (ii *IPIndexer) FromArgs(args ...interface{}) ([]byte, error)

func (*IPIndexer) FromObject

func (ii *IPIndexer) FromObject(obj interface{}) (bool, []byte, error)

func (*IPIndexer) PrefixFromArgs

func (ii *IPIndexer) PrefixFromArgs(args ...interface{}) ([]byte, error)

type Index

type Index string

Index is an opaque type pointing to a specific index on a table. Indexes are defined alongside the table schema.

type Iterator

type Iterator[Obj any] interface {
	// Next returns the next object and true, or zero value and false if iteration
	// has finished.
	Next() (Obj, bool)
}

Iterator for a set of items.

func Filter

func Filter[Obj any](it Iterator[Obj], keep func(obj Obj) bool) Iterator[Obj]

Filter wraps an iterator that only returns the objects for which 'keep' returns true.

type ObjectConstraints

type ObjectConstraints[Obj any] interface {
	DeepCopy() Obj
}

ObjectConstraints specifies the constraints that an object must fulfill for it to be stored in a table.

type Query

type Query struct {
	Index Index // The table index to query against
	Args  []any // The query argument(s).
}

Query against a table using a specific index and argument(s). Queries should be predefined with strong typing alongside the table schema definition.

func ByID

func ByID(id any) Query

ByID queries by the "ID" field. The type of "ID" is table specific, thus this function takes an 'any'.

func ByRevision

func ByRevision(rev uint64) Query

ByRevision queries the table by revision. The target table must include the RevisionIndexSchema.

func ByUUID

func ByUUID(uuid UUID) Query

ByUUID queries the table by UUID.

type ReadTransaction

type ReadTransaction interface {
	// contains filtered or unexported methods
}

ReadTransaction can be used to read data in tables. It provides a consistent snapshot of the database across all tables.

type StringerSliceFieldIndex

type StringerSliceFieldIndex struct {
	Field     string
	Lowercase bool
}

StringerSliceFieldIndex builds an index from a field on an object that is a slice of objects that implement fmt.Stringer. Each value within the string slice can be used for lookup.

func (*StringerSliceFieldIndex) FromArgs

func (s *StringerSliceFieldIndex) FromArgs(args ...interface{}) ([]byte, error)

func (*StringerSliceFieldIndex) FromObject

func (s *StringerSliceFieldIndex) FromObject(obj interface{}) (bool, [][]byte, error)

func (*StringerSliceFieldIndex) PrefixFromArgs

func (s *StringerSliceFieldIndex) PrefixFromArgs(args ...interface{}) ([]byte, error)

type Table

type Table[Obj ObjectConstraints[Obj]] interface {
	Name() TableName

	// Reader when given a read transaction returns a table reader
	// that can be used to read from the snapshot of the database.
	Reader(tx ReadTransaction) TableReader[Obj]

	// Writer when given a write transaction returns a table writer
	// that can be used to modify the table.
	Writer(tx WriteTransaction) TableReaderWriter[Obj]
}

Table provides read and write access to a specific table.

type TableName

type TableName string

TableName is an opaque type carrying the name a table. Returned by Table[T].Name().

type TableReader

type TableReader[Obj ObjectConstraints[Obj]] interface {
	// First returns the first matching object with the given query. Returned
	// object is nil if the object does not exist. Error is non-nil if the query
	// is malformed (e.g. unknown index).
	First(Query) (Obj, error)

	// Last returns the last matching object with the given query. Returned
	// object is nil if the object does not exist. Error is non-nil if the query
	// is malformed (e.g. unknown index).
	Last(Query) (Obj, error)

	// Get returns all objects matching the given query as a WatchableIterator
	// that allows iterating over the set of matching objects and to watch whether
	// the query has been invalidated by changes to the database. Returns
	// an error if the query is malformed.
	Get(Query) (WatchableIterator[Obj], error)

	// LowerBound returns objects that are equal to or higher than the query. The
	// comparison is performed against byte slices derived from the index argument(s).
	LowerBound(Query) (Iterator[Obj], error)
}

TableReader provides a set of read-only queries to a table.

It is encouraged to wrap these methods behind a table-specific API as these methods may fail if query is badly formed. E.g. wrap Get(ByName(...) into GetByName that constructs the query and panics on errors (since those are indication of the method or table schema being broken).

Objects returned by these methods are considered immutable and MUST never be mutated by the caller! To modify an object for insertion, it MUST be DeepCopy()'d first.

type TableReaderWriter

type TableReaderWriter[Obj ObjectConstraints[Obj]] interface {
	TableReader[Obj]

	// Insert an object into the table. May return an error if indexing fails.
	Insert(obj Obj) error

	// Delete an object from the table. May return an error if the "id" index key
	// cannot be computed.
	Delete(obj Obj) error

	// DeleteAll deletes all matching objects from the table. May return an error
	// on indexing issues.
	DeleteAll(Query) (n int, err error)
}

TableReaderWriter provides methods to modify a table.

It is encouraged to wrap these methods behind a safer table-specific API as these expose errors related to malformed indices that the user of the table should not need to handle.

type UUID

type UUID = string

func NewUUID

func NewUUID() UUID

type WatchableIterator

type WatchableIterator[Obj any] interface {
	Iterator[Obj]

	// Invalidated returns a channel that is closed when the results
	// returned by the iterator have changed in the database.
	Invalidated() <-chan struct{}
}

WatchableIterator is an Iterator that provides notification when the iterator results have been invalidated by a change in the database.

type WriteTransaction

type WriteTransaction interface {

	// Abort the transaction and throw away the changes.
	Abort()

	// Commit the transaction to the database. May fail if a commit hook
	// fails. On failure the changes are discarded and caller should retry
	// at a later point in time.
	Commit() error

	// Revision of the database. Revision is a simple counter of committed
	// transactions and can be used within the objects for detecting which
	// objects has changed.
	Revision() uint64

	// Defer registers a function to run after the transaction has been
	// successfully committed.
	Defer(fn func())
	// contains filtered or unexported methods
}

WriteTransaction can be used with one more 'Table's to make a set of atomic changes to them.