checkpoint

package
v0.7.2 Latest Latest
Warning

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

 Go to latest Published: Jul 19, 2026 License: MIT Imports: 13 Imported by: 0

Documentation

Index

Constants

View Source 
const SavepointPrefix = "savepoints/"

SavepointPrefix is the blobstore key prefix under which savepoints live.

Variables

View Source 
var ErrBlobNotFound = errors.New("checkpoint: blob not found")

ErrBlobNotFound is returned by Blobstore.Get for a missing key.

Functions

func ArchiveCheckpoint added in v0.5.0 

func ArchiveCheckpoint(src Storage, id string, w io.Writer) error

ArchiveCheckpoint writes a self-contained tar of checkpoint id — its JSON record plus a real copy of every per-owner state directory — to w. Unlike the live checkpoint (Pebble hard-links on one filesystem), the archive is portable: it can be moved to another volume, host, or object store and extracted anywhere.

func CreateSavepoint added in v0.5.0 

func CreateSavepoint(src Storage, bs Blobstore, label string) (string, error)

CreateSavepoint archives the latest completed checkpoint in src and stores it in bs under the given label. This is the promotion half of a stop-with-savepoint: the job has drained and written its final checkpoint, and here that checkpoint is copied into durable, named, never-GC'd storage the operator can restart from.

Returns the checkpoint ID that was promoted. Errors if src has no completed checkpoint (nothing to promote).

func ExtractCheckpoint added in v0.5.0 

func ExtractCheckpoint(dst Storage, r io.Reader) (string, error)

ExtractCheckpoint reads an archive produced by ArchiveCheckpoint and materializes it into dst as a completed checkpoint: state files are written under dst.StateDir(id) and the record is saved so that dst.LoadLatestCompleted() returns it. Returns the checkpoint ID.

func ListSavepoints added in v0.5.0 

func ListSavepoints(bs Blobstore) ([]string, error)

ListSavepoints returns the labels of all savepoints in bs.

func RestoreSavepoint added in v0.5.0 

func RestoreSavepoint(dst Storage, bs Blobstore, label string) (string, error)

RestoreSavepoint extracts the named savepoint from bs into dst, making it dst's latest completed checkpoint so the engine resumes from it on the next Execute. Returns the restored checkpoint ID.

func SavepointKey added in v0.5.0 

func SavepointKey(label string) string

SavepointKey returns the blobstore key for a named savepoint.

Types

type Blobstore added in v0.5.0 

type Blobstore interface {
	// Put stores the contents of r under key, overwriting any existing
	// blob. It reads r to EOF.
	Put(key string, r io.Reader) error
	// Get opens the blob at key. Returns ErrBlobNotFound if absent. The
	// caller closes the reader.
	Get(key string) (io.ReadCloser, error)
	// Exists reports whether a blob is present.
	Exists(key string) (bool, error)
	// List returns keys with the given prefix.
	List(prefix string) ([]string, error)
	// Delete removes a blob; absent keys are not an error.
	Delete(key string) error
}

Blobstore is a minimal object-store abstraction used for savepoints (and, later, object-store checkpoints). Keys are '/'-separated paths. It exists so the durability mechanic — archive a checkpoint, put it, get it, extract it — is written once against an interface: the filesystem implementation below serves local Docker today, and an S3/GCS adapter drops in for multi-host durability (plan phase P6) without touching the archive or savepoint code.

type CheckpointData 

type CheckpointData struct {
	ID        string            `json:"id"`
	Timestamp time.Time         `json:"timestamp"`
	Operators map[string][]byte `json:"operators"` // operator index -> state bytes
	Source    map[string][]byte `json:"source"`    // source-specific offset data
	Status    Status            `json:"status,omitempty"`
	TxnID     string            `json:"txn_id,omitempty"`     // sink transactional id (diagnostics)
	StateDirs map[string]string `json:"state_dirs,omitempty"` // ownerID -> relative path
}

CheckpointData holds the complete state of a pipeline at a point in time. It includes the state of each stateful operator (by index) and the source offset (if applicable). StateDirs maps owner IDs to relative paths of native (hard-linked) state directories, used by PebbleBackend for O(1)-delta checkpoints.

func (*CheckpointData) Completed added in v0.4.0 

func (d *CheckpointData) Completed() bool

Completed reports whether this checkpoint is safe to restore from.

type Coordinator added in v0.4.0 

type Coordinator struct {
	Storage Storage
	TxnID   string

	// CommitSink / AbortSink are the sink's transaction controls.
	CommitSink func(ctx context.Context, id string) error
	AbortSink  func(ctx context.Context, id string) error

	// CommitOffsets commits source offsets externally after a
	// checkpoint completes. Advisory: errors are logged, not fatal.
	// Nil when the source has no external commit.
	CommitOffsets func(ctx context.Context, offsets []byte) error

	// Hook is a test-only seam fired after each protocol step.
	Hook func(step Step, id string) HookAction

	// OnCompleted, if set, is called with the checkpoint ID once it is
	// promoted to StatusCompleted. Observability only (progress signal
	// for supervisors); must not block.
	OnCompleted func(id string)
	// contains filtered or unexported fields
}

Coordinator drives the two-phase checkpoint protocol: 

offsets (barrier injection) ─┐
operator state (barrier tap) ─┼─► sink prepared ─► persist(prepared)
sink ack (PreCommit done)    ─┘        │
                                       ▼
              sink.Commit ─► persist(completed) ─► source.CommitOffsets

Any failure before the sink commit aborts the sink transaction and fails the pipeline; recovery restores the latest completed checkpoint (or promotes a prepared one whose transaction provably committed — see the engine's recovery decision table).

func NewCoordinator added in v0.4.0 

func NewCoordinator(storage Storage, txnID string) *Coordinator

NewCoordinator creates a coordinator. Call Start before wiring it into a pipeline and Stop after the pipeline finishes.

func (*Coordinator) Fatal added in v0.4.0 

func (c *Coordinator) Fatal() <-chan error

Fatal returns a channel that receives the first unrecoverable coordination error (buffered; never closed).

func (*Coordinator) OnBarrierInjected added in v0.4.0 

func (c *Coordinator) OnBarrierInjected(id string, offsets []byte)

OnBarrierInjected records the barrier-aligned source offsets for a checkpoint at the moment its barrier entered the stream.

func (*Coordinator) OnSinkPrepared added in v0.4.0 

func (c *Coordinator) OnSinkPrepared(id string, sinkErr error)

OnSinkPrepared is called by the sink (via its notifier) once all pre-barrier output for this checkpoint is staged in the open transaction. It queues finalization on the coordinator worker — never finalize inline: this runs on the sink's Write goroutine, which must stay free to be unblocked by CommitSink.

func (*Coordinator) OnStateSnapshot added in v0.4.0 

func (c *Coordinator) OnStateSnapshot(id string, state map[string][]byte, stateDirs map[string]string)

OnStateSnapshot records operator/worker state captured when the barrier reached the pre-sink tap. stateDirs carries native (hard-link) state directory references for Checkpointable backends.

func (*Coordinator) Start added in v0.4.0 

func (c *Coordinator) Start(ctx context.Context)

Start launches the finalize worker. It runs until Stop is called; ctx bounds the sink/offset calls made while finalizing.

func (*Coordinator) Stop added in v0.4.0 

func (c *Coordinator) Stop()

Stop closes the event stream and waits for in-flight finalization. Idempotent.

type FileBlobstore added in v0.5.0 

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

FileBlobstore is a Blobstore backed by a directory tree. Each key maps to a file under root. In the container model this points at a shared volume mounted into every job, so a savepoint written by one job is visible to another — the same namespace semantics S3 gives across hosts.

func NewFileBlobstore added in v0.5.0 

func NewFileBlobstore(dir string) *FileBlobstore

NewFileBlobstore roots a blobstore at dir (created on first write).

func (*FileBlobstore) Delete added in v0.5.0 

func (b *FileBlobstore) Delete(key string) error

func (*FileBlobstore) Exists added in v0.5.0 

func (b *FileBlobstore) Exists(key string) (bool, error)

func (*FileBlobstore) Get added in v0.5.0 

func (b *FileBlobstore) Get(key string) (io.ReadCloser, error)

func (*FileBlobstore) List added in v0.5.0 

func (b *FileBlobstore) List(prefix string) ([]string, error)

func (*FileBlobstore) Put added in v0.5.0 

func (b *FileBlobstore) Put(key string, r io.Reader) error

type FileStorage 

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

FileStorage implements Storage using the local filesystem. Each checkpoint is written as a JSON file with the checkpoint ID in the filename. Writes are atomic (write to temp file, then rename).

func NewFileStorage 

func NewFileStorage(dir string) *FileStorage

NewFileStorage creates a FileStorage that writes checkpoints to the given directory. The directory is created if it doesn't exist. On startup, call SweepOrphans to clean up state directories from failed checkpoints.

func (*FileStorage) DeleteStateDirs added in v0.4.0 

func (fs *FileStorage) DeleteStateDirs(id string)

DeleteStateDirs removes the state directories for a checkpoint ID. Used during retention/GC.

func (*FileStorage) Load 

func (fs *FileStorage) Load() (*CheckpointData, error)

Load reads the most recent checkpoint (any status). Returns nil with no error if no checkpoint exists.

func (*FileStorage) LoadLatestCompleted added in v0.4.0 

func (fs *FileStorage) LoadLatestCompleted() (*CheckpointData, error)

LoadLatestCompleted reads the newest checkpoint that reached completed status. Falls back to nil (no error) when none exists.

func (*FileStorage) LoadSpecific 

func (fs *FileStorage) LoadSpecific(id string) (*CheckpointData, error)

LoadSpecific reads a checkpoint with the given ID.

func (*FileStorage) Save 

func (fs *FileStorage) Save(data *CheckpointData) error

Save writes checkpoint data to a JSON file atomically and durably (fsync before rename). Also maintains two pointers: latest.json (any status) and latest-completed.json (completed only).

func (*FileStorage) StateDir added in v0.4.0 

func (fs *FileStorage) StateDir(id string) string

StateDir returns the checkpoint-specific state directory for the given checkpoint ID. All per-owner state directories live under this single parent.

func (*FileStorage) SweepOrphans added in v0.4.0 

func (fs *FileStorage) SweepOrphans() error

SweepOrphans deletes any <id>.state directories whose matching checkpoint JSON does not exist. Call once at startup before any pipeline runs.

func (*FileStorage) UpdateStatus added in v0.4.0 

func (fs *FileStorage) UpdateStatus(id string, status Status) error

UpdateStatus rewrites an existing checkpoint with a new status (prepared → completed promotion after a successful sink commit, or during recovery when the transaction marker proves the commit).

type HookAction added in v0.4.0 

type HookAction int

HookAction is what a test hook tells the coordinator to do after a step. 

const (
	// HookContinue proceeds normally.
	HookContinue HookAction = iota
	// HookFail injects a failure at this step: the coordinator aborts
	// the sink transaction and reports a fatal error.
	HookFail
	// HookHalt simulates a process crash: the coordinator stops doing
	// anything, silently, forever. The pipeline is left to be
	// cancelled by the test.
	HookHalt
)

type Status added in v0.4.0 

type Status string

Status tracks a checkpoint through the two-phase commit protocol. 

prepared:  offsets + state persisted, sink transaction flushed but
           NOT yet committed. The commit decision is logged.
completed: sink transaction committed; safe to restore from.

Checkpoints written without an explicit status (legacy, or the uncoordinated at-least-once path) are treated as completed. 

const (
	StatusPrepared  Status = "prepared"
	StatusCompleted Status = "completed"
)

type Step added in v0.4.0 

type Step string

Step identifies a point in the coordinated checkpoint protocol. The test hook fires AFTER each step succeeds, which is how the crash-window tests halt the coordinator at exact protocol positions. 

const (
	StepPersistPrepared  Step = "persist-prepared"  // checkpoint file written with status=prepared
	StepSinkCommitted    Step = "sink-committed"    // sink transaction committed (EndTxn)
	StepPersistCompleted Step = "persist-completed" // status promoted to completed
	StepOffsetsCommitted Step = "offsets-committed" // advisory broker offset commit
)

type Storage 

type Storage interface {
	// Save writes checkpoint data to persistent storage.
	// The implementation must be atomic — a partial write must not
	// corrupt a previous checkpoint — and durable (fsync) before
	// returning, because the coordinator treats a successful Save of a
	// prepared checkpoint as the logged commit decision.
	Save(data *CheckpointData) error

	// Load reads the most recent checkpoint regardless of status.
	// Returns nil with no error if no checkpoint exists.
	Load() (*CheckpointData, error)

	// LoadLatestCompleted reads the most recent checkpoint whose
	// status is completed. Returns nil with no error if none exists.
	LoadLatestCompleted() (*CheckpointData, error)

	// LoadSpecific reads a checkpoint with the given ID.
	LoadSpecific(id string) (*CheckpointData, error)

	// UpdateStatus rewrites the status of an existing checkpoint
	// (prepared → completed promotion).
	// StateDir returns the root directory for native state snapshots
	// (e.g. Pebble hard-links) associated with a checkpoint.
	StateDir(id string) string

	UpdateStatus(id string, status Status) error
}

Storage is the interface for persisting checkpoint data. Implementations can write to local disk, S3, etc.

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.