Documentation
¶
Overview ¶
Package kv stores record streams in a clicky cache.Store, so one implementation runs in process (cache.NewMemory) and against valkey/redis (clicky/valkey.NewStore) — the backend a CLI writes to and a server reads.
A stream is keys under <prefix>/<stream>/, laid out by its kind. An unkeyed kind's rows arrive in bulk and are stored in chunks (chunks.go); a keyed kind's rows are stored one key each (rows.go), so what an append asks about a row is one read of that row, whatever else the stream holds. Both layouts share:
meta the stream's recordstore.Meta as JSON
The metadata is the commit point: its low and high seq bound everything a reader or a key lookup believes, so an entry an interrupted write left outside them is invisible, and the next write removes it.
Index ¶
- type Backend
- func (b *Backend) Append(ctx context.Context, stream, kind string, rows []recordstore.Row) (recordstore.AppendResult, error)
- func (b *Backend) Close() error
- func (b *Backend) Expire(ctx context.Context, stream string, ttl time.Duration) error
- func (b *Backend) Meta(ctx context.Context, stream string) (recordstore.Meta, error)
- func (b *Backend) Scan(ctx context.Context, stream string, afterSeq int64, ...) error
- func (b *Backend) Trim(ctx context.Context, stream string, before time.Time) (recordstore.Meta, error)
- type Options
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Backend ¶
type Backend struct {
// contains filtered or unexported fields
}
Backend is a recordstore.Backend over a cache.Store.
func (*Backend) Append ¶
func (b *Backend) Append(ctx context.Context, stream, kind string, rows []recordstore.Row) (recordstore.AppendResult, error)
Append writes the rows it keeps ahead of the metadata that makes them visible. A reader bounds itself by the metadata's high seq, so a crash part way leaves entries past it that no reader sees; the next append removes them before numbering its own rows over the same seqs.
func (*Backend) Scan ¶
func (b *Backend) Scan(ctx context.Context, stream string, afterSeq int64, fn func(int64, recordstore.Row) error) error
Scan reads stream's rows in seq order, from its low seq through its high seq. The store must hold every seq between them: one that lost a row, a chunk or an index under memory pressure reads as an error, never as a shorter stream.
type Options ¶
type Options struct {
// Store holds the keys. It is owned by the caller and not closed.
Store cache.Store
// Prefix namespaces every key the backend writes.
Prefix string
// Schema resolves a kind to its key and retention. A kind it refuses
// cannot be written or read.
Schema recordstore.SchemaResolver
// TTL is how long a stream lives from its first append unless Expire moves
// it, or how long a row of a kind retaining rows lives from its own append.
// It is required: the store is memory, and a stream nobody expires is a
// leak.
TTL time.Duration
// MaxChunkBytes caps one stored chunk, or one stored keyed row. An unkeyed
// append is split into as many chunks as it needs; a single row larger than
// the cap is refused with recordstore.ErrCapacity.
MaxChunkBytes int
// Now is the clock stream metadata is stamped with. Nil is time.Now.
Now func() time.Time
}
Options configure a kv backend.
Source Files