batch

package
v0.1.0-preview.4 Latest Latest
Warning

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

 Go to latest Published: Aug 11, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package batch provides restartable, explicitly persisted job and step execution for Spice applications.

Index

Examples

Constants

View Source 
const (

	// SQLBeginOutcomeStarted identifies a newly inserted instance.
	SQLBeginOutcomeStarted = "started"
	// SQLBeginOutcomeResumed identifies a new attempt over retained checkpoints.
	SQLBeginOutcomeResumed = "resumed"
	// SQLBeginOutcomeComplete identifies an already-complete instance.
	SQLBeginOutcomeComplete = "complete"
	// SQLBeginOutcomeRunning identifies an unexpired active attempt.
	SQLBeginOutcomeRunning = "running"
	// SQLBeginOutcomeChanged identifies incompatible ordered steps.
	SQLBeginOutcomeChanged = "changed"
	// SQLBeginOutcomeOverflow identifies an exhausted signed SQL attempt number.
	SQLBeginOutcomeOverflow = "overflow"
)

Variables

View Source 
var (
	// ErrPanicked identifies a contained batch step panic.
	ErrPanicked = errors.New("batch step panicked")
	// ErrAlreadyRunning identifies an active attempt for the same job instance.
	ErrAlreadyRunning = errors.New("batch instance is already running")
	// ErrStaleAttempt identifies a transition for an inactive or old attempt.
	ErrStaleAttempt = errors.New("batch attempt is stale")
	// ErrDefinitionChanged identifies a persisted instance whose ordered steps
	// differ from the current job definition.
	ErrDefinitionChanged = errors.New("batch definition changed")
	// ErrCapacity identifies an in-process store at its configured instance
	// limit.
	ErrCapacity = errors.New("batch store capacity reached")
)

Functions

This section is empty.

Types

type Attempt 

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

Attempt is immutable persisted restart metadata returned by a Store.

func NewAttempt 

func NewAttempt(spec AttemptSpec) (Attempt, error)

NewAttempt validates and freezes persisted restart metadata.

func (Attempt) Complete 

func (attempt Attempt) Complete() bool

Complete reports whether the store already completed this job instance.

func (Attempt) CompletedSteps 

func (attempt Attempt) CompletedSteps() []string

CompletedSteps returns a defensive copy of the durable completed prefix.

func (Attempt) Definition 

func (attempt Attempt) Definition() Definition

Definition returns the attempted job identity.

func (Attempt) Instance 

func (attempt Attempt) Instance() string

Instance returns the caller-owned idempotent job instance identity.

func (Attempt) Number 

func (attempt Attempt) Number() uint64

Number returns the one-based execution attempt.

type AttemptSpec 

type AttemptSpec struct {
	Definition     Definition
	Instance       string
	Number         uint64
	CompletedSteps []string
	Complete       bool
}

AttemptSpec is the inspectable input to NewAttempt.

type BeginRequest 

type BeginRequest struct {
	Definition Definition
	Instance   string
	Steps      []string
}

BeginRequest asks a Store to atomically begin or resume one job instance.

type ContextFactory 

type ContextFactory func() (context.Context, context.CancelFunc)

ContextFactory creates one fresh bounded context for a failure transition after a step context has failed or been canceled.

type Definition 

type Definition struct {
	ID     string
	Module string
}

Definition identifies one module-owned batch job.

type ExecutionSnapshot 

type ExecutionSnapshot struct {
	Definition      Definition
	Attempt         uint64
	CompletedSteps  []string
	Running         bool
	Complete        bool
	LastFailureStep string
	LastFailureKind FailureKind
}

ExecutionSnapshot is an immutable diagnostic view of one in-process execution. Instance identities are intentionally excluded.

type Failure 

type Failure struct {
	Attempt Attempt
	Step    string
	Kind    FailureKind
}

Failure releases one active attempt for a later restart. It intentionally omits the application error and instance payload from durable metadata.

type FailureKind 

type FailureKind string

FailureKind is bounded durable failure metadata. 

const (
	// FailureError identifies a returned step or persistence error.
	FailureError FailureKind = "error"
	// FailureCanceled identifies caller cancellation.
	FailureCanceled FailureKind = "canceled"
	// FailurePanic identifies a contained step panic.
	FailurePanic FailureKind = "panic"
)

type Job 

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

Job is an immutable ordered batch definition.

func NewJob 

func NewJob(definition Definition, specs []StepSpec) (*Job, error)

NewJob validates and freezes one ordered job.

func (*Job) Definition 

func (job *Job) Definition() Definition

Definition returns the job identity.

func (*Job) Steps 

func (job *Job) Steps() []Step

Steps returns a defensive copy of the ordered steps.

type MemoryStore 

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

MemoryStore is a concurrency-safe, capacity-bounded in-process Store.

State is not durable across process restarts. Use it for development, tests, and jobs whose restart state does not need to survive a process.

Example (Restart) 
package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/spice-framework/spice/batch"
)

func main() {
	store, err := batch.NewMemoryStore(100)
	if err != nil {
		panic(err)
	}
	loadAttempts := 0
	job, err := batch.NewJob(
		batch.Definition{
			ID:     "orders.import",
			Module: "example.com/shop/orders",
		},
		[]batch.StepSpec{
			{
				ID: "extract",
				Run: func(context.Context) error {
					fmt.Println("extract")
					return nil
				},
			},
			{
				ID: "load",
				Run: func(context.Context) error {
					loadAttempts++
					fmt.Println("load")
					if loadAttempts == 1 {
						return errors.New("database unavailable")
					}
					return nil
				},
			},
		},
	)
	if err != nil {
		panic(err)
	}
	failureContext := func() (context.Context, context.CancelFunc) {
		return context.WithTimeout(context.Background(), time.Second)
	}
	runner, err := batch.NewRunner(store, failureContext)
	if err != nil {
		panic(err)
	}

	first, firstErr := runner.Run(context.Background(), job, "2026-07-26")
	second, secondErr := runner.Run(context.Background(), job, "2026-07-26")
	third, thirdErr := runner.Run(context.Background(), job, "2026-07-26")
	fmt.Println(first.Attempt, first.StepsCompleted, firstErr != nil)
	fmt.Println(second.Attempt, second.StepsSkipped, secondErr)
	fmt.Println(third.Attempt, third.AlreadyComplete, thirdErr)

}

Output:
extract
load
load
1 1 true
2 1 <nil>
2 true <nil>

func NewMemoryStore 

func NewMemoryStore(capacity int) (*MemoryStore, error)

NewMemoryStore constructs an empty store with a fixed instance capacity.

func (*MemoryStore) Begin 

func (store *MemoryStore) Begin(
	ctx context.Context,
	request BeginRequest,
) (Attempt, error)

Begin atomically starts or resumes one execution attempt.

func (*MemoryStore) Checkpoint 

func (store *MemoryStore) Checkpoint(
	ctx context.Context,
	attempt Attempt,
	step string,
) error

Checkpoint atomically records the next ordered step.

func (*MemoryStore) Complete 

func (store *MemoryStore) Complete(
	ctx context.Context,
	attempt Attempt,
) error

Complete atomically marks a fully checkpointed attempt complete.

func (*MemoryStore) Delete 

func (store *MemoryStore) Delete(
	ctx context.Context,
	definition Definition,
	instance string,
) error

Delete removes inactive execution state and releases capacity.

func (*MemoryStore) Fail 

func (store *MemoryStore) Fail(ctx context.Context, failure Failure) error

Fail atomically releases an active attempt for a later restart.

func (*MemoryStore) Snapshot 

func (store *MemoryStore) Snapshot(
	ctx context.Context,
	definition Definition,
	instance string,
) (ExecutionSnapshot, bool, error)

Snapshot returns a defensive diagnostic view when an execution exists.

type Observation 

type Observation struct {
	Definition Definition
	Operation  Operation
	Step       string
	Attempt    uint64
	Duration   time.Duration
	Resumed    bool
	Completed  bool
	Err        error
	Panicked   bool
}

Observation contains bounded job metadata. Instance identities and application values are intentionally excluded.

type Observer 

type Observer func(context.Context, Observation)

Observer receives completed boundaries synchronously.

type Operation 

type Operation string

Operation identifies one observed batch boundary. 

const (
	// OperationStep identifies one executed step.
	OperationStep Operation = "step"
	// OperationJob identifies one completed or already-complete job attempt.
	OperationJob Operation = "job"
)

type PanicError 

type PanicError struct {
	Definition Definition
	Step       string
}

PanicError reports a contained step panic without exposing its recovered value.

func (*PanicError) Error 

func (err *PanicError) Error() string

Error describes the failed step.

func (*PanicError) Unwrap 

func (err *PanicError) Unwrap() error

Unwrap supports errors.Is(err, ErrPanicked).

type Result 

type Result struct {
	Attempt         uint64
	StepsSkipped    int
	StepsCompleted  int
	Resumed         bool
	AlreadyComplete bool
	Duration        time.Duration
}

Result summarizes one execution or restart.

type Runner 

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

Runner executes one immutable job at a time through an explicit Store.

func NewRunner 

func NewRunner(
	store Store,
	failureContext ContextFactory,
	observers ...Observer,
) (*Runner, error)

NewRunner constructs an instance-owned batch runner.

func (*Runner) Run 

func (runner *Runner) Run(
	ctx context.Context,
	job *Job,
	instance string,
) (Result, error)

Run atomically begins one instance, skips its durable completed prefix, executes remaining steps serially, and persists each successful checkpoint.

type SQLStatements 

type SQLStatements struct {
	Begin      string
	Checkpoint string
	Complete   string
	Fail       string
}

SQLStatements supplies dialect-owned atomic statements for the batch persistence protocol. Statement text is trusted startup configuration, never request input.

type SQLStore 

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

SQLStore implements Store through standard database/sql contracts.

func NewSQLStore 

func NewSQLStore(
	executor data.Executor,
	statements SQLStatements,
	options SQLStoreOptions,
) (*SQLStore, error)

NewSQLStore validates and freezes one driver-neutral SQL store. Construction performs no database operation.

func (*SQLStore) Begin 

func (store *SQLStore) Begin(
	ctx context.Context,
	request BeginRequest,
) (Attempt, error)

Begin atomically inserts, resumes, or observes one persisted instance.

The statement arguments are job ID, module, instance, canonical JSON step IDs, current UTC time, and lease expiry. It must return exactly one row with outcome, positive attempt number, and JSON completed step IDs.

func (*SQLStore) Checkpoint 

func (store *SQLStore) Checkpoint(
	ctx context.Context,
	attempt Attempt,
	step string,
) error

Checkpoint atomically records the next step and renews the attempt lease.

Statement arguments are job ID, module, instance, attempt number, step, current UTC time, and lease expiry.

func (*SQLStore) Complete 

func (store *SQLStore) Complete(
	ctx context.Context,
	attempt Attempt,
) error

Complete atomically completes the exact active attempt.

func (*SQLStore) Fail 

func (store *SQLStore) Fail(ctx context.Context, failure Failure) error

Fail atomically releases the exact active attempt for a later resume.

type SQLStoreOptions 

type SQLStoreOptions struct {
	// AttemptLease is the maximum time an attempt remains exclusively active
	// without a checkpoint. Steps may execute at least once when they outlive
	// this lease and another runner resumes the instance.
	AttemptLease time.Duration
	// Clock supplies current time. Nil selects time.Now.
	Clock func() time.Time
}

SQLStoreOptions controls durable attempt ownership.

type Step 

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

Step is one immutable job step.

func (Step) ID 

func (step Step) ID() string

ID returns the stable step identity.

type StepSpec 

type StepSpec struct {
	ID  string
	Run func(context.Context) error
}

StepSpec is the inspectable input to NewJob.

type Store 

type Store interface {
	Begin(context.Context, BeginRequest) (Attempt, error)
	Checkpoint(context.Context, Attempt, string) error
	Complete(context.Context, Attempt) error
	Fail(context.Context, Failure) error
}

Store owns atomic attempt and checkpoint transitions. An implementation defines whether that state survives process restarts.

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.