ratel

Documentation

Index

• Constants
• func GetIndexKeysForEvent(ev *event.T, ser *serial.T) (keyz [][]byte)
• func GetTagKeyElements(tagValue string, CA *createdat.T, ser *serial.T) (prf index.P, elems []keys.Element, err error)
• func GetTagKeyPrefix(tagValue string) (key []byte, err error)
• func NewLogger(logLevel int, label string) (l *logger)
• func PrepareQueries(f *filter.T) (qs []query, ext *filter.T, since uint64, err error)
• type B
  • func GetCounterKey(ser *serial.T) (key B)
• type Ctx
• type DelItems
• type E
• type N
• type Results
• type S
• type T
  • func GetBackend(Ctx context.T, WG *sync.WaitGroup, hasL2 bool, ...) (b *T)
  • func (r *T) Close() (err E)
  • func (r *T) CountEvents(c Ctx, f *filter.T) (count N, approx bool, err E)
  • func (r *T) DeleteEvent(c Ctx, eid *eventid.T) (err E)
  • func (r *T) Export(c context.T, w io.Writer)
  • func (r *T) GCCount() (unpruned, pruned count.Items, unprunedTotal, prunedTotal int, err error)
  • func (r *T) GCMark() (pruneEvents, pruneIndexes DelItems, err error)
  • func (r *T) GCRun() (err error)
  • func (r *T) GCSweep(evs, idxs DelItems) (err error)
  • func (r *T) GarbageCollector()
  • func (r *T) GetEventHeadroom() (hw, lw int)
  • func (r *T) GetIndexHeadroom() (hw, lw int)
  • func (r *T) Import(rr io.Reader)
  • func (r *T) Init(path S) (err E)
  • func (r *T) Nuke() (err E)
  • func (r *T) Path() S
  • func (r *T) QueryEvents(c Ctx, f *filter.T) (evs []*event.T, err E)
  • func (r *T) SaveEvent(c Ctx, ev *event.T) (err E)
  • func (r *T) Serial() (ser uint64, err error)
  • func (r *T) SerialBytes() (ser []byte, err error)
  • func (r *T) SerialKey() (idx []byte, ser *serial.T)
  • func (r *T) Sync() (err E)

Constants

const CounterLen = KeyLen + createdat.Len

const DefaultMaxLimit = 512

const KeyLen = serial.Len + 1

const PrunedLen = sha256.Size + KeyLen

const Version = 1

Functions

func GetIndexKeysForEvent

func GetIndexKeysForEvent(ev *event.T, ser *serial.T) (keyz [][]byte)

GetIndexKeysForEvent generates all the index keys required to filter for events. evtSerial should be the output of Serial() which gets a unique, monotonic counter value for each new event.

func GetTagKeyElements

func GetTagKeyElements(tagValue string, CA *createdat.T,
	ser *serial.T) (prf index.P,
	elems []keys.Element, err error)

func GetTagKeyPrefix

func GetTagKeyPrefix(tagValue string) (key []byte, err error)

GetTagKeyPrefix returns tag index prefixes based on the initial field of a tag.

There is 3 types of index tag keys:

- TagAddr: [ 8 ][ 2b Kind ][ 8b Pubkey ][ address/URL ][ 8b Serial ]

- Tag32: [ 7 ][ 8b Pubkey ][ 8b Serial ]

- Tag: [ 6 ][ address/URL ][ 8b Serial ]

This function produces the initial bytes without the index.

func NewLogger

func NewLogger(logLevel int, label string) (l *logger)

func PrepareQueries

func PrepareQueries(f *filter.T) (
	qs []query,
	ext *filter.T,
	since uint64,
	err error,
)

PrepareQueries analyses a filter and generates a set of query specs that produce key prefixes to search for in the badger key indexes.

Types

type B

type B = []byte

func GetCounterKey

func GetCounterKey(ser *serial.T) (key B)

GetCounterKey returns the proper counter key for a given event ID.

type Ctx

type Ctx = context.T

type DelItems

type DelItems []uint64

type E

type E = error

type N

type N = int

type Results

type Results struct {
	Ev  *event.T
	TS  *timestamp.T
	Ser *serial.T
}

type S

type S = string

type T

type T struct {
	Ctx context.T
	WG  *sync.WaitGroup

	// DBSizeLimit is the number of bytes we want to keep the data store from exceeding.
	DBSizeLimit int
	// DBLowWater is the percentage of DBSizeLimit a GC run will reduce the used storage down
	// to.
	DBLowWater int
	// DBHighWater is the trigger point at which a GC run should start if exceeded.
	DBHighWater int
	// GCFrequency is the frequency of checks of the current utilisation.
	GCFrequency    time.Duration
	HasL2          bool
	BlockCacheSize int
	InitLogLevel   int
	Logger         *logger
	// DB is the badger db enveloper
	*badger.DB

	// Threads is how many CPU threads we dedicate to concurrent actions, flatten and GC mark
	Threads int
	// MaxLimit is a default limit that applies to a query without a limit, to avoid sending out
	// too many events to a client from a malformed or excessively broad filter.
	MaxLimit int
	// ActuallyDelete sets whether we actually delete or rewrite deleted entries with a modified
	// deleted prefix value (8th bit set)
	ActuallyDelete bool
	// Flatten should be set to true to trigger a flatten at close... this is mainly
	// triggered by running an import
	Flatten bool
	// contains filtered or unexported fields
}

func GetBackend

func GetBackend(Ctx context.T, WG *sync.WaitGroup, hasL2 bool,
	blockCacheSize, logLevel, maxLimit int, params ...int) (b *T)

GetBackend returns a reasonably configured badger.Backend.

The variadic params correspond to DBSizeLimit, DBLowWater, DBHighWater and GCFrequency as an integer multiplier of number of seconds.

Note that the cancel function for the context needs to be managed by the caller.

func (*T) Close

func (r *T) Close() (err E)

func (*T) CountEvents

func (r *T) CountEvents(c Ctx, f *filter.T) (count N, approx bool, err E)

func (*T) DeleteEvent

func (r *T) DeleteEvent(c Ctx, eid *eventid.T) (err E)

func (*T) Export

func (r *T) Export(c context.T, w io.Writer)

func (*T) GCCount

func (r *T) GCCount() (unpruned, pruned count.Items, unprunedTotal,
	prunedTotal int, err error)

GCCount performs a census of events in the event store. It counts the number of events and their size, and if there is a layer 2 enabled, it counts the number of events that have been pruned and thus have indexes to count.

Both operations are more efficient combined together rather than separated, thus this is a fairly long function.

func (*T) GCMark

func (r *T) GCMark() (pruneEvents, pruneIndexes DelItems, err error)

GCMark first gathers the serial, data size and last accessed information about all events and pruned events using GCCount then sorts the results of the events and indexes by least recently accessed and generates the set of serials of events that need to be deleted

func (*T) GCRun

func (r *T) GCRun() (err error)

func (*T) GCSweep

func (r *T) GCSweep(evs, idxs DelItems) (err error)

func (*T) GarbageCollector

func (r *T) GarbageCollector()

GarbageCollector starts up a ticker that runs a check on space utilisation and when it exceeds the high-water mark, prunes back to the low-water mark.

This function should be invoked as a goroutine, and will terminate when the backend context is canceled.

func (*T) GetEventHeadroom

func (r *T) GetEventHeadroom() (hw, lw int)

func (*T) GetIndexHeadroom

func (r *T) GetIndexHeadroom() (hw, lw int)

func (*T) Import

func (r *T) Import(rr io.Reader)

Import accepts an event

func (*T) Init

func (r *T) Init(path S) (err E)

func (*T) Nuke

func (r *T) Nuke() (err E)

func (*T) Path

func (r *T) Path() S

func (*T) QueryEvents

func (r *T) QueryEvents(c Ctx, f *filter.T) (evs []*event.T, err E)

func (*T) SaveEvent

func (r *T) SaveEvent(c Ctx, ev *event.T) (err E)

func (*T) Serial

func (r *T) Serial() (ser uint64, err error)

Serial returns the next monotonic conflict free unique serial on the database.

func (*T) SerialBytes

func (r *T) SerialBytes() (ser []byte, err error)

SerialBytes returns a new serial value, used to store an event record with a conflict-free unique code (it is a monotonic, atomic, ascending counter).

func (*T) SerialKey

func (r *T) SerialKey() (idx []byte, ser *serial.T)

SerialKey returns a key used for storing events, and the raw serial counter bytes to copy into index keys.

func (*T) Sync

func (r *T) Sync() (err E)