packstore

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 Object
• type Option
  • func WithSegmentSize(n int64) Option
  • func WithSync(b bool) Option
• type Store
  • func Open(dir string, opts ...Option) (*Store, error)
  • func (s *Store) Close() 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) Missing(keys []key.Key) ([]key.Key, error)
  • func (s *Store) Put(k key.Key, data []byte) error
  • func (s *Store) SortByLocation(keys []key.Key)
  • func (s *Store) StoredSize(k key.Key) (uint64, bool, error)
  • func (s *Store) Verify(ctx context.Context) 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 = 256 << 20 // 256 MiB

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 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.

Types

type Object

type Object struct {
	Key  key.Key
	Data []byte
}

Object is one CAS object: its key and its serialized bytes.

type Option

type Option func(*config)

Option configures a Store at Open time.

func WithSegmentSize

func WithSegmentSize(n int64) Option

WithSegmentSize sets the rotation threshold in bytes. A single oversized record may push one segment past it.

func WithSync

func WithSync(b bool) Option

WithSync controls whether writes are fsynced for crash durability. Default is true; disabling it speeds bulk loads and tests.

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

func Open(dir string, opts ...Option) (*Store, error)

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) Close

func (s *Store) Close() error

Close fsyncs and closes the active segment (without sealing it), unmaps all sealed segments, and releases the directory lock.

func (*Store) Get

func (s *Store) Get(k key.Key) ([]byte, error)

Get returns the bytes stored under k, or ErrNotFound if k is absent. The returned slice is caller-owned.

func (*Store) GetRecord

func (s *Store) GetRecord(k key.Key) ([]byte, error)

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) Has

func (s *Store) Has(k key.Key) (bool, error)

Has reports whether an object is stored under k.

func (*Store) Missing

func (s *Store) Missing(keys []key.Key) ([]key.Key, error)

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) Put

func (s *Store) Put(k key.Key, data []byte) error

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) SortByLocation

func (s *Store) SortByLocation(keys []key.Key)

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

func (s *Store) StoredSize(k key.Key) (uint64, bool, error)

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) Verify

func (s *Store) Verify(ctx context.Context) error

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) WriteBatch

func (s *Store) WriteBatch(seq iter.Seq2[Object, error]) error

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

func (s *Store) WriteParallel(seq iter.Seq2[Object, error], opts WriteOpts) (WriteStats, error)

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 once more when the input is exhausted.

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 (a dedup hit against a record appended by a concurrent, uncommitted run rides on that run's eventual fsync). 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.

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.