Documentation
¶
Overview ¶
Package packstore persists Amber-Store CAS objects in log-structured, append-only segment (pack) 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). An active segment is indexed in its owner's memory and, for everybody else and for the next open, by a sidecar file beside it (sidecar.go). There is no global index. A directory may be open in any number of stores, in any number of processes: a writer owns an active segment of its own (active.go), readers lock nothing and look at the directory again when they miss (view.go), and one lock file keeps writers and a GC sweep apart (gate.go). All format integers are big-endian. Record framing lives in the amberpack package. See architecture/packstore.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) BeginSweep(ctx context.Context) (done func(), err error)
- func (s *Store) BeginWrite() (done func(), err error)
- 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) Refresh() 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, by goroutines and — several stores on one directory — by processes (active.go, view.go). Lock ordering: appendMu, then refreshMu, then mu, never the reverse. appendMu serializes the write path (append, fsync, seal, Close); refreshMu serializes refreshes of the view; mu guards sealed/active/foreign/closed for readers.
func Open ¶
Open opens (creating if necessary) a store rooted at dir. Any number of stores, in any number of processes, may have a directory open at once. Opening locks no segment and modifies none: sealed segments are mmap'd and validated, active ones indexed from their sidecars for reading. The store takes an active segment of its own at its first write (active.go).
The directory is flocked shared for the store's life. Releases from before stores could share a directory take that lock exclusively and assume they own the one active segment; this keeps them out, and keeps this store out while one of them is in.
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) BeginSweep ¶ added in v0.0.10
BeginSweep takes the store for a GC cycle: it waits for this store's write spans and every other store's to end, keeps other stores' writers out until done is called, and refreshes the view, which is then complete and stable. This store's own writers go on: the write barrier and the collector's reference lock are what holds them where needed. The view generation moves as soon as the lock is taken, so that every other store looks at the directory again before its next write, even if this process dies half way. Compact, Wipe and Remove take the gate themselves; inside a BeginSweep they find it held.
func (*Store) BeginWrite ¶ added in v0.0.10
BeginWrite opens a write span: until done is called, no other store can sweep. Every exported write opens one itself; callers bracket larger spans that must not straddle a sweep, such as a completeness walk followed by a reference put. If another store has swept since this one last looked, the view is refreshed before BeginWrite returns. Spans nest: a write, or another BeginWrite, inside an open span joins it at no cost and never waits.
A process that also runs a gc.Collector on this store opens its spans through the collector (Collector.BeginSpan), which takes its reference lock before this gate, the order a cycle takes them in. A span opened here and held across Collector.PrepareRef takes them the other way round, and the two can wait for each other.
func (*Store) Close ¶
Close fsyncs and closes the active segment this store owns (without sealing it, so that the next writer goes on filling it), unmaps all sealed segments, and releases the segment and directory locks.
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) Refresh ¶ added in v0.0.10
Refresh brings the store's view of the directory up to date: the segments other stores sealed, created or reaped, and what they appended. Lookups do it by themselves when they find nothing. A caller that is about to work from a snapshot of the view, as gc's advisory mark does, asks for it.
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: every segment, sealed or active, is closed and its file removed, leaving an empty, still-open store (the store-wipe operation). It refuses, deleting nothing, while another store owns an active segment: that writer would go on appending to a file that is gone. Readers are drained via the write lock before segments are detached; in-flight Verify walks are waited out before unmapping, exactly like Close.
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