pack

package
v1.66.0 Latest Latest
Warning

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

Go to latest
Published: Oct 13, 2026 License: Apache-2.0 Imports: 47 Imported by: 0

Documentation

Overview

Package pack stores objects in archives, each with an index of what it holds, in an ArchiveStorage such as a directory or a bucket. PackStorage is the store, built with NewPackStorage; it also compacts, garbage collects, scrubs and repairs its archives, and keeps each set's head.

Index

Constants

View Source
const (
	// DefaultClaimOpen is how long an archive stays open unless WithClaims
	// says otherwise.
	DefaultClaimOpen = 10 * time.Second
	// DefaultClaimGrace is how much longer a claim holds while its archive
	// is closed, unless WithClaims says otherwise.
	DefaultClaimGrace = 2 * time.Minute
)
View Source
const (
	SessionBeginExt = ".begin"
	SessionEndExt   = ".end"
)

SessionBeginExt and SessionEndExt name a session's markers in the storage: <id>.begin holds the session as it began and <id>.end how it ended. An end is only ever created once, so a commit and anything ending the session agree on a single outcome. The end is what says a session's archives are committed, so it is only deleted after its begin, once no archive of the session is stored, live or retired.

View Source
const (
	ArchiveSuffix = ".goback"
	IndexExt      = ".idx"
	// CommittedExt marks an archive of a session that a commit made
	// committed, in stores written before the session's end marker said so.
	// The store still honours such a marker and deletes it with its archive,
	// but no longer writes one.
	CommittedExt = ".committed"
)
View Source
const DefaultQuarantine = 24 * time.Hour

DefaultQuarantine is how long a retired archive is kept by default, so reads already under way, in this process or another, finish against it.

View Source
const GCExt = ".gc"

GCExt is the extension of the mark result stored beside an archive's index.

View Source
const HaltName = "collection.halt"

HaltName is the file a collection leaves when its pre-delete check finds a copy it was about to drop reachable. While it exists no collection drops anything; an operator deletes it once the cause is understood.

View Source
const HeadExt = ".head"

HeadExt names the pointers to each set's latest commit: sets/<set id>.head.

View Source
const PlanExt = ".plan"

PlanExt names the sweeps collections published instead of running: <generation>.plan lists, per archive, the records the rewrite drops. While one is left, collections wait for RewritePlan to run it.

View Source
const RetiredExt = ".retired"

RetiredExt marks an archive a rewrite retired: <archive>.retired names the day it was retired. The store no longer loads the archive, a collection deletes it once the quarantine period has passed, and RestoreQuarantined puts it back until then.

View Source
const SealExt = ".seal"

SealExt names the seals collections leave in the storage: <generation>.seal holds the versions of the tombstones that generation condemned with. A session that commits checks what it relied on against the seals written since it began.

Variables

View Source
var (
	ErrFileNotFound = errors.New("requested file was not found")
	// ErrFileExists is CreateNew's answer for a name that is taken.
	ErrFileExists       = errors.New("the file exists")
	ErrInvalidExtension = errors.New("the provided extension is invalid")
)
View Source
var ErrAtRestKeyMismatch = errors.New("record is sealed at rest under another key")

ErrAtRestKeyMismatch is returned when a record is sealed under a key the store's keyset does not hold.

View Source
var ErrAtRestKeyMissing = errors.New("record is sealed at rest and the store has no key")

ErrAtRestKeyMissing is returned when a record is sealed at rest and the store was opened without a key.

View Source
var ErrBrokenChain = errors.New("object header names the wrong predecessor")

ErrBrokenChain is the failure of an object whose header does not name the object written before it in the archive.

View Source
var ErrClaimLapsed = errors.New("the claim on the archive lapsed")

ErrClaimLapsed is what a ClaimIndex answers for an open archive whose claim ran out before it was finalized.

View Source
var ErrHalted = errors.New("collection halted: the pre-delete check found reachable copies the sweep would drop")

ErrHalted is what a collection reports when its check halts the store's collections.

View Source
var ErrNoSignedURL = errors.New("this storage hands out no URLs")

ErrNoSignedURL is a RangeSigner's answer when it cannot address the bytes it holds.

View Source
var ErrNotQuarantined = errors.New("the quarantine holds no such archive")

ErrNotQuarantined is RestoreQuarantined's answer for an archive the quarantine does not hold.

View Source
var (
	// ErrRecordNotFound is LocateObject's answer when no visible archive holds the object.
	ErrRecordNotFound = errors.New("couldn't find index record")
)
View Source
var ErrUploadMismatch = errors.New("stored archive does not match what was written")

ErrUploadMismatch is returned when the bytes an archive was written from are not the bytes its storage kept.

Functions

func DecodeRecord

func DecodeRecord(record []byte, atRestKey []byte) (*proto.Object, error)

DecodeRecord turns the body a Location yields back into its object, opening it with the location's at-rest key when it is sealed.

Types

type ArchiveBatchObserver

type ArchiveBatchObserver interface {
	ArchiveObserver
	// ArchivesDeleted is told the archives' files are gone from the storage.
	ArchivesDeleted(names []string)
}

ArchiveBatchObserver is an ArchiveObserver told about archives deleted together in one call, in place of a call to ArchiveDeleted for each.

type ArchiveIndex

type ArchiveIndex interface {
	SessionIndex

	LocateObject(ref *proto.Ref, scope Scope, exclude ...string) (IndexLocation, error)
	// LocateCopies returns every copy the archives visible to scope hold
	// of each ref, keyed by its hash; a ref none holds is absent.
	LocateCopies(refs []*proto.Ref, scope Scope) (map[string][]IndexLocation, error)
	// LocateTombstones is LocateCopies for refs of tombstones and
	// un-tombstones: it finds tombstone records only.
	LocateTombstones(refs []*proto.Ref, scope Scope) (map[string][]IndexLocation, error)
	LookupArchive(archive string) (ArchiveInfo, bool, error)
	// IndexArchive registers an archive; a pending one needs a live session.
	// Of an archive it already knows, it only records a creation time the
	// index lacks.
	IndexArchive(archive ArchiveInfo, index IndexFile) error
	// DeleteArchives forgets the archives and their objects, all or none.
	DeleteArchives(names []string) error
	Close() error
	CountObjects() (uint64, uint64, error)
}

ArchiveIndex maps refs to the archives holding them, records each archive's visibility, and keeps the live sessions.

type ArchiveInfo

type ArchiveInfo struct {
	Name    string
	Session string
	State   ArchiveState
	// Created is when the bucket created the archive's index file, which
	// versions every record the archive holds; zero while it has none.
	Created time.Time
}

ArchiveInfo is what the archive index records about an archive.

type ArchiveObserver

type ArchiveObserver interface {
	// ArchiveStored is told an archive and its index are in the storage,
	// the bytes both take up and the session that wrote them; the session
	// is empty for an archive the store rewrote itself.
	ArchiveStored(name string, bytes int64, session string)
	// ArchiveDeleted is told an archive's files are gone from the storage.
	ArchiveDeleted(name string)
}

ArchiveObserver is told about every archive the store puts into its storage and every one it deletes from there, so a deployment can account for what the storage holds. It is called on the path that stored or deleted the archive, so it must be quick.

type ArchiveState

type ArchiveState uint8

ArchiveState is the visibility of an archive: committed archives are served to everyone, pending ones only to the session that wrote them. Open and lost archives are only ever held by a ClaimIndex: an open one is still being written by some process, so its objects exist for the session but cannot be read yet; a lost one was open when its claim lapsed.

const (
	ArchiveCommitted ArchiveState = iota
	ArchivePending
	ArchiveOpen
	ArchiveLost
)

type ArchiveStorage

type ArchiveStorage interface {
	Create(name string) (File, error)
	// CreateNew stores data under name unless a file has it already, which
	// it answers with ErrFileExists; of several callers racing for a name,
	// exactly one succeeds.
	CreateNew(name string, data []byte) error
	Open(name string) (File, error)
	List(extension string) ([]string, error)
	Delete(name string) error
	DeleteAll() error
}

ArchiveStorage holds a store's files under slash-separated names; Open of a missing name is ErrFileNotFound.

type AtRestKey

type AtRestKey struct {
	// contains filtered or unexported fields
}

AtRestKey seals every payload on its way into an archive and opens it on the way out; headers stay readable, so an index can be rebuilt without the key. It is the server's own layer: an agent's sealed blobs are sealed once more, and what an agent sent in plaintext is plaintext again for every reader with the key.

Each archive is sealed under its own key, derived from this one and the archive's name, so a location can carry the key to one archive without giving away the store's. Its primary key seals and every key in it opens, so rotating is Rotate, and a key can leave the keyset once a rewrite has re-sealed the last record naming it (Scrub counts them).

func GenerateAtRestKey

func GenerateAtRestKey() (*AtRestKey, error)

GenerateAtRestKey makes a keyset of one fresh key.

func LoadAtRestKey

func LoadAtRestKey(path string) (*AtRestKey, error)

LoadAtRestKey reads a keyset written by Save.

func NewAtRestKey

func NewAtRestKey(h *keyset.Handle) (*AtRestKey, error)

NewAtRestKey wraps a keyset of AEAD key derivers, as GenerateAtRestKey makes them.

func UnwrapAtRestKey

func UnwrapAtRestKey(wrapped []byte, kek tink.AEAD, ad []byte) (*AtRestKey, error)

UnwrapAtRestKey opens a keyset Wrap sealed under kek with ad.

func (*AtRestKey) ID

func (k *AtRestKey) ID() []byte

ID marks the records sealed under the primary key.

func (*AtRestKey) Rotate

func (k *AtRestKey) Rotate() (*AtRestKey, error)

Rotate returns the keyset with a fresh primary key; the keys it held still open what they sealed.

func (*AtRestKey) Save

func (k *AtRestKey) Save(path string) error

Save writes the keyset, readable by its owner only.

func (*AtRestKey) Wrap

func (k *AtRestKey) Wrap(kek tink.AEAD, ad []byte) ([]byte, error)

Wrap returns the keyset sealed under kek, a key-encryption key such as one held in a KMS, bound to ad; UnwrapAtRestKey reverses it.

type Attribution

type Attribution struct {
	Group int64
	Set   int64
}

Attribution is what a root belongs to: the group its objects are counted in and the set they are recorded against. An object is counted once per group, so two groups that both hold it each carry it in full, and within a group it goes to the first set the mark reached it from.

type Checksummer

type Checksummer interface {
	Checksum(name string) ([]byte, error)
}

Checksummer is implemented by an ArchiveStorage whose backend reports a checksum of a stored file, so an upload can be checked without reading it back. An empty sum means the backend has none for that file.

type Claim

type Claim struct {
	Archive string
	Age     time.Duration
}

A Claim is an open archive of a session and how long ago it was claimed, by the index's clock.

type ClaimIndex

type ClaimIndex interface {
	// OpenArchive claims a new archive for a live session.
	OpenArchive(name, session string) error
	// AddObjects indexes objects written to an open archive, or answers
	// ErrClaimLapsed when the archive is no longer open.
	AddObjects(archive string, records []IndexRecord) error
	// FinalizeArchive turns an open archive pending, created at created,
	// when it was claimed less than within ago, and answers ErrClaimLapsed
	// otherwise.
	FinalizeArchive(name string, within time.Duration, created time.Time) error
	// Holds reports whether a pending or open archive of the session holds
	// the object.
	Holds(ref *proto.Ref, session string) (bool, error)
	// Claims lists the open archives of the session.
	Claims(session string) ([]Claim, error)
	// Abandon turns an open archive lost when it was claimed at least
	// within ago, and reports whether it did.
	Abandon(name string, within time.Duration) (bool, error)
	// Lost lists the objects the session's lost archives held that none of
	// its pending archives and no committed archive holds, leaving out
	// commits and the un-tombstones of the commits its lost and pending
	// archives hold, which belong to the commit that was refused.
	Lost(session string) ([]*proto.Ref, error)
}

A ClaimIndex lets several processes write one session. A process claims each archive it opens and indexes the archive's objects as it writes them, so every process sees them at once; the claim holds for a bounded time, and one that lapsed can never be finalized. Ages are measured by the index's own clock, the one every process shares.

type CollectOptions

type CollectOptions struct {
	// Readers is the number of parallel object reads during the mark (32).
	Readers int
	// DeadRatio selects an archive for sweeping once this share of its bytes
	// is dead (0.3).
	DeadRatio float64
	// ErasureBound selects an archive that has held dead objects this long
	// regardless of ratio (14 days).
	ErasureBound time.Duration
	// MinAge keeps every object younger than this, whatever the mark says (24h).
	MinAge time.Duration
	// NoSweep stops after the mark.
	NoSweep bool
	// Handoff publishes the sweep as a plan for RewritePlan instead of
	// running it.
	Handoff bool
	// Quarantine is how long the files of an archive a rewrite retired
	// are kept before a collection deletes them (DefaultQuarantine).
	Quarantine time.Duration
	// TempDir holds the live-set runs and the checkpoints a crashed run
	// resumes from (os.TempDir()).
	TempDir string
	// Now is the snapshot time (time.Now()).
	Now time.Time
	// Owner names what a root belongs to, so the mark can attribute the
	// objects it reaches; nil, or a zero Attribution, attributes nothing.
	Owner func(root []byte) Attribution
}

CollectOptions tunes one garbage collection run. Zero values take the defaults documented on each field.

type CollectReport

type CollectReport struct {
	Generation  uint64
	Archives    int
	Objects     uint64
	Roots       int
	Marked      uint64
	DeadObjects uint64
	DeadBytes   uint64
	// Resumed counts the root batches taken from a previous, interrupted run.
	Resumed int
	// ErasedArchives counts the archives flagged for holding the objects of
	// an erased commit.
	ErasedArchives int
	// SetBytes is what each set's live objects take up in the archives,
	// keyed by the set ids Owner returned. Within a group an object counts
	// once, for the first set the mark reached it from; across groups it
	// counts in each.
	SetBytes map[int64]uint64
	// SetDeduplicated is the uncompressed size of the distinct file content
	// each set's live objects carry, attributed as SetBytes is.
	SetDeduplicated map[int64]uint64
	// SetAlone is what each set's live objects would take up were it the
	// only set of its group, and SetExclusive what of that no other set of
	// the group reaches: what deleting the set would free.
	SetAlone, SetExclusive map[int64]uint64
	// SetDeduplicatedAlone is the uncompressed size of the distinct file
	// content each set's live objects carry were it the only set of its group.
	SetDeduplicatedAlone map[int64]uint64
	// Unattributed is what the live objects no set reaches take up, such
	// as those only a commit without an owner reaches; zero without Owner.
	Unattributed uint64
	// Condemned counts the unreachable objects the run stored tombstones for.
	Condemned int
	// Waiting is the generation whose published plan has not been
	// rewritten yet; a run that waits does nothing else.
	Waiting uint64
	// SweepSkipped names the reason when the run marked but did not sweep.
	SweepSkipped string
	// Published counts the archives of the plan the run published.
	Published        int
	Swept            int
	ReclaimedObjects uint64
	ReclaimedBytes   uint64
	// CopiedBytes is what the sweep wrote again to reclaim that.
	CopiedBytes uint64
	// Purged counts the quarantined files the run deleted.
	Purged int
	// ArchiveBytes is what the archives the run marked take up in storage,
	// dead objects included, and RetiredBytes what the archives rewrites
	// retired still take up until they are purged.
	ArchiveBytes, RetiredBytes uint64
	// OldestDead is when the archive that has held dead objects longest
	// without being rewritten first held them; zero when none holds any.
	OldestDead time.Time
	Duration   time.Duration
}

CollectReport summarizes one garbage collection run.

func (*CollectReport) Summary

func (r *CollectReport) Summary() string

Summary renders the report as a few lines for a log or an operator.

type Collector

type Collector interface {
	Collect(ctx context.Context, opts CollectOptions) (*CollectReport, error)
}

Collector is implemented by stores that can garbage collect themselves.

type CompactReport

type CompactReport struct {
	// Candidates counts the small committed archives Compact found, and
	// CandidateBytes what they take up; it merges them only once they add
	// up to a batch or are more than MinimumCandidates.
	Candidates     int
	CandidateBytes uint64
	// Rewritten counts the candidates rewritten and retired, and ReadBytes
	// what they took up.
	Rewritten int
	ReadBytes uint64
	// Written counts the archives written, and WrittenBytes what they take
	// up.
	Written      int
	WrittenBytes uint64
	// Moved counts the objects copied into the written archives, and
	// CopiedBytes their bytes.
	Moved       uint64
	CopiedBytes uint64
	// Superseded counts the objects not copied because another copy,
	// outside the candidates or written by the same run, stands in for them.
	Superseded uint64
	// ReclaimedBytes is ReadBytes less WrittenBytes, or zero.
	ReclaimedBytes uint64
	Duration       time.Duration
}

CompactReport summarizes one Compact. Alongside an error it counts what the compaction finished before it stopped.

func (*CompactReport) Summary

func (r *CompactReport) Summary() string

Summary renders the report as a line for a log or an operator.

type CompactionConfig

type CompactionConfig struct {
	// Small is the size under which a committed archive is merged; zero
	// means 16 MiB.
	Small uint64
	// Batch is how much the small archives must add up to before Compact
	// merges them; zero means 256 MiB.
	Batch uint64
	// MinimumCandidates is how many small archives make Compact merge them
	// whatever they add up to.
	MinimumCandidates int
	// Workers is how many archives a rewrite, by Compact or Collect, reads
	// at once; zero means four per GOMAXPROCS, at most 16. Compact gives
	// each its own output, Collect each set and age of what survives.
	Workers int
	// Chunk is how many archives a rewrite takes at a time, retiring each
	// chunk as it finishes; zero means 1000.
	Chunk int
}

CompactionConfig tunes Compact; when it runs is the caller's business.

type File

type File interface {
	io.Reader
	io.Writer
	io.Seeker
	io.Closer

	Stat() (fs.FileInfo, error)
}

File is one archive, index or gc file held by an ArchiveStorage.

func OpenListed

func OpenListed(storage ArchiveStorage, file ListedFile) (File, error)

OpenListed opens a listed file through the storage's ListedOpener when it is one and Open otherwise; a wrapping ArchiveStorage implements ListedOpener with it.

type Head struct {
	SetID        uint64 `json:"setId"`
	Set          string `json:"set"`
	Commit       []byte `json:"commit"`
	ReceivedAtNs int64  `json:"receivedAtNs"`
}

Head is the latest complete commit of a set the store holds, as kept in the storage beside the archives, so the set's backups are found without any index.

func HeadsOf

func HeadsOf(storage ArchiveStorage) ([]Head, error)

HeadsOf lists the heads a store's storage holds, without opening the store.

type InMemoryIndex

type InMemoryIndex struct {

	// Now is the clock claims are aged by; nil means time.Now.
	Now func() time.Time
	// contains filtered or unexported fields
}

InMemoryIndex is an ArchiveIndex kept entirely in memory, for tests and short-lived tools.

func NewInMemoryIndex

func NewInMemoryIndex() *InMemoryIndex

NewInMemoryIndex returns an empty index.

func (*InMemoryIndex) Abandon

func (i *InMemoryIndex) Abandon(name string, within time.Duration) (bool, error)

Abandon implements ClaimIndex.

func (*InMemoryIndex) AddObjects

func (i *InMemoryIndex) AddObjects(archive string, records []IndexRecord) error

AddObjects implements ClaimIndex.

func (*InMemoryIndex) BeginSession

func (i *InMemoryIndex) BeginSession(s *backup.Session) error

BeginSession implements SessionIndex.

func (*InMemoryIndex) Claims

func (i *InMemoryIndex) Claims(session string) ([]Claim, error)

Claims implements ClaimIndex.

func (*InMemoryIndex) Close

func (i *InMemoryIndex) Close() error

Close implements io.Closer.

func (*InMemoryIndex) CommitSession

func (i *InMemoryIndex) CommitSession(id string) error

CommitSession implements SessionIndex.

func (*InMemoryIndex) CountObjects

func (i *InMemoryIndex) CountObjects() (uint64, uint64, error)

CountObjects reports the indexed and the distinct object count.

func (*InMemoryIndex) DeleteArchives

func (i *InMemoryIndex) DeleteArchives(names []string) error

DeleteArchives implements pack.ArchiveIndex.

func (*InMemoryIndex) EndSession

func (i *InMemoryIndex) EndSession(id string) ([]string, error)

EndSession implements SessionIndex.

func (*InMemoryIndex) FinalizeArchive

func (i *InMemoryIndex) FinalizeArchive(name string, within time.Duration, created time.Time) error

FinalizeArchive implements ClaimIndex.

func (*InMemoryIndex) GetSession

func (i *InMemoryIndex) GetSession(id string) (*backup.Session, error)

GetSession implements SessionIndex.

func (*InMemoryIndex) Holds

func (i *InMemoryIndex) Holds(ref *proto.Ref, session string) (bool, error)

Holds implements ClaimIndex.

func (*InMemoryIndex) IndexArchive

func (i *InMemoryIndex) IndexArchive(archive ArchiveInfo, index IndexFile) error

IndexArchive implements pack.ArchiveIndex.

func (*InMemoryIndex) ListSessions

func (i *InMemoryIndex) ListSessions() ([]*backup.Session, error)

ListSessions implements SessionIndex.

func (*InMemoryIndex) LocateCopies

func (i *InMemoryIndex) LocateCopies(refs []*proto.Ref, scope Scope) (map[string][]IndexLocation, error)

LocateCopies implements pack.ArchiveIndex.

func (*InMemoryIndex) LocateObject

func (i *InMemoryIndex) LocateObject(ref *proto.Ref, scope Scope, exclude ...string) (IndexLocation, error)

LocateObject implements pack.ArchiveIndex.

func (*InMemoryIndex) LocateTombstones

func (i *InMemoryIndex) LocateTombstones(refs []*proto.Ref, scope Scope) (map[string][]IndexLocation, error)

LocateTombstones implements pack.ArchiveIndex.

func (*InMemoryIndex) LookupArchive

func (i *InMemoryIndex) LookupArchive(archive string) (ArchiveInfo, bool, error)

LookupArchive implements pack.ArchiveIndex.

func (*InMemoryIndex) Lost

func (i *InMemoryIndex) Lost(session string) ([]*proto.Ref, error)

Lost implements ClaimIndex.

func (*InMemoryIndex) OpenArchive

func (i *InMemoryIndex) OpenArchive(name, session string) error

OpenArchive implements ClaimIndex.

func (*InMemoryIndex) PendingArchives

func (i *InMemoryIndex) PendingArchives(id string) ([]string, error)

PendingArchives implements SessionIndex.

func (*InMemoryIndex) TouchSession

func (i *InMemoryIndex) TouchSession(id string, at time.Time) error

TouchSession implements SessionIndex.

type IndexFile

type IndexFile []IndexRecord

IndexFile is an archive's index; a stored one is sorted by Sum.

func (IndexFile) Len

func (idx IndexFile) Len() int

Len implements sort.Interface.

func (IndexFile) Less

func (idx IndexFile) Less(i, j int) bool

Less implements sort.Interface.

func (*IndexFile) ReadFrom

func (idx *IndexFile) ReadFrom(reader io.Reader) (int64, error)

ReadFrom implements io.ReaderFrom.

func (IndexFile) Swap

func (idx IndexFile) Swap(i, j int)

Swap implements sort.Interface.

func (IndexFile) WriteTo

func (idx IndexFile) WriteTo(writer io.Writer) (int64, error)

WriteTo implements io.WriterTo.

type IndexLocation

type IndexLocation struct {
	Archive string
	Record  IndexRecord
}

IndexLocation is where an ArchiveIndex found an object.

type IndexRecord

type IndexRecord struct {
	Sum    [proto.HashSize]byte
	Offset uint32
	Length uint32
	Type   uint32
	// CarriedTime and CarriedOffset are the version a rewrite carried the
	// record over with; a zero CarriedTime means the record has its
	// archive's version.
	CarriedTime   int64
	CarriedOffset uint32
}

IndexRecord locates one object in its archive.

func (IndexRecord) Carry

func (r IndexRecord) Carry(v Version) IndexRecord

Carry returns the record, at a new place, keeping the version v.

func (IndexRecord) Version

func (r IndexRecord) Version(created time.Time) Version

Version is the version of the record in an archive created at created.

type InfoLister

type InfoLister interface {
	ListInfo(extension string) ([]ListedFile, error)
}

InfoLister is implemented by an ArchiveStorage that lists its files with their sizes and modification times, as List names them, in one listing.

type ListedFile

type ListedFile struct {
	Name     string
	Size     int64
	Modified time.Time
}

ListedFile is a file a listing found, with its size and when it was stored.

func ListInfo

func ListInfo(storage ArchiveStorage, extension string) ([]ListedFile, error)

ListInfo lists the storage's files with their sizes and modification times, through its InfoLister when it is one and by opening each file otherwise; a wrapping ArchiveStorage implements InfoLister with it.

type ListedOpener

type ListedOpener interface {
	OpenListed(file ListedFile) (File, error)
}

ListedOpener is implemented by an ArchiveStorage that opens a file a listing found from what the listing said of it, without asking the storage again.

type PackOption

type PackOption func(p *packOptions)

PackOption configures NewPackStorage.

func WithArchiveIndex

func WithArchiveIndex(index ArchiveIndex) PackOption

WithArchiveIndex is the index of archives and sessions; required.

func WithArchiveObserver

func WithArchiveObserver(observer ArchiveObserver) PackOption

WithArchiveObserver tells observer about every archive stored and deleted.

func WithArchiveStorage

func WithArchiveStorage(storage ArchiveStorage) PackOption

WithArchiveStorage is where the archives live; required.

func WithAtRestKey

func WithAtRestKey(key *AtRestKey) PackOption

WithAtRestKey seals every payload written from now on under the key and opens the ones sealed under any key of its keyset; records written without a key stay readable. A rewrite re-seals what it opens under the primary key.

func WithClaims

func WithClaims(open, grace time.Duration) PackOption

WithClaims sets how long an archive of a session stays open, and how much longer its claim holds while the archive is closed and verified, when the index is a ClaimIndex. A commit waits up to their sum for an archive another process holds open.

func WithCloseBeforeRead

func WithCloseBeforeRead(do bool) PackOption

WithCloseBeforeRead finalizes an open archive before an object is read from it, for storages that cannot read an upload in flight.

func WithCompaction

func WithCompaction(config CompactionConfig) PackOption

WithCompaction tunes Compact.

func WithIdleFinalize

func WithIdleFinalize(d time.Duration) PackOption

WithIdleFinalize finalizes a session's open archive after it has not been written to for the given duration.

func WithIndexCache

func WithIndexCache(dir string) PackOption

WithIndexCache keeps a copy of every archive index the store reads under dir, so reading it again costs no storage request. A copy keeps the original's creation time, which versions the archive's records.

func WithLogger

func WithLogger(logger *slog.Logger) PackOption

WithLogger is where the store reports; the default is slog.Default.

func WithMaxParallel

func WithMaxParallel(max uint) PackOption

WithMaxParallel bounds the archives open for writing at once.

func WithMaxSize

func WithMaxSize(max uint64) PackOption

WithMaxSize is the size at which an open archive is finalized.

func WithMetadataCache

func WithMetadataCache(cache backup.ObjectStore) PackOption

WithMetadataCache keeps commits, trees and files in cache as well.

func WithOwned

func WithOwned(closers ...io.Closer) PackOption

WithOwned hands the store things it closes after itself, such as an archive index nobody else holds.

func WithSessionLease

func WithSessionLease(d time.Duration) PackOption

WithSessionLease ends sessions that have not written for the given duration, dropping what they did not commit.

type PackStorage

type PackStorage struct {
	// contains filtered or unexported fields
}

PackStorage stores objects in archives. Writes made under a session (see BeginSession) go to that session's own archives and are visible only to the session until it stores a commit; writes without a session go to root archives and are visible at once.

func NewPackStorage

func NewPackStorage(options ...PackOption) (*PackStorage, error)

NewPackStorage builds a store from options; WithArchiveStorage and WithArchiveIndex are required.

func (*PackStorage) AdvanceHead

func (ps *PackStorage) AdvanceHead(object *proto.Object) error

AdvanceHead implements backup.HeadKeeper. Checkpoints never become heads.

func (*PackStorage) BeginSession

func (ps *PackStorage) BeginSession(ctx context.Context, s *backup.Session) (context.Context, error)

BeginSession registers the session, assigning an id when it has none, and returns a context under which writes belong to it.

func (*PackStorage) Close

func (ps *PackStorage) Close() error

Close implements backup.ObjectStore.

func (*PackStorage) Collect

func (ps *PackStorage) Collect(ctx context.Context, opts CollectOptions) (*CollectReport, error)

Collect marks every object reachable from a live commit or pin, records the result beside each archive and rewrites archives whose dead share justifies it. An unreachable object gets a tombstone, and the next generation drops it if it is still unreachable and no session that relied on it took the tombstone back as it committed. The context must not carry a session.

func (*PackStorage) Compact

func (ps *PackStorage) Compact(ctx context.Context) (*CompactReport, error)

Compact merges the small committed archives into archives at the root once they add up to a batch, or are more than MinimumCandidates, and returns what it did when it is done. What it writes is too large to be merged again.

func (*PackStorage) Count

func (ps *PackStorage) Count() (uint64, uint64, error)

Count implements backup.Counter.

func (*PackStorage) Delete

func (ps *PackStorage) Delete(ctx context.Context, ref *proto.Ref) error

Delete implements backup.ObjectStore.

func (*PackStorage) EndSession

func (ps *PackStorage) EndSession(ctx context.Context) error

EndSession drops what the context's session has not committed.

func (*PackStorage) Erase

func (ps *PackStorage) Erase(ctx context.Context, ref *proto.Ref) error

Erase implements backup.Eraser.

func (*PackStorage) EscrowedKeys

func (ps *PackStorage) EscrowedKeys(context.Context) ([]backup.EscrowedKey, error)

EscrowedKeys implements backup.KeyEscrow.

func (*PackStorage) Flush

func (ps *PackStorage) Flush() error

Flush finalizes the open archives of every session.

func (*PackStorage) Get

func (ps *PackStorage) Get(ctx context.Context, ref *proto.Ref) (*proto.Object, error)

Get implements backup.ObjectStore.

func (*PackStorage) Has

func (ps *PackStorage) Has(ctx context.Context, ref *proto.Ref) (bool, error)

Has reports whether the store holds a copy of the object that the caller may rely on. The newest record of the object decides: a tombstone newer than every copy makes it absent, unless an un-tombstone is newer still.

func (*PackStorage) HasAll

func (ps *PackStorage) HasAll(ctx context.Context, refs []*proto.Ref) ([]bool, error)

HasAll implements backup.HasAller: Has for each of refs, from one lookup of their tombstones and one of their copies.

func (*PackStorage) Heads

func (ps *PackStorage) Heads() ([]Head, error)

Heads lists the head of every set that has committed.

func (*PackStorage) Holds

func (ps *PackStorage) Holds(_ context.Context, commits []*proto.Ref) ([]bool, error)

Holds implements backup.Reviver.

func (*PackStorage) LastCollected

func (ps *PackStorage) LastCollected() (time.Time, error)

LastCollected returns when the newest collection marked the store, zero when none has.

func (*PackStorage) LocateRecords

func (ps *PackStorage) LocateRecords(ctx context.Context, refs []*proto.Ref) ([]*proto.LocatedRun, error)

LocateRecords implements backup.RecordLocator for objects in archives the storage can sign ranges of. A commit, tree or file, and any object the cache holds, is never located: the store serves those itself.

func (*PackStorage) LookupSession

func (ps *PackStorage) LookupSession(_ context.Context, id string) (*backup.Session, error)

LookupSession returns a live session by id.

func (*PackStorage) Open

func (ps *PackStorage) Open() error

Open implements backup.ObjectStore.

func (*PackStorage) PendingPlans

func (ps *PackStorage) PendingPlans() ([]uint64, error)

PendingPlans lists the generations whose published plan RewritePlan has not run yet, oldest first.

func (*PackStorage) PurgeQuarantine

func (ps *PackStorage) PurgeQuarantine(period time.Duration, now time.Time) (int, error)

PurgeQuarantine deletes the archives retired longer than period ago and returns how many it deleted.

func (*PackStorage) Put

func (ps *PackStorage) Put(ctx context.Context, object *proto.Object) error

Put implements backup.ObjectStore.

func (*PackStorage) PutEscrowedKey

func (ps *PackStorage) PutEscrowedKey(ctx context.Context, key backup.EscrowedKey) error

PutEscrowedKey implements backup.KeyEscrow. Each copy is a file of its own under keys/<key id>/, named by its content.

func (*PackStorage) ReadRecords

func (ps *PackStorage) ReadRecords(ctx context.Context, refs []*proto.Ref) ([]*proto.Object, error)

ReadRecords implements backup.RecordReader with one range read per run of neighbouring records.

func (*PackStorage) Repair

func (ps *PackStorage) Repair(ctx context.Context) (*RepairReport, error)

Repair scrubs the store and rewrites every archive holding damage, leaving the corrupt records behind. Objects an intact copy exists for survive the pass; the rest are gone from the store, so a later backup stores them again instead of finding them present.

func (*PackStorage) RestoreLeases

func (ps *PackStorage) RestoreLeases(_ context.Context) ([]*proto.Ref, error)

RestoreLeases implements backup.RestoreLeaser over the sessions whose lease has not run out.

func (*PackStorage) RestoreQuarantined

func (ps *PackStorage) RestoreQuarantined(name string) error

RestoreQuarantined puts a retired archive back, where the store picks it up as committed.

func (*PackStorage) Revive

func (ps *PackStorage) Revive(ctx context.Context, commits []*proto.Ref, dryRun bool) ([]backup.Revival, error)

Revive implements backup.Reviver. A revival is an un-tombstone of the commit newer than its tombstone, which a collection takes as a root; what the commit reaches gets un-tombstones as a session's skipped refs do. It holds off this process's collections while it runs and refuses while a published plan waits for its rewrite.

func (*PackStorage) Revived

func (ps *PackStorage) Revived(_ context.Context, commit *proto.Ref) (bool, error)

Revived implements backup.Reviver.

func (*PackStorage) RewritePlan

func (ps *PackStorage) RewritePlan(ctx context.Context) (*RewriteReport, error)

RewritePlan runs the sweeps collections published, oldest first, and deletes each plan once its archives are rewritten. A run that stops part way leaves the plan for the next one, which skips the archives already retired.

func (*PackStorage) Scrub

func (ps *PackStorage) Scrub(ctx context.Context) (*ScrubReport, error)

Scrub rehashes every object in every archive and reports the ones that no longer match their ref or break their archive's predecessor chain. It never modifies the store.

func (*PackStorage) SessionLease

func (ps *PackStorage) SessionLease() time.Duration

SessionLease implements backup.Leased: zero when sessions never expire.

func (*PackStorage) Sweep

func (ps *PackStorage) Sweep(now time.Time)

Sweep finalizes archives idle past the idle timeout and ends sessions whose lease ran out, as of now. Nothing runs it but the caller.

func (*PackStorage) Walk

Walk implements backup.ObjectStore.

func (*PackStorage) WalkArchives

func (ps *PackStorage) WalkArchives(ctx context.Context, t proto.ObjectType, skip func(string) bool, fn backup.ObjectReceiver) ([]string, error)

WalkArchives implements backup.ArchiveWalker. An archive still open to writes is walked but not named.

func (*PackStorage) WalkHeaders

func (ps *PackStorage) WalkHeaders(ctx context.Context, t proto.ObjectType, fn func(*proto.ObjectHeader) error) error

WalkHeaders implements backup.HeaderWalker over the committed archives.

type Placement

type Placement struct {
	Session string
}

Placement is the decoded prefix of an archive name: the session that wrote it and has not committed (<session>/<id>), or none for a committed archive at the root.

func ParsePlacement

func ParsePlacement(name string) Placement

ParsePlacement decodes the prefix of an archive name.

func (Placement) Dir

func (p Placement) Dir() string

Dir is the prefix under which archives of this placement are written.

type RangeSigner

type RangeSigner interface {
	// SignRange returns a GET whose body is bytes [offset, offset+length)
	// of name, good for ttl, carrying whatever headers the request must
	// repeat. It returns ErrNoSignedURL when it cannot sign.
	SignRange(ctx context.Context, name string, offset, length int64, ttl time.Duration) (*proto.Location, error)
}

A RangeSigner is an ArchiveStorage whose files can be fetched directly, one byte range at a time.

type RepairReport

type RepairReport struct {
	// Scrub is the pass that found the damage.
	Scrub *ScrubReport
	// Archives is how many archives were rewritten.
	Archives uint64
	// Recovered names the objects the store still holds after the pass,
	// because another archive had an intact copy.
	Recovered []*proto.Ref
	// Lost names the objects no archive holds any more. A restore of one
	// leaves a hole until a backup of a source that still has it puts it
	// back.
	Lost []*proto.Ref
}

RepairReport summarises one repair pass.

type RewriteReport

type RewriteReport struct {
	Plans            int
	Swept            int
	ReclaimedObjects uint64
	ReclaimedBytes   uint64
	CopiedBytes      uint64
}

RewriteReport summarizes the plans one RewritePlan ran.

type Scope

type Scope struct {
	Session string
}

Scope filters LocateObject: the committed archives plus the pending archives of Session.

func ScopeOf

func ScopeOf(ctx context.Context) Scope

ScopeOf is the scope of a context: the session it carries, else the committed archives alone.

func (Scope) Visible

func (s Scope) Visible(a ArchiveInfo) bool

Visible reports whether an archive is served to this scope.

type ScrubFailure

type ScrubFailure struct {
	Archive string
	Ref     *proto.Ref
	Err     error
}

ScrubFailure names one object that failed verification.

type ScrubReport

type ScrubReport struct {
	Archives uint64
	Objects  uint64
	Bytes    uint64
	// Corrupt lists the objects whose stored bytes do not hash to their
	// header or whose header names the wrong predecessor.
	Corrupt []ScrubFailure
	// Sealed counts the objects per hex at-rest key id, the ones stored
	// in the clear under the empty string. A rotation is over once no
	// retired key id is left here.
	Sealed map[string]uint64
}

ScrubReport summarises one pass over every stored object.

type SessionIndex

type SessionIndex interface {
	BeginSession(s *backup.Session) error
	TouchSession(id string, at time.Time) error
	GetSession(id string) (*backup.Session, error)
	ListSessions() ([]*backup.Session, error)
	// EndSession drops the session and its uncommitted archives, returning
	// the names of the archives dropped.
	EndSession(id string) ([]string, error)
	// PendingArchives names the session's archives that a commit would
	// make committed.
	PendingArchives(id string) ([]string, error)
	// CommitSession flips the session's pending archives to committed and
	// forgets its lost ones.
	CommitSession(id string) error
}

SessionIndex keeps the live sessions of a store, so any node can tell a session id from a stale one and a janitor can expire the dead.

type SharedClock

type SharedClock interface {
	SharedNow(ctx context.Context) (time.Time, error)
}

SharedClock is implemented by an index whose processes share one clock. The store stamps when a session begins and when a collection completes by it, because a collection compares the two.

type Version

type Version struct {
	Time   time.Time
	Offset uint32
}

Version orders the records of one object: by the bucket's creation time of the index file of the archive that first stored the record, then by the record's offset within that archive. A rewrite keeps the version of what it moves.

func (Version) Before

func (v Version) Before(o Version) bool

Before reports whether v is older than o.

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL