fragment

README

Fragment

The fragment package provides journal fragment management for Gazette brokers. It handles the mapping of journal offsets to protocol.Fragments and manages local/remote journal content through various storage backends.

What it does

• Fragment Lifecycle: Manages fragments from creation (Spool) through persistence to storage backends
• Storage Abstraction: Provides unified interface for file://, s3://, gcs://, and azure:// storage schemes
• Fragment Indexing: Maintains queryable indexes of local and remote fragments with offset-to-fragment mapping
• Replication Support: Handles fragment spooling during broker replication operations

Architecture

Core Types

• Fragment: Wrapper around protocol.Fragment with optional local file backing
• Spool: In-progress fragment being written during replication, with compression and checksumming
• Index: Queryable index mapping journal offsets to best-covering fragments
• CoverSet: Ordered collection of fragments with overlap resolution
• Persister: Asynchronous background service for persisting completed spools to storage

Storage Backends

The package abstracts over multiple storage providers through a common backend interface:

• File System (store_fs.go): Local filesystem storage
• Amazon S3 (store_s3.go): AWS S3 compatible storage
• Google Cloud Storage (store_gcs.go): GCS object storage
• Azure Blob (store_azure.go): Azure Blob Storage

Each backend implements operations for signing URLs, checking existence, opening, persisting, listing, and removing fragments.

Fragment Spooling

During journal replication:

1. Spool Creation: New Spool created for incoming writes
2. Content Buffering: Writes accumulated with compression and SHA1 checksumming
3. Commit/Rollback: Transactional operations extend or revert fragment content
4. Completion: Finished spools queued for background persistence

Persistence

The Persister manages asynchronous fragment persistence:

• Primary replicas attempt immediate persistence
• Secondary replicas queue spools for later persistence
  • In the common case this is a cheap existence check
  • If the primary broker has crashed, it ensures the fragment is persisted
• Failed persistence operations are retried with backoff
• Supports multiple storage backends

Key Operations

• Fragment Queries: Map journal offsets to covering fragments
• Spool Management: Handle transactional fragment construction
• Store Operations: Abstract storage operations across backends
• Index Maintenance: Keep fragment indexes current with storage state

Usage

The fragment package is used internally by broker services for:

• Serving read requests by locating appropriate fragments
• Managing fragment creation during journal replication
• Persisting completed fragments to configured storage
• Maintaining indexes for efficient fragment lookup

Documentation

Overview

Package fragment is a broker-only package concerned with the mapping of journal offsets to protocol.Fragments, and from there to corresponding local or remote journal content.

It implements file-like operations over the FragmentStore schemes supported by Gazette, such as listing, opening, signing, persisting, and removing fragments. See FileStoreConfig, S3StoreConfig, and GSStoreConfig for further configuration of store operations.

The package implements a Fragment wrapper type which composes a protocol.Fragment with an open file descriptor, and an Index over local or remote Fragments which maps a journal offset to a best-covering Fragment.

Spool is a Fragment which is in the process of being constructed from an ongoing broker Replicate RPC. It is the transactional "memory" of brokers which are participating in the replication of a journal. Once closed, or "rolled", a Spool Fragment is persisted to its configured FragmentStore by a Persister.

Index

• func Open(ctx context.Context, fragment pb.Fragment) (io.ReadCloser, error)
• func Persist(ctx context.Context, spool Spool, spec *pb.JournalSpec, isExiting bool) error
• func Remove(ctx context.Context, fragment pb.Fragment) error
• func SignGetURL(fragment pb.Fragment, d time.Duration) (string, error)
• type CoverSet
  • func CoverSetDifference(a, b CoverSet) CoverSet
  • func WalkAllStores(ctx context.Context, name pb.Journal, allStores []pb.FragmentStore) (CoverSet, error)
  • func (s CoverSet) Add(fragment Fragment) (CoverSet, bool)
  • func (s CoverSet) BeginOffset() int64
  • func (s CoverSet) EndOffset() int64
  • func (s CoverSet) LongestOverlappingFragment(offset int64) (ind int, found bool)
• type File
• type Fragment
• type Index
  • func NewIndex(ctx context.Context) *Index
  • func (fi *Index) FirstRefreshCh() <-chan struct{}
  • func (fi *Index) Inspect(ctx context.Context, callback func(CoverSet) error) error
  • func (fi *Index) Query(ctx context.Context, req *pb.ReadRequest) (*pb.ReadResponse, File, error)
  • func (fi *Index) ReplaceRemote(set CoverSet)
  • func (fi *Index) SpoolCommit(frag Fragment)
  • func (fi *Index) Summary() (int64, int64, int64)
• type Persister
  • func NewPersister(ks *keyspace.KeySpace) *Persister
  • func (p *Persister) Finish()
  • func (p *Persister) Serve()
  • func (p *Persister) SpoolComplete(spool Spool, primary bool)
• type Spool
  • func NewSpool(journal pb.Journal, observer SpoolObserver) Spool
  • func (s *Spool) Apply(r *pb.ReplicateRequest, primary bool) (pb.ReplicateResponse, error)
  • func (s *Spool) MustApply(r *pb.ReplicateRequest)
  • func (s *Spool) Next() pb.Fragment
  • func (s Spool) String() string
• type SpoolObserver

Functions

func Open

func Open(ctx context.Context, fragment pb.Fragment) (io.ReadCloser, error)

Open a Reader of the Fragment on the store. The returned ReadCloser does not perform any applicable client-side decompression, but does request server decompression in the case of GZIP_OFFLOAD_DECOMPRESSION.

func Persist

func Persist(ctx context.Context, spool Spool, spec *pb.JournalSpec, isExiting bool) error

Persist a Spool to the JournalSpec's store. If the Spool Fragment is already present, this is a no-op. If the Spool has not been compressed incrementally, it will be compressed before being persisted. If `isExiting` and the preferred store returns an AuthZ error, it will try subsequent stores.

func Remove

func Remove(ctx context.Context, fragment pb.Fragment) error

Remove `fragment` from its BackingStore.

func SignGetURL

func SignGetURL(fragment pb.Fragment, d time.Duration) (string, error)

SignGetURL returns a URL authenticating the bearer to perform a GET operation of the Fragment for the provided Duration from the current time.

Types

type CoverSet

type CoverSet []Fragment

CoverSet maintains Fragments ordered on |Begin| and |End|, with the invariant that no Fragment is fully overlapped by another Fragment in the set (though it may be overlapped by a combination of other Fragments). Intuitively, CoverSet represents the set of offsets which are "covered" by a collection of Fragments, and is able to map, for each byte offset, a "best" covering Fragment. It employs a heuristic of preferring larger fragments (and will replace spans of overlapped smaller fragments). An implication of its invariant is that no two Fragments have the same |Begin| or |End| (as that would imply an overlap). Both are monotonically increasing in the set: set[0].Begin represents the minimum offset, and set[len(set)-1].End represents the maximum offset.

func CoverSetDifference

func CoverSetDifference(a, b CoverSet) CoverSet

CoverSetDifference returns the subset of Fragments in |a| which cover byte offsets not also covered by Fragments in |b|.

func WalkAllStores

func WalkAllStores(ctx context.Context, name pb.Journal, allStores []pb.FragmentStore) (CoverSet, error)

WalkAllStores enumerates Fragments from each of |stores| into the returned CoverSet, or returns an encountered error.

func (CoverSet) Add

func (s CoverSet) Add(fragment Fragment) (CoverSet, bool)

Add the Fragment to the CoverSet. The CoverSet is returned, along with an indication of whether an offset span was updated to reflect Fragment. All updates occur in-place.

func (CoverSet) BeginOffset

func (s CoverSet) BeginOffset() int64

BeginOffset returns the first (lowest) Begin offset of any Fragment in the CoverSet.

func (CoverSet) EndOffset

func (s CoverSet) EndOffset() int64

EndOffset returns the last (largest) End offset of any Fragment in the set.

func (CoverSet) LongestOverlappingFragment

func (s CoverSet) LongestOverlappingFragment(offset int64) (ind int, found bool)

LongestOverlappingFragment finds and returns the index |ind| of the Fragment covering |offset| which also has the most content following |offset|. If no fragment covers |offset|, the index of the next Fragment beginning after |offset| is returned (which may be beyond the current CoverSet range). |found| indicates whether an overlapping Fragment was found.

type File

type File interface {
	io.ReaderAt
	io.Seeker
	io.WriterAt
	io.Writer
	io.Closer
}

File is the subset of os.File used in backing Fragments with local files.

type Fragment

type Fragment struct {
	protocol.Fragment
	// Local uncompressed file of the Fragment, or nil iff the Fragment is remote.
	File File
}

Fragment wraps the protocol.Fragment type with a nil-able backing local File.

type Index

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

Index maintains a queryable index of local and remote journal Fragments.

func NewIndex

func NewIndex(ctx context.Context) *Index

NewIndex returns a new, empty Index.

func (*Index) FirstRefreshCh

func (fi *Index) FirstRefreshCh() <-chan struct{}

FirstRefreshCh returns a channel which signals if the Index as been refreshed with a remote store(s) listing at least once.

func (*Index) Inspect

func (fi *Index) Inspect(ctx context.Context, callback func(CoverSet) error) error

Inspect invokes the callback with a snapshot of all fragments in the Index. The callback must not modify the CoverSet, and during callback invocation no changes will be made to it. If an initial refresh of remote fragment store(s) hasn't yet been applied, Inspect will first block until it does (or context cancellation).

func (*Index) Query

func (fi *Index) Query(ctx context.Context, req *pb.ReadRequest) (*pb.ReadResponse, File, error)

Query the Index for a Fragment matching the ReadRequest.

func (*Index) ReplaceRemote

func (fi *Index) ReplaceRemote(set CoverSet)

ReplaceRemote replaces all remote Fragments in the index with |set|.

func (*Index) SpoolCommit

func (fi *Index) SpoolCommit(frag Fragment)

SpoolCommit adds local Spool Fragment |frag| to the index.

func (*Index) Summary

func (fi *Index) Summary() (int64, int64, int64)

Summary returns the [Begin, End) offset range of all Fragments in the index, and the persisted ModTime of the last Fragment (or zero, if it's local).

type Persister

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

Persister asynchronously persists completed Fragments to their backing pb.FragmentStore.

func NewPersister

func NewPersister(ks *keyspace.KeySpace) *Persister

NewPersister returns an empty, initialized Persister.

func (*Persister) Finish

func (p *Persister) Finish()

func (*Persister) Serve

func (p *Persister) Serve()

func (*Persister) SpoolComplete

func (p *Persister) SpoolComplete(spool Spool, primary bool)

type Spool

type Spool struct {
	// Fragment at time of last commit.
	Fragment
	// FirstAppendTime is the UTC Time of the first commit of the Spool Fragment.
	FirstAppendTime time.Time
	// Registers of the journal.
	Registers pb.LabelSet
	// contains filtered or unexported fields
}

Spool is a Fragment which is in the process of being created, backed by a local *os.File. As commits occur and the file extent is updated, the Spool Fragment is also updated to reflect the new committed extent. At all times, the Spool Fragment is a consistent, valid Fragment.

func NewSpool

func NewSpool(journal pb.Journal, observer SpoolObserver) Spool

NewSpool returns an empty Spool of |journal|.

func (*Spool) Apply

func (s *Spool) Apply(r *pb.ReplicateRequest, primary bool) (pb.ReplicateResponse, error)

Apply the ReplicateRequest to the Spool, returning any encountered error.

func (*Spool) MustApply

func (s *Spool) MustApply(r *pb.ReplicateRequest)

MustApply applies the ReplicateRequest, and panics if a !OK status is returned or error occurs. MustApply is a convenience for cases such as rollbacks, where the request is derived from the Spool itself and cannot reasonably fail.

func (*Spool) Next

func (s *Spool) Next() pb.Fragment

Next returns the next Fragment which can be committed by the Spool.

func (Spool) String

func (s Spool) String() string

String returns a debugging representation of the Spool.

type SpoolObserver

type SpoolObserver interface {
	// SpoolCommit is called when the Spool Fragment is extended.
	SpoolCommit(Fragment)
	// SpoolComplete is called when the Spool has been completed.
	SpoolComplete(_ Spool, primary bool)
}

SpoolObserver is notified of important events in the Spool lifecycle.