Documentation
¶
Overview ¶
Package packstore persists Amber-Store CAS objects in log-structured, append-only segment (pack) files. The store directory contains only segment files: sealed segments are immutable, mmap'd whole, and self-indexed by a footer (fanout index on the last key byte + binary fuse filter + fixed trailer); the single active segment is recovered by a tail-scan. There is no global index. All format integers are big-endian. Record framing lives in the amberpack package. See docs/superpowers/specs/2026-06-13-packstore-design.md.
Index ¶
- Constants
- Variables
- type CompactOpts
- type CompactStats
- type MarkSet
- type Object
- type Option
- type SegmentInfo
- type SegmentLiveness
- type Store
- func (s *Store) AbortBarrier()
- func (s *Store) AppendRecord(k key.Key, raw []byte) error
- func (s *Store) BeginBarrier()
- func (s *Store) Close() error
- func (s *Store) Compact(live func(key.Key) bool, opts CompactOpts) (CompactStats, error)
- func (s *Store) Get(k key.Key) ([]byte, error)
- func (s *Store) GetRecord(k key.Key) ([]byte, error)
- func (s *Store) Has(k key.Key) (bool, error)
- func (s *Store) HasOutside(id uint64, k key.Key) (bool, error)
- func (s *Store) Liveness(live func(key.Key) bool) ([]SegmentLiveness, error)
- func (s *Store) Missing(keys []key.Key) ([]key.Key, error)
- func (s *Store) NewMarkSet() *MarkSet
- func (s *Store) ObserveKeys(keys []key.Key)
- func (s *Store) OldestInflightWrite() (time.Time, bool)
- func (s *Store) Put(k key.Key, data []byte) error
- func (s *Store) PutVerified(k key.Key, data []byte) error
- func (s *Store) Record(id, off uint64) ([]byte, error)
- func (s *Store) Remove(id uint64) error
- func (s *Store) ScanIndex(id uint64, fn func(k key.Key, off uint64, slen uint32)) error
- func (s *Store) Segments() ([]SegmentInfo, error)
- func (s *Store) SortByLocation(keys []key.Key)
- func (s *Store) StoredSize(k key.Key) (uint64, bool, error)
- func (s *Store) Sync() error
- func (s *Store) Verify(ctx context.Context) error
- func (s *Store) Wipe() error
- func (s *Store) WriteBatch(seq iter.Seq2[Object, error]) error
- func (s *Store) WriteParallel(seq iter.Seq2[Object, error], opts WriteOpts) (WriteStats, error)
- type WriteOpts
- type WriteStats
Constants ¶
const DefaultBatchSize = 16 << 20 // 16 MiB
DefaultBatchSize is the byte threshold at which a writer fsyncs the active segment, making everything appended so far durable.
const DefaultSegmentSize = 2 << 30 // 2 GiB
DefaultSegmentSize is the default rotation threshold: the active segment is sealed once it reaches this many bytes.
Variables ¶
var ErrClosed = errors.New("packstore: store closed")
ErrClosed is returned by operations on a closed store.
var ErrCorrupt = amberpack.ErrCorrupt
ErrCorrupt wraps every structural-corruption error (bad record framing, bad footer, scrub findings). It aliases amberpack's record-corruption sentinel so a single errors.Is target covers both record- and footer-level corruption.
var ErrNotFound = errors.New("packstore: object not found")
ErrNotFound is returned by Get for a key that is not present in the store.
var ErrUnknownSegment = errors.New("packstore: no such segment")
ErrUnknownSegment reports an id that names no sealed segment (never sealed, or already removed).
var ErrVerify = errors.New("packstore: object verification failed")
ErrVerify is returned (wrapped) when an object's key does not match its payload. Callers distinguish it with errors.Is to map to a client error.
Functions ¶
This section is empty.
Types ¶
type CompactOpts ¶
type CompactOpts struct {
// MinDeadRatio is the selection line: a sealed segment whose dead
// bytes reach this fraction of its record bytes is rewritten.
MinDeadRatio float64
// Horizon, when nonzero, excludes segments sealed at or after it —
// the gc grace period. The zero value means every sealed segment is
// eligible.
Horizon time.Time
// Pace, when non-nil, is called with each copied record's size; the
// collector uses it to cap copy bandwidth.
Pace func(n int)
}
CompactOpts tune one Compact pass.
type CompactStats ¶
type MarkSet ¶
type MarkSet struct {
// contains filtered or unexported fields
}
MarkSet is a liveness mark over a snapshot: one bit per sealed record, slotted by footer index, plus a map for the active segment. Not concurrency-safe. Run it under the writer quiesce GC requires. Snapshots own their footer indexes, using 44 additional bytes per sealed key. This keeps lookups valid after a repair replaces and unmaps a segment.
type Object ¶
Object is one CAS object to store: its key and either its serialized bytes (Data) or, for an object that was encoded elsewhere, the complete record as amberpack.EncodeRecord produced it (Record). Exactly one of the two is set. A Record is parsed (framing, CRC, canonical key, key equal to Key) and appended verbatim, so a caller that already holds encoded records, say a pack it staged on disk, skips the compression round trip; with WriteOpts.Verify its payload is decoded and rehashed like Data is.
type Option ¶
type Option func(*config)
Option configures a Store at Open time.
func WithSegmentSize ¶
WithSegmentSize sets the rotation threshold in bytes. A single oversized record may push one segment past it.
type SegmentInfo ¶
type SegmentInfo struct {
ID uint64
Sealed time.Time // .seg mtime: the footer write is the file's last write
Body int64 // record bytes: body length minus the header magic
Keys uint64
}
SegmentInfo describes one sealed segment.
type SegmentLiveness ¶
type SegmentLiveness struct {
ID uint64
Sealed bool
LiveKeys int
DeadKeys int
LiveBytes uint64
DeadBytes uint64
}
SegmentLiveness is one segment's liveness breakdown. Byte counts cover record bytes only, not footer overhead.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is an on-disk content-addressable store over segment files. It is safe for concurrent use. Lock ordering: appendMu before mu, never the reverse. appendMu serializes the write path (append, fsync, seal, Close); mu guards sealed/active/closed for readers.
func Open ¶
Open opens (creating if necessary) a store rooted at dir. Only one Store may have a given dir open at a time (flock on the directory). Sealed segments are mmap'd and validated; the active segment, if any, is tail-scanned and truncated to its last valid record.
func (*Store) AbortBarrier ¶
func (s *Store) AbortBarrier()
AbortBarrier discards a capture without compacting.
func (*Store) AppendRecord ¶
AppendRecord re-appends an already-encoded record through the normal append path: no decode, no re-encode, no fsync — callers batch appends and call Sync. raw must be exactly one record, CRC-valid, keyed k. A nil raw is rejected outright: prepare would read it as an Object carrying Data instead of a Record and happily append an empty payload under k.
func (*Store) BeginBarrier ¶
func (s *Store) BeginBarrier()
BeginBarrier starts grey capture: keys a write observes are live for the next Compact, letting ingests run concurrently with the caller's mark (specs/gc.qnt). Compact itself must still not overlap ingests.
func (*Store) Close ¶
Close fsyncs and closes the active segment (without sealing it), unmaps all sealed segments, and releases the directory lock.
func (*Store) Compact ¶
func (s *Store) Compact(live func(key.Key) bool, opts CompactOpts) (CompactStats, error)
Compact seals the active segment, rewrites eligible sealed segments whose dead ratio reaches opts.MinDeadRatio (re-verifying live records while copying), and deletes victims only after the copies are durable. The caller must guarantee no ingest or reference publication overlaps (specs/gc.qnt). A grey set captured since BeginBarrier is consumed and kept alongside live.
func (*Store) Get ¶
Get returns the bytes stored under k, or ErrNotFound if k is absent. The returned slice is caller-owned.
func (*Store) GetRecord ¶
GetRecord returns a caller-owned copy of the full on-disk record stored under k — its 46-byte header plus the stored (still-compressed) payload, exactly as written by amberpack.EncodeRecord — or ErrNotFound if k is absent. This is the zero-copy push path: the record is wire-format-identical, so a caller can hand it to amberpack.Writer.AddRecord without decompressing and re-encoding. Like Get, it does not CRC-check; the receiving Reader validates framing and CRC.
func (*Store) HasOutside ¶
HasOutside reports whether k is stored anywhere but segment id: the active segment or any other sealed segment. Reaping uses it to skip records that already survived.
func (*Store) Liveness ¶
Liveness classifies every record by live(k) and reports per segment, ascending by id with the active segment last: the GC dry run.
func (*Store) Missing ¶
Missing reports which of keys are absent from the store, preserving the input's order and multiplicity. Lookups run concurrently over contiguous chunks of the input.
func (*Store) NewMarkSet ¶
func (*Store) ObserveKeys ¶
ObserveKeys marks keys live for the next Compact, like the write paths' per-key observation. A reference PUT that lands while a mark is running calls it with the root's whole closure, so a reference committed during the cycle never dangles.
func (*Store) OldestInflightWrite ¶
OldestInflightWrite returns the start time of the oldest Put, WriteBatch or WriteParallel still in progress. The GC horizon never passes it.
func (*Store) Put ¶
Put stores a single object under k, deduplicating against existing content. A dedup hit returns success without fsyncing; if the matching record was appended by a still-running batch, its durability rides on that batch's commit.
func (*Store) PutVerified ¶ added in v0.0.5
PutVerified verifies data and repairs every corrupt indexed copy of k. Repairs preserve segment IDs and unrelated records through a synced rename. Existing readers retain their old mappings until their walks finish. New objects use the configured write durability, like Put.
func (*Store) Record ¶
Record returns the raw record at off in segment id, CRC-checked. The returned bytes are caller-owned and re-appendable verbatim.
func (*Store) Remove ¶
Remove drops sealed segment id: out of the probe list under the write lock (draining in-flight reads, like a seal), munmap after in-flight scrubs (as Close does), unlink, fsync the directory. The caller ensures every live record was copied out first.
waitScrubs is a global gate: it waits out every in-flight scrub, not just this segment's pins; under sustained ScanIndex/Verify traffic that can stall Remove (and a gc deletePack holding its removal lock). Per-segment pinning is the known follow-up before a long-running embedder leans on this.
func (*Store) ScanIndex ¶
ScanIndex walks segment id's footer index in index order (fanout on the key's last byte, then full key), calling fn with each entry's key, record offset and stored payload length. No pack body is read.
func (*Store) Segments ¶
func (s *Store) Segments() ([]SegmentInfo, error)
Segments lists the sealed segments, ascending id. The active segment is not listed; it is never a reaping victim.
func (*Store) SortByLocation ¶
SortByLocation reorders keys in place to follow the store's on-disk layout — grouped by segment, ascending offset within a segment — so reading them in order is a near-sequential sweep per segment rather than scattered random access. Absent keys sort last (their reads surface ErrNotFound later). It is a no-op on a closed store.
func (*Store) StoredSize ¶
StoredSize returns the stored (post-compression) payload length of the object under k and whether k was found, reading only the index — no payload read. It sizes objects for byte-balanced push batching against the bytes that actually travel.
func (*Store) Sync ¶
Sync fsyncs the active segment, making every append so far durable — AppendRecord batches end with one Sync.
func (*Store) Verify ¶
Verify scrubs every sealed segment: walks the body record by record (validating framing, CRCs, and that each payload re-hashes to its key), recomputes the index section and compares it bytewise with the footer's, and checks the filter contains every body key. The active segment is covered by tail-scan on reopen, not by Verify. Segments sealed by rotations that happen after Verify snapshots the segment list are not covered by that call. Close blocks until in-flight Verify walks finish; cancel the context to stop a scrub early.
func (*Store) Wipe ¶
Wipe deletes every object: the active segment and all sealed segments are closed and their files removed, leaving an empty, still-open store (the store-wipe operation). Readers are drained via the write lock before segments are detached; in-flight Verify walks are waited out before unmapping, exactly like Close. nextID stays monotonic so segment names never repeat within a session.
func (*Store) WriteBatch ¶
WriteBatch stores every object the iterator yields, fsyncing once at the end (when WithSync is enabled): on return, all yielded objects are durable. It is NOT atomic — a crash or iterator error can leave a valid prefix stored. In a content-addressed store that prefix is harmless: identical re-pushed content deduplicates. Objects repeated within the batch, or already present, are written once. When WriteBatch returns an error after appending part of the batch, it best-effort fsyncs that prefix first, so Has-visible records never stay non-durable.
func (*Store) WriteParallel ¶
WriteParallel stores every object the iterator yields using multiple concurrent workers. Compression and (optional) verification run in parallel; appends serialize on the active segment. Each worker fsyncs after appending BatchSize bytes, and the run fsyncs once more before returning.
Like WriteBatch, WriteParallel is durable-on-return but NOT atomic: on error or crash a valid prefix remains, which a content-addressed re-run deduplicates. The final fsync also covers dedup hits against records a concurrent, uncommitted run appended. If the iterator yields an error, WriteParallel stops and returns it. With opts.Verify, a key/payload mismatch stops the run with a wrapped ErrVerify. An Object may carry a pre-encoded Record instead of Data; it is validated and appended as is (see Object).
type WriteOpts ¶
type WriteOpts struct {
Writers int // concurrent writers; <= 0 means GOMAXPROCS
BatchSize int // fsync when a writer has appended this many bytes; <= 0 means DefaultBatchSize
Verify bool // recompute and check each new object's key before storing it
}
WriteOpts configures WriteParallel.
type WriteStats ¶
type WriteStats struct {
Stored int // objects newly written
Deduped int // objects skipped (already present, or duplicated in the stream)
BytesStored int64 // payload bytes of newly-written objects (uncompressed)
}
WriteStats summarizes one WriteParallel run. On a non-nil error, the stats reflect the work done before the abort.
Source Files