README
¶
[!WARNING] Experimental: lokv is not ready for production use. APIs and the on-disk format may change without notice.
lokv (Log over K/V)
lokv implements an append-only log over a sorted, create-only key/value store (e.g. S3, if so configured). Values are JSON, and every commit is a historical snapshot.
lokv supports event sourcing: use the log as the source of truth and replay
its events to build application state. State[T, S] maintains that state as an
in-memory projection, also called a materialized view.
Each record is an atomic batch: Record[T].Value is a nonempty []T, and
Append(ctx, a, b, c) commits all three values in argument order at one revision.
AppendTo(ctx, snapshot, values...) conditionally commits a whole batch.
import (
"context"
"fmt"
"github.com/tailscale/lokv"
"github.com/tailscale/lokv/memstore"
)
ctx := context.Background()
lg, err := lokv.Open[string](lokv.Config{
Store: new(memstore.Store),
Prefix: "audit",
})
if err != nil { return err }
record, err := lg.Append(ctx, "created", "enabled") // both at revision 1
if err != nil { return err }
fmt.Println(record.Revision)
snap, err := lg.LoadHead(ctx)
if err != nil { return err }
err = lg.Scan(ctx, snap, lokv.All(),
func(r lokv.Record[string]) error {
fmt.Println(r.Revision, r.Value)
return nil
})
if err != nil { return err }
Head returns (record, ok, error). LoadHead returns a nil snapshot for an
empty log, and LoadRevision loads a historical snapshot. Use Scan with
lokv.All() to read the whole snapshot, lokv.StartingAt(rev) to start at a
revision, or lokv.After(applied) to read newer records. Verify checks the
integrity of the records and tree objects reachable from a snapshot.
Event sourcing and projections
For readers familiar with event sourcing, the terminology maps to lokv as follows:
| Term | lokv API |
|---|---|
| Event log | Log[T], used as the application's source of truth |
| Event | One T value; Record[T] contains an atomic batch of events |
| Projection / materialized view | The application value S maintained by State[T, S] |
| Event handler / reducer | The func(*S, T) error callback passed to LoadState |
| Projection position | State.Revision(), the last fully applied batch's revision |
| Optimistic concurrency control | AppendTo commits only if its base snapshot is still current |
In the username registration example, a
userRegistration is an event, userIndex is the projection, and its apply
method is the event handler. LoadState builds the projection by replaying the
log; Sync applies newer events on later calls. AppendTo lets a registration
decision based on that projection commit only if another writer has not advanced
the log. On conflict, the client syncs and recomputes the decision.
Following the log with an in-memory index
LoadState builds a State[T, S] by applying the whole log to your initial
application value. Supply a function that mutates *S for one T value at a
time. State handles iteration in revision order and append argument order within
each batch. For example, count each event in the string log above:
state, err := lokv.LoadState(ctx, lg, make(map[string]int),
func(counts *map[string]int, event string) error {
(*counts)[event]++
return nil
})
if err != nil { return err }
fmt.Println(state.Revision(), state.Value())
// On a poll or wakeup, apply only newly appended batches.
if _, err := state.Sync(ctx); err != nil { return err }
fmt.Println(state.Revision(), state.Value())
Sync returns the snapshot matching the updated state. When a write depends
on the indexes, make the decision from state.Value() and pass that snapshot to
AppendTo. On ErrConflict, catch up and recompute the decision. After a
successful append, state.SyncTo(ctx, newSnapshot) applies the new batch with
no store I/O. It also accepts snapshots obtained with LoadHead or LoadRevision.
It rejects snapshots older than the applied revision and never rewinds state.
The executable username registration example, also
available as the State example in Go documentation, maintains indexes in both
directions between usernames and allocated user IDs. Each client has its own
state. A competing writer may take the proposed name or ID, so a conflict causes
the client to recheck the name and choose the next ID before retrying. Every
writer must follow this protocol for the indexes to remain unique.
The initial value must represent the empty log. State owns and mutates it;
Value returns a shallow copy, with maps and pointers still referring to the
same data. State is not safe for concurrent use. Share it only with caller
synchronization covering catch-up, decisions, and reads of referenced data.
The applied revision advances only after every item in a batch succeeds. Any
apply error permanently poisons the State. Err(), Sync, and SyncTo return
the same sticky error, wrapping the callback error with its revision and item
number. Further syncs perform no I/O or callback calls. The application value
may contain a partially applied batch, including mutations from the failing
call, and must not be used. After addressing the cause, rebuild with LoadState
and a fresh initial value. There is no rollback or reset of a poisoned State.
Store and scan errors remain resumable: LoadState returns the partial State
alongside the error, and if state.Err() is nil, Sync can retry without
reapplying successful batches. Cancellation outside apply takes effect between
batches; a context error returned by apply itself poisons the State. A snapshot
is a fixed upper bound: writes arriving during sync are left for the next call.
The lower-level following example tracks the index and
applied revision directly using LoadHead, Scan, and After. Details of
request counts and range reads are in the
Scan documentation.
Log and Store have no watch API. Call state.Sync from one
goroutine using a ticker, an application-provided wakeup channel, or both:
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for {
if _, err := state.Sync(ctx); err != nil { return err }
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
case _, ok := <-wake: // optional <-chan struct{}; nil disables wakeups
if !ok { wake = nil }
}
}
Treat external notifications, such as S3 notifications for committed log keys,
as hints to catch up. Duplicate or coalesced hints work; a periodic poll covers
missed hints. If other goroutines read the State, hold a mutex across sync and
index access, and check Err() before using the value. For applications managing
their own index with Scan, callbacks receive whole batches; stage each batch's
updates or provide rollback if apply failures need to be retried. When persisting
that index, commit the whole batch's updates and last-applied revision in the same
transaction. Log atomicity does not make application index updates atomic.
Making decisions against an indexed snapshot
AppendTo provides optimistic concurrency control by making a write conditional
on the snapshot used for the decision. For example, after catching up a
name-reservation index, check that a name is free and call AppendTo with that
exact snapshot. On ErrConflict, catch up and check again: another writer may
have reserved the name in the meantime. Append retries the same batch
automatically and cannot recheck application-specific conditions.
The AppendTo example demonstrates that race. Keep the
snapshot returned by a successful AppendTo for the next operation, and apply
its record to your index before using that index for another decision. When a
catch-up scan fails, finish catching up before making decisions against its head.
Storage and concurrency
lokv packs historical events into compressed ranges so clients can load a log without fetching each original record separately. Reads and compaction stream through temporary files to keep memory use proportional to batches and codec buffers; temporary disk use grows with the range being processed.
Log methods are safe for concurrent use. Store implementations must provide
atomic create-if-absent, consistent reads and sorted listing, and immutable
values. The Store contract
documents the requirements; storetest.Test checks additional adapters.
An append error can leave the caller unsure whether the write committed. Applications that need durable deduplication across retries or process restarts should include their own request ID in each event. See the append documentation for retry behavior.
S3
The core package imports no AWS SDK packages. s3store uses AWS SDK for Go v2:
awsCfg, err := config.LoadDefaultConfig(ctx)
if err != nil { return err }
store, err := s3store.New(s3store.Config{
Client: s3.NewFromConfig(awsCfg),
Bucket: "my-general-purpose-bucket",
})
if err != nil { return err }
lg, err := lokv.Open[Event](lokv.Config{Store: store, Prefix: "audit"})
Use a general-purpose S3 bucket configured to prohibit overwrites, deletes, and lifecycle expiration of log objects. The s3store package documentation explains the recommended IAM and bucket configuration and links to the example bucket policy.
Configuration and operation-specific rules are documented in the Go reference. The storage protocol, key layout, compression, and hash definitions are in DESIGN.md.
Local disk caching
cachestore combines an authoritative store, such as the S3 store above, with
a cache store. diskstore provides a persistent local filesystem store:
cache, err := diskstore.New("/var/cache/myapp/audit-bucket")
if err != nil { return err }
cached, err := cachestore.New(cachestore.Config{
Origin: store,
Cache: cache,
})
if err != nil { return err }
lg, err := lokv.Open[Event](lokv.Config{Store: cached, Prefix: "audit"})
Reads use cached immutable objects when available. Complete origin reads and
successful origin writes fill the cache; cache lookup or fill failures do not
fail successful origin operations. List always reaches the origin, so other
writers remain visible. A warm cache can eliminate origin GETs for repeated
scans and idle State.Sync calls, while each sync still makes an origin LIST.
Give each origin its own cache directory. There is no automatic eviction; the cache is disposable and can be removed while clients are stopped. See the cachestore documentation and its executable example for streaming, error handling, and ownership details.
Validation
go test ./...
go test -race ./...
go vet ./...
go test -run '^$' -bench . -benchmem
Tests cover atomic batches, concurrent writers, conflict retries, corruption, stream ownership, and recovery from interrupted operations. Benchmarks measure request counts, compaction costs, and memory use. Parser fuzz targets are in fuzz_test.go.
The live S3 test is opt-in and permanently writes objects under a unique
lokv-integration/ prefix. It never cleans them up. Supply a general-purpose
test bucket, suitable permissions, and standard AWS credentials/region:
LOKV_S3_INTEGRATION=1 LOKV_S3_BUCKET=my-test-bucket \
go test ./s3store -run '^TestLiveS3$' -v
Documentation
¶
Overview ¶
Package lokv implements an append-only log over a sorted, create-only key/value store (e.g. S3, if so configured). The name means Log over K/V. It supports event sourcing: the log serves as the source of truth, and application state is derived by replaying its events.
Use Open to create a Log backed by a Store. Log.Append publishes events as atomic batches, each represented by a Record. Log.LoadHead returns a snapshot of the log, and Log.Scan reads events from that snapshot in order.
LoadState builds an in-memory projection (materialized view) by applying the log to an application-defined value. Its apply callback is the event handler, sometimes called a reducer. State.Sync applies newly appended events and State.Revision tracks the projection's position. Applications arrange their own polling or notifications.
Log.AppendTo provides optimistic concurrency control for decisions based on a snapshot: the append succeeds only if that snapshot is still current. The State example uses this to register unique usernames and allocate user IDs.
Example ¶
package main
import (
"context"
"fmt"
"github.com/tailscale/lokv"
"github.com/tailscale/lokv/memstore"
)
func main() {
ctx := context.Background()
lg, err := lokv.Open[string](lokv.Config{Store: new(memstore.Store), Prefix: "audit"})
if err != nil {
panic(err)
}
// The first two values commit atomically at revision 1, in one object.
snapshot, err := lg.AppendTo(ctx, nil, "created", "updated")
if err != nil {
panic(err)
}
snapshot, err = lg.AppendTo(ctx, snapshot, "archived")
if err != nil {
panic(err)
}
err = lg.Scan(ctx, snapshot, lokv.All(), func(record lokv.Record[string]) error {
fmt.Println(record.Revision, record.Value)
return nil
})
if err != nil {
panic(err)
}
}
Output: 1 [created updated] 2 [archived]
Index ¶
- Constants
- Variables
- type CommitID
- type Config
- type Log
- func (lg *Log[T]) Append(ctx context.Context, value ...T) (Record[T], error)
- func (lg *Log[T]) AppendTo(ctx context.Context, base *Snapshot[T], value ...T) (*Snapshot[T], error)
- func (lg *Log[T]) Head(ctx context.Context) (_ Record[T], ok bool, _ error)
- func (lg *Log[T]) LoadHead(ctx context.Context) (*Snapshot[T], error)
- func (lg *Log[T]) LoadRevision(ctx context.Context, revision int64) (*Snapshot[T], error)
- func (lg *Log[T]) Scan(ctx context.Context, snap *Snapshot[T], r Range, yield func(Record[T]) error) error
- func (lg *Log[T]) Verify(ctx context.Context, snap *Snapshot[T]) error
- type Range
- type Record
- type RecordHash
- type SizeReaderAt
- type Snapshot
- type State
- type Store
Examples ¶
Constants ¶
const MaxRevision int64 = 1<<53 - 1
MaxRevision is the maximum revision and maximum number of batch records in a log. Revisions start at 1; values <= 0 or > MaxRevision are invalid. The limit counts batches, regardless of how many individual events each batch contains. It equals JavaScript's Number.MAX_SAFE_INTEGER (9007199254740991, or 2^53 - 1), so all valid revisions can pass through JavaScript Numbers without losing precision. Appending to a log at MaxRevision returns ErrExhausted. Empty snapshots and states with no applied batches report revision zero.
Variables ¶
var ( ErrNotFound = errors.New("lokv: object not found") ErrExists = errors.New("lokv: object already exists") ErrConflict = errors.New("lokv: append conflict") ErrCorrupt = errors.New("lokv: corrupt log") ErrRange = errors.New("lokv: invalid range") ErrTooLarge = errors.New("lokv: append batch too large") ErrExhausted = errors.New("lokv: revision space exhausted") ErrEmptyBatch = errors.New("lokv: empty batch") )
Functions ¶
This section is empty.
Types ¶
type CommitID ¶
type CommitID [16]byte
CommitID identifies an append invocation. It contains 16 cryptographically random bytes generated once per Append or AppendTo call and reused on retries.
type Config ¶
type Config struct {
// Prefix optionally namespaces the log. It is often empty when an S3 bucket
// is dedicated to a single log. Use distinct prefixes to keep multiple logs
// in the same store. Leading and trailing slashes are stripped.
//
// The normalized prefix must be at most 906 UTF-8 bytes, leaving room for
// every generated key within S3's 1024-byte limit. This applies to all stores.
// It must be valid UTF-8 with no backslashes or Unicode control characters.
// If nonempty, it consists of slash-separated components; no component may
// be empty, ".", or "..". Other characters, including spaces, punctuation,
// and non-ASCII text, are allowed. Matching is literal and case-sensitive;
// no URL decoding, path cleaning, or Unicode normalization is performed.
// Open rejects invalid prefixes before any store I/O.
Prefix string
Store Store
MaxConflictRetries int // Default 32; negative values are invalid. AppendTo never retries conflicts.
// MaxEventBytes limits the complete JSON array of a new append batch.
// Zero defaults to 1 MiB. It does not limit reads or compaction of stored data.
MaxEventBytes int64
// TempDir selects the directory for temporary downloads, validated records,
// and pending compaction uploads. Empty uses the operating system default.
// The directory must already exist when temporary storage is needed. Files
// are unlinked immediately on non-Windows systems and removed on close on
// Windows. Allow space for concurrent operations; see DESIGN.md for costs.
TempDir string
}
Config selects the immutable namespace and resource limits. Zero limits use defaults.
type Log ¶
type Log[T any] struct { // contains filtered or unexported fields }
Log is a concurrency-safe handle to one namespace. Configuration is immutable.
func Open ¶
Open validates configuration without I/O. The caller must ensure Store obeys the strong-consistency, ordering, immutability, and atomic-create contract.
func (*Log[T]) Append ¶
Append atomically appends value as one batch, preserving argument order. It marshals each item once and retries conflicts with the same batch and commit ID. An empty batch returns ErrEmptyBatch without store I/O. MaxEventBytes limits the complete JSON array. An ordinary append creates one object for the batch; a radix carry may create additional aggregate objects for preceding records. The first record has revision 1. Appending after MaxRevision returns ErrExhausted. Transport retries belong to the adapter. An unresolved transport error is returned, since the core cannot classify arbitrary backend errors as retryable. Cancellation or a failed lookup after a create error can leave the caller unsure whether the batch committed. A new invocation uses a new commit ID; durable deduplication across calls or restarts needs an application event ID.
func (*Log[T]) AppendTo ¶
func (lg *Log[T]) AppendTo(ctx context.Context, base *Snapshot[T], value ...T) (*Snapshot[T], error)
AppendTo atomically appends value as one batch only if base is still the current head when the commit is created. A nil base means the log must still be empty; its successor has revision 1. If another writer has advanced the log, AppendTo returns ErrConflict without appending any item. A base at MaxRevision returns ErrExhausted. An empty batch returns ErrEmptyBatch. As with Log.Append, the batch occupies one revision, and MaxEventBytes limits its complete JSON array.
Use AppendTo for optimistic concurrency control when choosing value depends on the log's state. For example, scan a snapshot into an index of reserved names, check that a name is free, then append its reservation against that same snapshot. On ErrConflict, catch up and check again: another writer may have reserved the name. Log.Append automatically retries the same value, so it cannot recheck that decision. State.Sync maintains such an index and returns a snapshot suitable as the base. The State example demonstrates unique username registration and user ID allocation, including recomputing both decisions after a conflict.
AppendTo returns the new snapshot on success. A sequential writer can reuse it as the next base. Reusing a loaded or returned snapshot avoids the List and head Get performed by Append; carries may still read historical objects. Both base and the returned snapshot remain immutable historical views.
Example ¶
package main
import (
"context"
"errors"
"fmt"
"github.com/tailscale/lokv"
"github.com/tailscale/lokv/memstore"
)
func main() {
ctx := context.Background()
lg, err := lokv.Open[string](lokv.Config{Store: new(memstore.Store)})
if err != nil {
panic(err)
}
// Suppose each event reserves a name. Read the current log and decide
// whether "alice" is available against exactly this snapshot.
base, err := lg.LoadHead(ctx)
if err != nil {
panic(err)
}
taken := make(map[string]bool)
err = lg.Scan(ctx, base, lokv.All(), func(r lokv.Record[string]) error {
for _, name := range r.Value {
taken[name] = true
}
return nil
})
if err != nil {
panic(err)
}
if taken["alice"] {
panic("name already taken")
}
// Another writer wins after our read but before our write.
if _, err := lg.Append(ctx, "alice"); err != nil {
panic(err)
}
_, err = lg.AppendTo(ctx, base, "alice", "bob")
fmt.Println("must recheck:", errors.Is(err, lokv.ErrConflict))
// Catch up the index and recheck the decision instead of blindly retrying
// the reservation. For repeated catch-ups, use the Scan follow example.
head, err := lg.LoadHead(ctx)
if err != nil {
panic(err)
}
err = lg.Scan(ctx, head, lokv.After(base.Revision()), func(r lokv.Record[string]) error {
for _, name := range r.Value {
taken[name] = true
}
return nil
})
if err != nil {
panic(err)
}
fmt.Println("name already taken:", taken["alice"])
fmt.Println("bob reserved:", taken["bob"])
fmt.Println("committed reservations:", head.Revision())
}
Output: must recheck: true name already taken: true bob reserved: false committed reservations: 1
func (*Log[T]) Head ¶
Head returns the latest batch. ok reports whether a record was returned. An empty log returns (zero, false, nil).
func (*Log[T]) LoadHead ¶
LoadHead discovers the root using one List with limit 1, followed by one Get for a nonempty namespace. An empty namespace returns nil without a Get. A malformed first key is never skipped.
LoadState builds an in-memory index and State.Sync keeps it current. To manage the applied revision directly, scan the returned snapshot with All. On later polls, use After with the last successfully applied revision to scan through the new snapshot's revision. See the follow example for Log.Scan.
Every nonempty LoadHead call fetches the root, even if it has not changed. Log does not provide notifications; callers arrange polling or wakeups. A caller that already knows a committed revision can use Log.LoadRevision to load that root with one Get and no List.
func (*Log[T]) LoadRevision ¶
LoadRevision loads a historical root without listing. An absent requested revision returns ErrNotFound; missing dependencies encountered later are corrupt. A revision outside [1, MaxRevision] returns ErrRange without store I/O.
For frequent idle polling, probing the last applied revision plus 1 costs one Get. ErrNotFound means no successor was visible at that lookup. Check for MaxRevision before incrementing. If a successor exists, LoadHead can discover the latest root for a batch catch-up, at the cost of an extra probe Get on active polls. A loaded snapshot can be applied with State.SyncTo. See the follow example for Log.Scan for managing the applied revision directly.
func (*Log[T]) Scan ¶
func (lg *Log[T]) Scan(ctx context.Context, snap *Snapshot[T], r Range, yield func(Record[T]) error) error
Scan visits the records in both r and snap in increasing revision order, stopping immediately on callback error. It returns ErrRange for invalid bounds: First and Last must be in [1, MaxRevision], with First <= Last, except that the zero Range is empty. An empty range, a valid range on an empty snapshot, or one starting past the snapshot's revision yields nothing and performs no store I/O. yield must be non-nil.
Use All to scan the whole snapshot, StartingAt to include a revision, or After to resume after an already-applied revision. Scan does not wait for future records. It validates each fetched object, including complete segments that partially intersect the range. Disjoint subtrees are skipped. Each callback receives one entire batch; ranges and revisions count batches, not individual values within them. For N > 0 records, a full scan uses the sum of the hexadecimal digits of N-1 Get calls, in addition to loading the snapshot (at most 75 Gets including the head for up to one million batches, excluding retries).
State manages an application's value and applied revision for catch-ups. The follow example shows how to track them directly: keep the last successfully applied revision and scan only the interval after it. Scan performs no List calls and does not refetch the snapshot's own record: a range containing only that record needs no I/O. Full scans read each frontier object once. For suffixes, Scan estimates payload, metadata, and request costs and may load a segment's historical end commit to find smaller packed ranges or raw commits. For example, catching up from revision 65,535 to 65,537 reads commit 65,536 and uses the loaded head's batch, avoiding the 65,536-record segment. The target snapshot bounds all reads; Scan never discovers a newer head.
Each fetched segment is validated in full before yielding any of its records. Historical roots must match the original range's hash boundaries, and their selected records must form a verified chain to its end before being yielded. A range ending inside a segment uses that segment's packed object. Missing or corrupt objects fail the scan; Scan does not switch paths to hide a failure. The cost estimate cannot know historical batch sizes or compression ratios before reading them and does not promise the minimum possible byte count.
Successfully yielded records are not rolled back if a later read or callback fails. Advance an application's last-applied revision only after the entire batch succeeds, and keep it consistent with the application's index on retries. Applications must provide any atomicity needed for their own index updates.
Example (Follow) ¶
This example maintains an in-memory key/value index by applying log events. The same catchUp function handles the initial load, polling, and wakeups from an application-provided notification source.
package main
import (
"context"
"fmt"
"github.com/tailscale/lokv"
"github.com/tailscale/lokv/memstore"
)
func main() {
type change struct {
Key string
Value string
Delete bool
}
ctx := context.Background()
lg, err := lokv.Open[change](lokv.Config{Store: new(memstore.Store)})
if err != nil {
panic(err)
}
appendChanges := func(value ...change) {
if _, err := lg.Append(ctx, value...); err != nil {
panic(err)
}
}
appendChanges(change{Key: "alice", Value: "reader"}, change{Key: "bob", Value: "reader"})
index := make(map[string]string)
var applied int64 // Zero means no records have been applied yet.
var indexedHead *lokv.Snapshot[change]
catchUp := func() error {
head, err := lg.LoadHead(ctx)
if err != nil {
return err
}
if head.Revision() < applied {
return fmt.Errorf("head moved behind applied revision %d", applied)
}
if head.Revision() > applied {
err = lg.Scan(ctx, head, lokv.After(applied), func(r lokv.Record[change]) error {
// Only this goroutine accesses the index. If readers share it,
// hold their mutex across the entire batch and checkpoint update.
for _, c := range r.Value {
if c.Delete {
delete(index, c.Key)
} else {
index[c.Key] = c.Value
}
}
// Advance only after applying the whole batch. If Scan later fails,
// the next catchUp resumes after the last successfully applied batch.
applied = r.Revision
return nil
})
if err != nil {
return err
}
}
// Only use this snapshot for AppendTo once the index has fully caught up.
indexedHead = head
return nil
}
// Initial LoadHead and Scan read all records into the index. An empty log
// would return a nil head with revision 0, so catchUp would skip Scan.
if err := catchUp(); err != nil {
panic(err)
}
fmt.Println("initial:", applied, index["alice"], index["bob"])
// Another writer appends. In a service, call catchUp from a single goroutine
// on a ticker tick or external wakeup. Neither Log nor Store provides Watch.
appendChanges(change{Key: "alice", Value: "admin"}, change{Key: "bob", Delete: true})
if err := catchUp(); err != nil {
panic(err)
}
// This scan applied only the batch at revision 2; it did not replay revision 1.
fmt.Println("caught up:", applied, index["alice"], len(index))
// A poll with no new entries still uses LoadHead's one List and one Get,
// but skips Scan. A poll with exactly one new batch needs no extra Get for
// Scan: the head already contains that batch. Larger catch-ups choose between
// packed ranges and historical roots with smaller ranges. A short catch-up
// across a large carry need not download the whole compacted history.
if err := catchUp(); err != nil {
panic(err)
}
fmt.Println("unchanged:", indexedHead.Revision(), len(index))
}
Output: initial: 1 reader reader caught up: 2 admin 1 unchanged: 2 1
type Range ¶
type Range struct{ First, Last int64 }
Range is an inclusive interval of revisions. First and Last must both be in [1, MaxRevision], with First <= Last. As a special case, the zero Range is empty. Log.Scan visits the intersection of this range and its snapshot.
func After ¶
After returns the range of revisions strictly greater than revision. After(0) is equivalent to All, and After(MaxRevision) returns an empty range. A revision outside [0, MaxRevision] produces an invalid range, which Log.Scan rejects with ErrRange. See the follow example for Log.Scan for using After to catch up an in-memory index.
type Record ¶
type Record[T any] struct { // Revision is the record's sequence number, from 1 through [MaxRevision]. // Values <= 0 or > MaxRevision are invalid. Revision int64 CommitID CommitID Value []T // Nonempty, in append argument order. RecordHash RecordHash }
Record is an atomically appended batch and its identity. Each successful append creates one record and consumes one revision, regardless of the batch's size.
type RecordHash ¶
type RecordHash [32]byte
RecordHash is a record's logical SHA-256 hash. It covers the record's revision, commit ID, predecessor's hash, and JSON-encoded batch, linking it to its history. Hashes detect corruption; they do not authenticate writers.
type SizeReaderAt ¶
SizeReaderAt is a fixed-size byte source. Size returns its nonnegative length. ReadAt must support concurrent calls, as required by io.ReaderAt.
type Snapshot ¶
type Snapshot[T any] struct { // contains filtered or unexported fields }
Snapshot is a loaded historical root. A nil pointer represents an empty log. Its private wire bytes remain independent of mutations to values returned by Record. Snapshots may be used with the Log that loaded or appended them.
func (*Snapshot[T]) Record ¶
Record returns the root's batch, or a zero record for an empty snapshot. Value and any maps, slices, or pointers within it are shallow copies. Mutating them cannot change the stored batch or future appends/scans.
func (*Snapshot[T]) Revision ¶
Revision returns the root's revision, from 1 through MaxRevision. It returns zero for an empty snapshot; zero is never a valid record revision.
type State ¶
type State[T, S any] struct { // contains filtered or unexported fields }
State maintains an in-memory projection (materialized view) S by applying events from a Log[T]. This derives application state through event sourcing. It tracks the last successfully applied revision, starting at zero for an empty log. Construct a State with LoadState; its zero value is not usable. An apply error permanently poisons the State; see State.Err.
State is mutable and must not be copied or used concurrently. Callers sharing a State must synchronize access to both its methods and the application value, including references returned by State.Value. Catch-up updates the existing value; it does not produce independent historical copies.
Example ¶
package main
import (
"context"
"errors"
"fmt"
"github.com/tailscale/lokv"
"github.com/tailscale/lokv/memstore"
)
type userRegistration struct {
Username string
UserID int64
}
// userIndex maps usernames to IDs and back. Its zero value is ready for use.
type userIndex struct {
ByName map[string]int64
ByID map[int64]string
MaxID int64
}
// apply handles one registration; State handles iteration over batches. Invalid
// stored registrations poison the State, requiring a rebuild after addressing
// the cause. Names are case-sensitive and exact; normalize before registering if
// the application needs different rules.
func (idx *userIndex) apply(reg userRegistration) error {
// This application chooses the same safe-integer limit for user IDs.
if reg.UserID != idx.MaxID+1 || reg.UserID > lokv.MaxRevision {
return fmt.Errorf("invalid user ID %d; want %d", reg.UserID, idx.MaxID+1)
}
if reg.Username == "" {
return errors.New("empty username")
}
if _, exists := idx.ByName[reg.Username]; exists {
return fmt.Errorf("duplicate username %q", reg.Username)
}
if _, exists := idx.ByID[reg.UserID]; exists {
return fmt.Errorf("duplicate user ID %d", reg.UserID)
}
if idx.ByName == nil {
idx.ByName = make(map[string]int64)
}
if idx.ByID == nil {
idx.ByID = make(map[int64]string)
}
idx.ByName[reg.Username] = reg.UserID
idx.ByID[reg.UserID] = reg.Username
idx.MaxID = reg.UserID
return nil
}
var errUsernameTaken = errors.New("username already registered")
// Each client owns a separate State. A service sharing one client across
// goroutines would lock across the entire register operation and index reads.
type userClient struct {
lg *lokv.Log[userRegistration]
state *lokv.State[userRegistration, userIndex]
}
func newUserClient(ctx context.Context, lg *lokv.Log[userRegistration]) (*userClient, error) {
state, err := lokv.LoadState(ctx, lg, userIndex{}, (*userIndex).apply)
if err != nil {
return nil, err
}
return &userClient{lg: lg, state: state}, nil
}
// register allocates the next ID only if the username is still free. All writers
// must follow this protocol; lokv does not enforce application-level constraints.
// A transport or cancellation error may leave a committed registration, just as
// with AppendTo. Durable retries would include an application request ID in the
// event and index it to distinguish a retry from someone else's registration.
func (c *userClient) register(ctx context.Context, username string) (int64, error) {
if username == "" {
return 0, errors.New("empty username")
}
for {
base, err := c.state.Sync(ctx)
if err != nil {
return 0, err
}
idx := c.state.Value()
if _, exists := idx.ByName[username]; exists {
return 0, errUsernameTaken
}
if idx.MaxID >= lokv.MaxRevision {
return 0, errors.New("user ID space exhausted")
}
id := idx.MaxID + 1
head, err := c.lg.AppendTo(ctx, base, userRegistration{Username: username, UserID: id})
if errors.Is(err, lokv.ErrConflict) {
// Another writer may have taken either this name or this ID.
// Catch up, recheck uniqueness, and choose a new ID before retrying.
continue
}
if err != nil {
return 0, err
}
// The new snapshot contains our complete batch, so applying it to the
// local index needs no store reads. Never update indexes before commit.
if err := c.state.SyncTo(ctx, head); err != nil {
return 0, fmt.Errorf("registration committed but local catch-up failed: %w", err)
}
return id, nil
}
}
func main() {
ctx := context.Background()
lg, err := lokv.Open[userRegistration](lokv.Config{Store: new(memstore.Store)})
if err != nil {
panic(err)
}
first, err := newUserClient(ctx, lg)
if err != nil {
panic(err)
}
second, err := newUserClient(ctx, lg)
if err != nil {
panic(err)
}
for _, req := range []struct {
client *userClient
name string
}{{first, "alice"}, {second, "bob"}, {second, "alice"}} {
id, err := req.client.register(ctx, req.name)
fmt.Println(req.name, id, err)
}
// On a ticker tick or an external wakeup, the first client catches up only
// the missing records. This also returns the snapshot for conditional writes.
head, err := first.state.Sync(ctx)
if err != nil {
panic(err)
}
idx := first.state.Value()
fmt.Println("revision:", first.state.Revision(), "head:", head.Revision())
fmt.Println("bob's ID:", idx.ByName["bob"], "user 1:", idx.ByID[1])
// A new client builds both indexes from the whole log in one LoadState call.
third, err := newUserClient(ctx, lg)
if err != nil {
panic(err)
}
fmt.Println("reloaded:", third.state.Revision(), "users:", len(third.state.Value().ByID))
}
Output: alice 1 <nil> bob 2 <nil> alice 0 username already registered revision: 2 head: 2 bob's ID: 2 user 1: alice reloaded: 2 users: 2
func LoadState ¶
func LoadState[T, S any](ctx context.Context, lg *Log[T], initial S, apply func(*S, T) error) (*State[T, S], error)
LoadState loads the current head and applies the whole log to initial, which must represent the application's empty state. It takes ownership of initial, including any maps, slices, or pointers within it; it does not clone them. Both lg and apply must be non-nil.
apply is the event handler, sometimes called a reducer. It receives a pointer to the application value and one T at a time, in revision order and then append argument order within each batch. The applied revision advances only after every item in a batch succeeds. Cancellation is checked between batches, so it does not interrupt a partially applied batch.
Any error returned by apply, including a context error, permanently poisons the State. Updates already made by apply are not rolled back, including those from the failing call. State.Err, State.Sync, and State.SyncTo return the same sticky error, wrapping the apply error with its revision and one-based item number. Discard the poisoned State and rebuild with a fresh initial value after addressing the cause; its application value is no longer valid.
On a load, scan, or apply error, LoadState returns the partially loaded State and the error. If State.Err is nil, State.Sync can resume after the last successfully applied batch. Invalid arguments return a nil State. New records appended after head discovery are left for a later sync. See the username registration example for State.
func (*State[T, S]) Err ¶
Err returns the sticky apply error, or nil if the State has not been poisoned. Once non-nil, it never changes: Sync and SyncTo return it without store I/O or further apply calls. Rebuild with LoadState and a fresh initial value. Store, scan, and context errors originating outside apply do not poison State.
func (*State[T, S]) Revision ¶
Revision returns the projection's position: the last successfully applied batch's revision, or zero if no batch has been applied. A failing apply call does not advance this revision, but leaves Value invalid; see State.Err.
func (*State[T, S]) Sync ¶
Sync loads the current head and applies only batches after State.Revision. It updates only the local state and does not write to the log. Records appended after head discovery are left for a later Sync.
On success it returns the snapshot matching the updated value and revision; the snapshot is nil for an empty log. Use it as the base for Log.AppendTo when deciding what to append depends on the state. On ErrConflict, catch up and recompute the decision. The State example demonstrates username registration.
Polling costs one List and, for a nonempty log, one head Get. An unchanged head or exactly one new batch requires no further I/O. Larger catch-ups have Log.Scan's range costs; use State.SyncTo if a snapshot is already known. Callers arrange their own polling or wakeups; Sync does not wait for writes.
On error it returns a nil snapshot. If State.Err is nil, a later Sync resumes after the last successfully applied batch. Otherwise it returns the sticky apply error without I/O or further apply calls, even if ctx is canceled. Sync never rewinds the state.
func (*State[T, S]) SyncTo ¶
SyncTo applies batches after State.Revision through snap's revision. snap must belong to the State's Log; nil represents an empty log. A snapshot behind the applied revision returns ErrRange without I/O. On success the value and revision match snap, which can then be used with Log.AppendTo.
SyncTo performs no head lookup. Applying a Log.AppendTo result one revision ahead of the state needs no store I/O. Larger catch-ups use Log.Scan's choice of packed ranges and historical roots. On error, a retry resumes after the last successful batch unless apply poisoned the State. If State.Err is non-nil, SyncTo returns it without I/O or further apply calls, even if ctx is canceled or snap is invalid.
func (*State[T, S]) Value ¶
func (s *State[T, S]) Value() S
Value returns the application value after State.Revision. It is a shallow copy: maps, slices, and pointers still refer to the State's data. Callers must not mutate that data outside apply or access it concurrently with catch-up. If State.Err is non-nil, the value may contain a partially applied batch and mutations from the failing apply call; it is invalid and must not be used.
type Store ¶
type Store interface {
// List returns the first at most limit keys beginning with prefix, in
// ascending bytewise order. Prefix matching is literal; an empty prefix
// matches all keys. Returned keys are complete, including the prefix.
// limit must be positive. A successful Create is immediately visible to List.
List(ctx context.Context, prefix string, limit int) ([]string, error)
// Get opens an object, or returns an error matching ErrNotFound.
// The caller must close the reader. ctx governs reads until it is closed.
// Each call returns an independent stream, with errors reported by Read.
Get(ctx context.Context, key string) (io.ReadCloser, error)
// Create atomically publishes a complete value only if key is absent.
// Exactly one concurrent creator wins; losers return ErrExists after the
// winner is visible. Create must never overwrite an existing value.
// It reads exactly value.Size() bytes starting at offset zero, and may reread
// them for retries. It does not close value. A source read error must not
// publish a partial value.
Create(ctx context.Context, key string, value SizeReaderAt) error
}
Store is a sorted, create-only key/value namespace. Implementations must be safe for concurrent use and honor context cancellation.
Create must atomically publish a complete value only if the key is absent, with exactly one winner among concurrent creators. Successful creation must be immediately visible to both Get and List. The caller owns Get's reader and must close it. Create must not retain its input after returning; the caller keeps the source open and unchanged until then.
Values are immutable. The storage authority must prohibit overwrites, deletion, and lifecycle expiration. Open cannot verify these operational preconditions. The s3store subpackage adapts general-purpose S3 buckets; directory buckets are unsupported because their listing is unordered.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
Package cachestore wraps an authoritative lokv.Store with an optional local copy of its immutable objects.
|
Package cachestore wraps an authoritative lokv.Store with an optional local copy of its immutable objects. |
|
Package diskstore provides a filesystem-backed lokv.Store.
|
Package diskstore provides a filesystem-backed lokv.Store. |
|
Package memstore provides a concurrency-safe in-memory lokv.Store for tests and ephemeral logs.
|
Package memstore provides a concurrency-safe in-memory lokv.Store for tests and ephemeral logs. |
|
Package s3store adapts general-purpose Amazon S3 buckets to lokv.Store.
|
Package s3store adapts general-purpose Amazon S3 buckets to lokv.Store. |
|
Package storetest provides the lokv.Store behavioral conformance suite.
|
Package storetest provides the lokv.Store behavioral conformance suite. |