Documentation
¶
Overview ¶
Package sqlite is the agent's durable, single-file backend. One SQLite database (pure-Go modernc.org/sqlite, no cgo) backs every persistence domain: the state provider (sessions, skills, memory) and the event spine share one file and one connection, so all durable data lives in one place and a state mutation's event and its projection commit in one transaction.
Writes go through the command path: every mutation is stamped once by the shared state.Stamper (IDs, HLC, versions, the sync envelope, CAS, tombstones), appended to the event spine, and projected into the tables, all inside a single transaction. The live path projects the typed record the Stamper returned (the same post-image the event payload carries, written verbatim from RawPayload); applyEvent performs the identical projection from the payload when Rebuild reprojects the log, so a rebuilt-from-log database is identical to a live one and the event log can never drift from the tables. Both paths share the same row-projection helpers, which keeps full event-sourcing reachable: state is a fold of the spine.
A Store implements state.Provider directly and exposes the event log via Log(). Both pass the shared conformance suites (statetest.RunSuite, spinetest.RunSuite), so this backend stays byte-for-byte interchangeable with the in-memory ones.
Index ¶
- type MerkleNodeStore
- type Option
- type Store
- func (s *Store) ArchiveSealedBlobs(ctx context.Context) (moved int, err error)
- func (s *Store) CheckpointAt(ctx context.Context, stream string, size uint64) (cose []byte, ok bool, err error)
- func (s *Store) Close() error
- func (s *Store) InstanceID() string
- func (s *Store) Jobs() jobs.Queue
- func (s *Store) LatestCheckpoint(ctx context.Context, stream string) (size uint64, cose []byte, ok bool, err error)
- func (s *Store) Log() spine.Log
- func (s *Store) Memory() state.MemoryStore
- func (s *Store) MerkleNodes(stream string) *MerkleNodeStore
- func (s *Store) Name() string
- func (s *Store) PayloadBlobStats(ctx context.Context) (distinct, bytes, refs int64, err error)
- func (s *Store) Rebuild(ctx context.Context) error
- func (s *Store) Resources(reg *resource.Registry) resource.Store
- func (s *Store) SaveCheckpoint(ctx context.Context, stream string, size uint64, cose []byte) error
- func (s *Store) Sessions() state.SessionStore
- func (s *Store) Skills() state.SkillStore
- func (s *Store) SnapshotState(ctx context.Context) error
- func (s *Store) WarmBlobStats(ctx context.Context) (count, rawBytes, packedBytes int64, err error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type MerkleNodeStore ¶
type MerkleNodeStore struct {
// contains filtered or unexported fields
}
MerkleNodeStore is the durable, tiled chain.NodeStore for one stream. It writes nodes into fixed-width tiles: the rightmost partial tile per level is held in memory and written back to SQLite when it fills or on Flush, then evicted, so a long-lived log's resident proof material is bounded by its height rather than its length. Reads of an evicted tile load its blob from SQLite. A MerkleNodeStore is not safe for concurrent use; one belongs to one stream's single-writer append path.
func (*MerkleNodeStore) Flush ¶
func (m *MerkleNodeStore) Flush() error
Flush persists every resident partial tile, so the log's proof material up to the current size is durable. Call it before sealing a checkpoint (and before closing), so a reopen through chain.LoadTree finds every frontier node.
func (*MerkleNodeStore) Node ¶
Node implements chain.NodeStore: the hash at (level, index), from the resident tile if present, else the tile's persisted blob. A missing node reports absent, not an error, so proof assembly names it precisely.
func (*MerkleNodeStore) PutNode ¶
func (m *MerkleNodeStore) PutNode(level uint, index uint64, hash []byte) error
PutNode implements chain.NodeStore: it records the hash at (level, index) into the tile's resident blob, growing the blob's filled prefix. When the write fills the tile it is persisted and evicted from memory; a tile that is not yet full is persisted on Flush. A node is written once, so this never overwrites a different hash.
func (*MerkleNodeStore) Resident ¶
func (m *MerkleNodeStore) Resident() int
Resident reports how many partial tiles are held in memory. It is the O(log n) frontier of the log, not its length; tests assert the bound to guard against a regression that stops evicting full tiles.
type Option ¶
type Option func(*Store)
Option configures the Store.
func WithClock ¶
WithClock sets the time source for record timestamps, event times, and the hybrid logical clock (default: clock.System). Tests and deterministic replay pass a clock.Manual.
func WithIDGenerator ¶
WithIDGenerator sets the source of record IDs (default: a generator on the store's clock with crypto/rand entropy). Supply a generator seeded with a deterministic clock and entropy so a re-run with the same seeds produces the exact same IDs, the basis of deterministic replay.
func WithInstanceID ¶
WithInstanceID sets the origin/last-writer instance stamped onto records this backend creates (default "local"), so fleet/P2P merge can attribute writes.
func WithPayloadBlobThreshold ¶
WithPayloadBlobThreshold sets the stored-byte length at or above which an event payload is externalized into the content-addressed blobs table instead of kept inline. Zero or negative keeps every payload inline. It overrides defaultPayloadBlobThreshold and lets a caller (or a test) force or forgo externalization.
func WithSnapshotCodec ¶
func WithSnapshotCodec(c spine.SnapshotCodec) Option
WithSnapshotCodec makes the resource stream's snapshots verified: Snapshot seals the projection payload through the codec before saving it, and Rebuild opens (verifies) a stored snapshot through it before restoring - one that fails to open is skipped and the stream is folded from the start instead. With a codec set, an unsigned or tampered snapshot is never restored.
func WithSnapshotEvery ¶
WithSnapshotEvery makes the resource store checkpoint itself automatically: after every k successful mutations it writes a snapshot, so a rebuild folds at most k events past the last checkpoint instead of the whole stream. The snapshot is written after the mutation's transaction commits, off the hot write path, and best effort: a snapshot failure never fails the write (a missing snapshot is only slower, never wrong). Zero or negative disables automatic snapshots (the default).
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the SQLite backend. It implements state.Provider (sessions, skills, memory) and exposes the event spine via Log(), all over one database and one connection so cross-domain work shares a single file and transaction.
func Open ¶
Open opens (creating if needed) a SQLite database at dsn and migrates it to the latest schema (the state tables and the event log). dsn is a file path, or ":memory:" for an ephemeral store.
The connection pool is capped at a single connection: SQLite serialises writers anyway, and one connection keeps a ":memory:" database alive with a consistent view. Because every domain shares this one connection, a cross-domain write can be one transaction.
func (*Store) ArchiveSealedBlobs ¶
ArchiveSealedBlobs relocates every hot payload body whose referencing events are all sealed into the warm store, compressed, records the relocation as one governed retention event, and returns how many bodies moved. It is the hot->warm tiering step: after it runs, the hot `blobs` table holds only bodies still referenced by unsealed (recent) events, so the hot file's body footprint tracks the working set instead of all history.
It runs in two phases so the mutation is atomic with its own audit record. First every eligible body is copied to warm (idempotent by content id) and confirmed durable there. Then, in a single write transaction, the confirmed bodies are deleted from hot and a RetentionArchived event is appended to the reserved retention stream naming the count, the bytes reclaimed and gained, and a manifest digest committing to exactly which bodies moved. Because the delete and the event commit together, a reader never sees bodies relocated with no event explaining where they went - tiering is never a silent mutation.
The phase split preserves the crash guarantees of a copy-then-delete move. A crash after phase one leaves the copied bodies in both stores (harmless - the hot read wins) and no event; the retried sweep re-copies (idempotent), re-confirms, and completes phase two. A crash inside phase two rolls the transaction back atomically, so the bodies stay hot and no partial record is written. A body is addressed by its SHA-256 and the envelope commits to that id, so the warm copy rehydrates to the identical bytes and no checkpoint, proof, or event over the run streams changes. Freed hot pages return to SQLite's free list for reuse by later appends (bounding steady-state growth); reclaiming them to the OS is a separate compaction step. A no-op sweep (nothing sealed, or no warm tier) moves nothing and records nothing - only a real relocation is worth an event.
func (*Store) CheckpointAt ¶
func (s *Store) CheckpointAt(ctx context.Context, stream string, size uint64) (cose []byte, ok bool, err error)
CheckpointAt returns the signed tree head stored at exactly size for a stream, or ok=false when none is stored there. It is the earlier head a consistency proof is anchored to when proving the log grew append-only from that point.
func (*Store) Close ¶
Close closes the underlying database (and the read pool of a file database), releasing both the state provider and the event log.
func (*Store) InstanceID ¶
InstanceID is the origin/last-writer instance id this backend stamps onto the records it creates. It identifies this process on the fleet, so a running flynn can register and address its own Instance resource.
func (*Store) Jobs ¶
Jobs returns the durable work queue backed by the same database as the state provider and event spine. Jobs are a plain projected table, not event-sourced: they are operational records mutated in place, so they do not flow through the command path. The returned Queue is valid until the Store is closed.
func (*Store) LatestCheckpoint ¶
func (s *Store) LatestCheckpoint(ctx context.Context, stream string) (size uint64, cose []byte, ok bool, err error)
LatestCheckpoint returns the newest signed tree head for a stream (the greatest size), or ok=false when the stream has none. A reopened log recovers its authenticated length from it and rebuilds its frontier from the tiles at that size.
func (*Store) Log ¶
Log returns the durable event spine backed by the same database, so events and state share one file. The returned Log uses the Store's connections and clock and is valid until the Store is closed.
func (*Store) Memory ¶
func (s *Store) Memory() state.MemoryStore
Memory returns the durable memory store.
func (*Store) MerkleNodes ¶
func (s *Store) MerkleNodes(stream string) *MerkleNodeStore
MerkleNodes returns the durable Merkle node store for a stream, so the stream's verifiable log can page its proof nodes into tiles instead of holding them all in memory. Reopen a persisted log's tree with chain.LoadTree over the returned store.
func (*Store) PayloadBlobStats ¶
PayloadBlobStats reports the content-addressed payload store's accounting: the number of distinct bodies held, their total stored bytes, and the total number of event references to them. Distinct versus referenced is the deduplication ratio - identical bodies are stored once but referenced by every event that carries them - and the byte total is the externalized share a warm or cold tier can relocate independently of the event rows. It reads committed state on the read pool.
func (*Store) Rebuild ¶
Rebuild reprojects the state tables from the event log: it folds the state stream through the same applyEvent the live write path uses, so the projection is reconciled to the log. It resumes from the stream's latest usable snapshot and folds only the events after it, so a rebuild stays bounded as the stream grows; a snapshot that cannot be verified (with a codec set) or decoded is skipped and the whole stream is folded instead - only slower, never wrong. It is idempotent (every record is applied by id), so running it repeatedly is safe.
func (*Store) Resources ¶
Resources returns a durable resource.Store backed by this Store's database, so resource events share the spine, the file, and a transaction with state. The returned store admits writes against reg.
func (*Store) SaveCheckpoint ¶
SaveCheckpoint persists a signed tree head for a stream: the COSE-signed commitment to the Merkle root over the first size events. Every checkpoint is kept, keyed by size, so a verifier can anchor a consistency proof at any earlier head; re-saving the same size replaces its signature.
func (*Store) Sessions ¶
func (s *Store) Sessions() state.SessionStore
Sessions returns the durable conversation store.
func (*Store) Skills ¶
func (s *Store) Skills() state.SkillStore
Skills returns the scoped, FTS5-searchable skill store.
func (*Store) SnapshotState ¶
SnapshotState checkpoints the current state projection (sessions, turns, skills, and memory) onto the event log as a spine.Snapshot on the state stream, anchored at the stream's head Seq, so a Rebuild resumes from it. The payload is sealed by the configured codec, so a stored state snapshot is verified before it is ever restored.
func (*Store) WarmBlobStats ¶
WarmBlobStats reports the warm store's accounting: the number of bodies archived, their original (uncompressed) total bytes, and their compressed total bytes. The ratio of the two is the warm-tier compression report the size gate reads; distinct-versus-referenced dedup is unchanged from the hot tier and reported there.
Source Files