packstore

package
v0.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 3, 2026 License: GPL-3.0, LGPL-3.0 Imports: 29 Imported by: 0

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

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

View Source 
const DefaultSegmentSize = 2 << 30 // 2 GiB

DefaultSegmentSize is the default rotation threshold: the active segment is sealed once it reaches this many bytes.

Variables

View Source 
var ErrClosed = errors.New("packstore: store closed")

ErrClosed is returned by operations on a closed store.

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

View Source 
var ErrNotFound = errors.New("packstore: object not found")

ErrNotFound is returned by Get for a key that is not present in the store.

View Source 
var ErrUnknownSegment = errors.New("packstore: no such segment")

ErrUnknownSegment reports an id that names no sealed segment (never sealed, or already removed).

View Source 
var ErrUnsupportedVersion = errors.New("packstore: unsupported segment format version")

ErrUnsupportedVersion reports a segment whose header is this format's magic with another version byte: data written by a release with a different layout. Such a file is neither read nor modified.

View Source 
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. Bytes the store already held that fail the same check are its own damage, and wrap ErrCorrupt as well (Verify, GetVerified): look for that one first.

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)
	// MaxCopyBytes, when non-nil, caps the live record bytes this pass
	// copies. Victim selection skips a segment whose live bytes exceed what
	// is left of the cap and goes on to smaller ones, so the pass never
	// copies more. Record bytes only: footers and allocation overhead sit
	// outside it. At a cap of 0 fully dead segments are still reclaimed,
	// since those need no copy, and nothing is sealed, since a footer would
	// grow the store. Nil, the zero value, means no cap.
	MaxCopyBytes *uint64
	// MarkVerified says that live comes from a walk that verified every object it
	// read (GetVerified), so that a bad record found here cannot have hidden anything
	// from it. The pass then leaves that record's segment in place and goes on with
	// the others. Without it, any bad record stops the pass with nothing removed: a
	// walk may have read a bad directory as something smaller, and the pass may meet
	// that directory or another bad record first, which says nothing about it.
	//
	// A pass reads only its victims, so even a stop promises nothing about a bad
	// record in a segment that is not one. A caller that needs that promise has to
	// verify what it marks.
	MarkVerified bool
}

CompactOpts tune one Compact pass.

type CompactStats 

type CompactStats struct {
	SegmentsScanned   int // sealed segments considered
	SegmentsCompacted int
	Victims           []uint64         // ids of the segments removed, ascending
	Skipped           []SkippedSegment // victims left in place, ascending by id
	RecordsCopied     int
	BytesCopied       uint64
	BytesFreed        uint64 // removed file sizes, footers included
}

type CompressionFunc added in v0.10.0 

type CompressionFunc func(k key.Key, data []byte, def amberpack.Compression) amberpack.Compression

CompressionFunc returns the compression to use for one object. def is the store's WithCompression value: returning it accepts the store's setting, returning the zero Compression stores the object raw, and anything else overrides the setting for this object.

It runs on whichever goroutine is writing, several at once under WriteParallel, so it must be safe for concurrent use. It must not modify or keep data, and it must not call the store: a repair calls it with the store's append lock held. A returned value that does not validate fails that object's write with an error wrapping amberpack.ErrInvalidCompression.

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.

func (*MarkSet) Contains 

func (m *MarkSet) Contains(k key.Key) bool

func (*MarkSet) Mark 

func (m *MarkSet) Mark(k key.Key) (newly, present bool)

Mark marks k, reporting whether it was unmarked before and whether it is present in the snapshot at all.

func (*MarkSet) MarkWalked added in v0.11.0 

func (m *MarkSet) MarkWalked(k key.Key) (first bool)

MarkWalked records that a walk follows k's links, reporting whether it is the first to. It is apart from Mark because an object can be marked without being read: a git key is marked through a link that calls it a blob, and may still be named as a directory by another. It reports false for a key that is not in the snapshot.

func (*MarkSet) Marked 

func (m *MarkSet) Marked() int

Marked returns the number of distinct keys marked so far.

type Object 

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

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.EncodeRecordWith 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 WithCompression added in v0.10.0 

func WithCompression(c amberpack.Compression) Option

WithCompression sets the compression for the objects this store encodes: those given to Put, PutVerified and PutVerifiedDeferred, and those a batch carries as Data. The default is no compression. The setting belongs to this handle and is stored nowhere; reading never depends on it, and records that arrive already encoded keep the codec they have.

func WithCompressionFor added in v0.10.0 

func WithCompressionFor(f CompressionFunc) Option

WithCompressionFor makes the store ask f for every object it encodes, whatever the WithCompression value is. f is not asked for an object the store already holds, nor for a pre-encoded record. A nil f means no callback.

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 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 SkippedError added in v0.12.0 

type SkippedError struct {
	Skipped []SkippedSegment
}

SkippedError is what Compact returns, beside its stats, when it left victims in place, which it does only for a caller with a verified mark (CompactOpts.MarkVerified). The rest of the pass was carried out: the stats say what was removed. A segment left behind reports again whenever it is a victim again, until the record is repaired with PutVerified or nothing live refers to it.

func (*SkippedError) Error added in v0.12.0 

func (e *SkippedError) Error() string

func (*SkippedError) Unwrap added in v0.12.0 

func (e *SkippedError) Unwrap() []error

Unwrap gives errors.Is and errors.As each segment's cause.

type SkippedSegment added in v0.12.0 

type SkippedSegment struct {
	ID     uint64
	Path   string
	Key    key.Key // the record's key
	Offset uint64  // the record's offset in the segment
	Err    error   // wraps ErrCorrupt, and ErrVerify when the payload is not what the key says
}

SkippedSegment is a victim that a pass left in place, because a live record in it failed verification. The first such record found is the one named; there may be more.

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 

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

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 

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

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 

func (s *Store) BeginSweep(ctx context.Context) (done func(), err error)

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 

func (s *Store) BeginWrite() (done func(), err error)

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 

func (s *Store) Close() error

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. Under opts.MaxCopyBytes a segment the cap does not cover is left for a later pass, and at a cap of 0 the active segment is not sealed either.

A live record that fails verification stops the pass with nothing removed, unless the caller's mark verified what it read (CompactOpts.MarkVerified). Then its segment stays where it is, the pass goes on with the others, and it returns a *SkippedError beside stats that say what it did. Any other error means nothing was removed, unless it came from removing the victims themselves.

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.EncodeRecordWith — 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) GetVerified added in v0.12.0 

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

GetVerified is Get, and checks that the bytes it read hash to the key. Get trusts them, which is what a read of file content can afford. A walk that decides what is alive cannot: a bad directory can still decode, and walks as a smaller one (a collector's mark, and a reference's completeness check). A payload that decodes and hashes to something else wraps both ErrCorrupt and ErrVerify; one that does not even decode fails as it does for Get. What a key's length says is for whoever reads the object: a commit keyed by an older rule is told apart by the walk, which says what to do about it (fstree's decodeCommit).

A failure gets one more try on a fresh view: a repair replaces a segment under its name and moves no generation, so another store's may be what this one has yet to see.

func (*Store) Has 

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

Has reports whether an object is stored under k.

func (*Store) HasOutside 

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

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 

func (s *Store) Liveness(live func(key.Key) bool) ([]SegmentLiveness, error)

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 

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

func (s *Store) NewMarkSet() *MarkSet

func (*Store) ObserveKeys 

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

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 

func (s *Store) OldestInflightWrite() (time.Time, bool)

OldestInflightWrite returns the start time of the oldest Put, WriteBatch or WriteParallel still in progress. The GC horizon never passes it.

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. Records that PutVerifiedDeferred left unsynced are the exception: a dedup hit syncs them first.

func (*Store) PutVerified added in v0.0.5 

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

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.

A bad copy in another store's active segment is reached by taking that segment over, and sealing it. One that another store holds, in an active segment or a crashed seal, cannot be repaired, and lookups may read it before any other copy, so that is an error: call again when that store has let go.

func (*Store) PutVerifiedDeferred added in v0.0.12 

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

PutVerifiedDeferred is PutVerified without the fsync of a new or healthy active record. The caller ends a run of these with Sync, and must not treat the content as stored before that returns. Repairs still sync.

It looks at the directory at most once a second (deferredLookEvery), so it does not answer for a copy that appeared since. Before it repairs or refuses anything it looks.

func (*Store) Record 

func (s *Store) Record(id, off uint64) ([]byte, error)

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 

func (s *Store) Refresh() error

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 

func (s *Store) Remove(id uint64) error

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 

func (s *Store) ScanIndex(id uint64, fn func(k key.Key, off uint64, slen uint32)) error

ScanIndex walks segment id's footer index in index order (ascending key order), 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 

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

func (s *Store) Sync() error

Sync fsyncs the active segment, making every append so far durable — AppendRecord batches end with one Sync.

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

func (s *Store) Wipe() error

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 

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

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.