Help us shape the future of Go! Take the 2026 Go Developer Survey

worker

package
v0.1.1 Latest Latest
Warning

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

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

Documentation

Overview

Package task owns the workflow-neutral task contract and execution engine.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrEffectAlreadyAdmitted = repository.ErrTaskEffectAlreadyAdmitted
	ErrUnknownType           = errors.New("unknown task type")
	ErrInputConflict         = errors.New("task input conflict")
	ErrRetryUnsupported      = errors.New("task retry is not supported")
	ErrInvalidResult         = errors.New("invalid task result")
	ErrRegistryFrozen        = errors.New("task registry is frozen")
	ErrEffectForbidden       = errors.New("external effects are forbidden during recovery")
	ErrCodecPanic            = errors.New("task input codec panicked")
	ErrInvalidCanonical      = errors.New("task input codec returned invalid canonical JSON")
)
View Source 
var CycleCodec = StrictJSONCodec(func(input *CycleInput) error {
	if input.Cycle < 1 {
		return errors.New("cycle must be positive")
	}
	return nil
})

CycleCodec validates scheduled system task input.

View Source 
var ErrInvalidPolicy = errors.New("invalid task execution policy")

Functions

func DecodeCheckpoint 

func DecodeCheckpoint[T any](execution Execution) (T, bool, error)

func DecodeInput 

func DecodeInput[T any](execution Execution) (T, error)

func RecoverableEngineFailure 

func RecoverableEngineFailure(reason string) bool

RecoverableEngineFailure reports whether a failure the Engine recorded itself may succeed on a later retry, for example once a build that can run the task is deployed. Manual Retry runs in recover mode, which inspects the recorded evidence before any new external effect, so retrying is as safe as recovering after a crash at the same point. An input whose hash no longer matches is corrupt and never recovers.

Types

type BackoffPolicy 

type BackoffPolicy struct {
	InitialDelay time.Duration `json:"initial_delay"`
	Multiplier   float64       `json:"multiplier"`
	MaximumDelay time.Duration `json:"maximum_delay"`
	Jitter       float64       `json:"jitter"`
}

func DefaultBackoffPolicy 

func DefaultBackoffPolicy() BackoffPolicy

type Codec 

type Codec interface {
	Canonicalize(json.RawMessage) (json.RawMessage, error)
}

Codec validates input and returns its canonical JSON representation.

func StrictJSONCodec 

func StrictJSONCodec[T any](validate func(*T) error) Codec

StrictJSONCodec rejects unknown fields and trailing values before applying validate. Re-marshalling produces stable object-key ordering.

type CodecFunc 

type CodecFunc func(json.RawMessage) (json.RawMessage, error)

CodecFunc adapts a validation function to Codec.

func (CodecFunc) Canonicalize 

func (f CodecFunc) Canonicalize(input json.RawMessage) (json.RawMessage, error)

type CycleInput 

type CycleInput struct {
	Cycle int64 `json:"cycle"`
}

CycleInput is the input of a scheduled system task; Cycle numbers its round.

type Definition 

type Definition struct {
	Type model.TaskType
	// MaxConcurrency limits simultaneous executions of this type; zero uses only the global limit.
	MaxConcurrency int
	WorkStart      WorkStartPolicy
	InputVersion   int
	Codec          Codec
	Policy         ExecutionPolicy
	AllowRetry     bool
	// CanManualRetry optionally narrows AllowRetry for one failed task based on
	// its durable failure evidence. Nil preserves the type-wide policy.
	CanManualRetry func(*model.Task) bool
	// OnEngineFailure settles domain state when the engine fails a claim before
	// the handler can return its own result.
	OnEngineFailure func(*model.Task, string) Settlement
	// Identity derives the task's identity from its canonical input, so a
	// caller can never enqueue a round under another row's identity.
	Identity     IdentityFunc
	InspectRetry func(context.Context, *repository.Repositories, *model.Task) error
	PrepareRetry func(context.Context, *repository.Repositories, *model.Task) (RetryPreparation, error)
	// BindSuccessor moves domain ownership from a superseded round to its
	// successor in the same transaction, unless PrepareRetry overrides it.
	BindSuccessor func(context.Context, *repository.Repositories, *model.Task, *model.Task) error
	// RetryCheckpoint derives the successor evidence of a coordinator's budgeted
	// retry inside its transaction, so it must not call the network or wait on a
	// process lock. Nil inherits the failed round's checkpoint.
	RetryCheckpoint     func(context.Context, *repository.Repositories, *model.Task) (json.RawMessage, error)
	NextCycleCheckpoint func(context.Context, *repository.Repositories, *model.Task) (json.RawMessage, error)
}

Definition is the runtime contract for a task type.

type Engine 

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

Engine is the only task claimant and lease owner.

func NewEngine 

func NewEngine(config EngineConfig, repos *repository.Repositories, registry *Registry, logger *slog.Logger) (*Engine, error)

func (*Engine) Healthy 

func (e *Engine) Healthy() bool

func (*Engine) Name 

func (e *Engine) Name() string

func (*Engine) Run 

func (e *Engine) Run(ctx context.Context) error

type EngineConfig 

type EngineConfig struct {
	Concurrency   int
	PollInterval  time.Duration
	LeaseDuration time.Duration
	OnTaskSettled func(*model.Task, repository.TaskTransition)
	// SettlementRetryWait overrides waiting after the claim's lease is renewed.
	// A nil callback uses lease-preserving waits with the engine's retry schedule.
	SettlementRetryWait func(context.Context, int64, int64, time.Duration) error
}

type EnqueueRequest 

type EnqueueRequest struct {
	Type        model.TaskType
	Input       any
	AvailableAt time.Time
}

type Execution 

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

Execution is an immutable view of one fenced claim.

func (Execution) Checkpoint 

func (e Execution) Checkpoint() json.RawMessage

func (Execution) ClaimGeneration 

func (e Execution) ClaimGeneration() int64

func (Execution) DependencyRetryAt 

func (e Execution) DependencyRetryAt(retries int) (time.Time, bool)

func (Execution) ID 

func (e Execution) ID() int64

func (Execution) Input 

func (e Execution) Input() json.RawMessage

func (Execution) InputVersion 

func (e Execution) InputVersion() int

func (Execution) LastError 

func (e Execution) LastError() string

func (Execution) Mode 

func (e Execution) Mode() model.TaskResumeMode

func (Execution) ObserveOperation 

func (e Execution) ObserveOperation(ctx context.Context, key string) (time.Time, error)

func (Execution) Policy 

func (e Execution) Policy() ExecutionPolicy

func (Execution) RecordEffectEvidence added in v0.1.1 

func (e Execution) RecordEffectEvidence(ctx context.Context, value any, settlement Settlement) error

RecordEffectEvidence persists evidence an external effect already returned. It survives caller cancellation so shutdown cannot drop an obtained credential; the claim fence is still checked in the same transaction.

func (Execution) ResendAt 

func (e Execution) ResendAt(sentAt time.Time, sends int) (time.Time, error)

func (Execution) ResolveOperation 

func (e Execution) ResolveOperation(ctx context.Context, key string, checkpoint any, settlement Settlement) error

func (Execution) RetryCount 

func (e Execution) RetryCount() int

func (Execution) Type 

func (e Execution) Type() model.TaskType

func (Execution) WithCheckpointedEffect 

func (e Execution) WithCheckpointedEffect(
	ctx context.Context,
	operationKey string,
	checkpoint any,
	settlement Settlement,
	effect func(context.Context) error,
) (attempted bool, err error)

WithCheckpointedEffect admits an external effect, persists its recovery evidence, and only then invokes effect. A false attempted result guarantees that effect was not called; recovery after a process crash must still rely on the durable checkpoint rather than this return value.

func (Execution) WriteCheckpoint 

func (e Execution) WriteCheckpoint(ctx context.Context, value any) error

func (Execution) WriteCheckpointWith 

func (e Execution) WriteCheckpointWith(ctx context.Context, value any, settlement Settlement) error

WriteCheckpointWith atomically persists recovery identity and monotonic domain evidence before an external effect can be attempted.

type ExecutionPolicy 

type ExecutionPolicy struct {
	MaxAttempts         int            `json:"max_attempts"`
	DependencyRetries   int            `json:"dependency_retries,omitempty"`
	Backoff             BackoffPolicy  `json:"backoff"`
	ResendBackoff       *BackoffPolicy `json:"resend_backoff,omitempty"`
	InvocationTimeout   time.Duration  `json:"invocation_timeout"`
	ObservationWindow   time.Duration  `json:"observation_window"`
	ObservationInterval time.Duration  `json:"observation_interval,omitempty"`
	// WaitInterval controls ordinary waits; zero uses the engine poll interval.
	WaitInterval time.Duration `json:"wait_interval,omitempty"`
}

type FuncHandler 

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

FuncHandler adapts functions to one registered task type.

func NewFuncHandler 

func NewFuncHandler(definition Definition, execute, recover func(context.Context, Execution) Result) *FuncHandler

func (*FuncHandler) Definition 

func (h *FuncHandler) Definition() Definition

func (*FuncHandler) Execute 

func (h *FuncHandler) Execute(ctx context.Context, execution Execution) Result

func (*FuncHandler) Recover 

func (h *FuncHandler) Recover(ctx context.Context, execution Execution) Result

type Handler 

type Handler interface {
	Definition() Definition
	Execute(context.Context, Execution) Result
	Recover(context.Context, Execution) Result
}

Handler owns exactly one task type.

type Identity 

type Identity struct {
	SubjectType string
	SubjectKey  string
	Generation  *int64
}

Identity names the domain row a task works on and the round of work on it. Enqueue returns the current round of an identity instead of creating another; a nil Generation opts the task out of deduplication.

type IdentityFunc 

type IdentityFunc func(canonical json.RawMessage) (Identity, error)

IdentityFunc derives a task's identity from its canonical input.

func CycleIdentity 

func CycleIdentity(subjectKey string) IdentityFunc

CycleIdentity identifies the cycles of one scheduled system task.

func IdentityFromInput 

func IdentityFromInput[T any](subjectType string, identify func(T) (rowID, generation int64)) IdentityFunc

IdentityFromInput decodes the canonical input as T; identify returns the subject row ID, keyed by its decimal form, and the generation of work on it.

type Manager 

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

Manager exposes the single task engine through the application's worker lifecycle and health contracts.

func NewManager 

func NewManager(engine *Engine, logger *slog.Logger) *Manager

func (*Manager) Start 

func (m *Manager) Start(ctx context.Context)

Start runs the universal task engine until ctx is cancelled.

func (*Manager) WorkerHealth 

func (m *Manager) WorkerHealth() map[string]bool

type Message 

type Message interface {
	MessageType() string
}

type MessageHandler 

type MessageHandler func(context.Context, *repository.Repositories, Message) error

type Messenger 

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

Messenger dispatches declared messages synchronously in the source transaction.

func (*Messenger) Handover 

func (m *Messenger) Handover(ctx context.Context, tx *repository.Repositories, message Message) error

func (*Messenger) Notify 

func (m *Messenger) Notify(ctx context.Context, tx *repository.Repositories, message Message) error

type Registry 

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

Registry is immutable after Engine starts.

func NewRegistry 

func NewRegistry() *Registry

func (*Registry) Definition 

func (r *Registry) Definition(taskType model.TaskType) (Definition, bool)

func (*Registry) Handler 

func (r *Registry) Handler(taskType model.TaskType) (Handler, bool)

func (*Registry) Messenger 

func (r *Registry) Messenger(producer string, handovers, notices []string) (*Messenger, error)

func (*Registry) OwnHandover 

func (r *Registry) OwnHandover(id, kind string, handler MessageHandler) error

func (*Registry) Register 

func (r *Registry) Register(handler Handler) error

func (*Registry) RegisterSchedule 

func (r *Registry) RegisterSchedule(def ScheduleDefinition) error

func (*Registry) Subscribe 

func (r *Registry) Subscribe(id, kind string, handler MessageHandler) error

func (*Registry) Types 

func (r *Registry) Types() []model.TaskType

type Result 

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

Result is a closed set constructed through the helpers below.

func Cancel 

func Cancel(message string, settlement Settlement) Result

func Complete 

func Complete(message string, settlement Settlement) Result

func CompleteCycle 

func CompleteCycle(message string, settlement Settlement) Result

func Fail 

func Fail(err error, failureReason string, settlement Settlement) Result

func Retry 

func Retry(err error, failureReason string, settlement Settlement) Result

func RetryInMode 

func RetryInMode(err error, reason string, mode model.TaskResumeMode, settlement Settlement) Result

func Wait 

func Wait(mode model.TaskResumeMode, reason, message string, settlement Settlement) Result

func WaitForObservation 

func WaitForObservation(mode model.TaskResumeMode, reason, message string, settlement Settlement) Result

func WaitUntil 

func WaitUntil(mode model.TaskResumeMode, until time.Time, reason, message string, settlement Settlement) Result

func (Result) WithContentionSettlement 

func (r Result) WithContentionSettlement(settlement Settlement) Result

WithContentionSettlement retains only recovery evidence when database contention defers the business result. It must not complete or fail the operation or authorize another external effect.

func (Result) WithDiagnostic 

func (r Result) WithDiagnostic(err error) Result

WithDiagnostic preserves an observation failure without consuming a retry. Only Wait results may carry a diagnostic.

func (Result) WithObservationWindow 

func (r Result) WithObservationWindow(stageKey string, timeoutErr error, timeoutRetryMode model.TaskResumeMode) Result

WithObservationWindow lets the engine retry an observation stage when its policy window expires. The same stage retains its start across claims.

func (Result) WithResumeMode 

func (r Result) WithResumeMode(mode model.TaskResumeMode) Result

func (Result) WithRetrySettlements 

func (r Result) WithRetrySettlements(onRetry RetrySettlement, onExhausted Settlement) Result

func (Result) WithWorkStartedAt 

func (r Result) WithWorkStartedAt(startedAt time.Time) Result

WithWorkStartedAt carries observed operation timing into fenced settlement. Zero means no operation was observed; existing first-start evidence is kept.

type ResultKind 

type ResultKind uint8

type RetryPreparation 

type RetryPreparation struct {
	Request    EnqueueRequest
	Checkpoint json.RawMessage
	ResumeMode model.TaskResumeMode
	Validate   func(context.Context, *repository.Repositories, *model.Task) error
	Bind       func(context.Context, *repository.Repositories, *model.Task, *model.Task) error
	Release    func()
}

type RetrySettlement 

type RetrySettlement func(context.Context, *repository.Repositories, time.Time) error

RetrySettlement receives the next opportunity's earliest execution time.

type ScheduleDefinition 

type ScheduleDefinition struct {
	Key      string
	Type     model.TaskType
	Interval time.Duration
}

type Scheduler 

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

Scheduler restricts a task owner to its declared enqueue and wake types.

func (*Scheduler) EnqueueInTransaction 

func (s *Scheduler) EnqueueInTransaction(ctx context.Context, tx *repository.Repositories, request EnqueueRequest) (*model.Task, bool, error)

func (*Scheduler) EnqueueRecoveryInTransaction 

func (s *Scheduler) EnqueueRecoveryInTransaction(ctx context.Context, tx *repository.Repositories, sourceID int64, request EnqueueRequest) (*model.Task, error)

EnqueueRecoveryInTransaction connects replanned work to its stopped source. The domain caller validates and binds the business generation in this transaction.

func (*Scheduler) RetryInTransaction 

func (s *Scheduler) RetryInTransaction(ctx context.Context, tx *repository.Repositories, sourceID int64, availableAt time.Time) (*model.Task, error)

RetryInTransaction is reserved for a coordinator's budgeted retry settlement. It performs no network preparation; the type's RetryCheckpoint decides the successor evidence.

func (*Scheduler) WakeInTransaction 

func (s *Scheduler) WakeInTransaction(ctx context.Context, tx *repository.Repositories, ids []int64, filter WakePendingFilter) (int, error)

type Service 

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

func NewService 

func NewService(registry *Registry, repos *repository.Repositories) (*Service, error)

func (*Service) Acknowledge 

func (s *Service) Acknowledge(ctx context.Context, id int64) error

func (*Service) AcknowledgeMatching 

func (s *Service) AcknowledgeMatching(ctx context.Context, filter repository.TaskAcknowledgeFilter) (int, error)

AcknowledgeMatching acknowledges a backlog of failures in one step and reports how many it acknowledged.

func (*Service) Acknowledgeable 

func (s *Service) Acknowledgeable(task *model.Task) bool

func (*Service) CountAcknowledgeable 

func (s *Service) CountAcknowledgeable(ctx context.Context, filter repository.TaskAcknowledgeFilter) (int, error)

CountAcknowledgeable reports how many failures AcknowledgeMatching would acknowledge under the same filter.

func (*Service) Enqueue 

func (s *Service) Enqueue(ctx context.Context, request EnqueueRequest) (*model.Task, bool, error)

func (*Service) EnqueueInTransaction 

func (s *Service) EnqueueInTransaction(
	ctx context.Context,
	txRepos *repository.Repositories,
	request EnqueueRequest,
) (*model.Task, bool, error)

EnqueueInTransaction uses a transaction-scoped repository set supplied by the caller. It exists for domain mutations that must bind the task in the same transaction; callers still cannot bypass registry validation.

func (*Service) EnqueueOrReplaceTerminalInTransaction 

func (s *Service) EnqueueOrReplaceTerminalInTransaction(
	ctx context.Context,
	txRepos *repository.Repositories,
	request EnqueueRequest,
) (*model.Task, bool, error)

EnqueueOrReplaceTerminalInTransaction is the cached-content upload-plan entrypoint. A new live reference creates a successor to terminal work.

func (*Service) EnqueueTx 

func (s *Service) EnqueueTx(
	ctx context.Context,
	request EnqueueRequest,
	bind func(context.Context, *repository.Repositories, *model.Task, bool) error,
) (*model.Task, bool, error)

EnqueueTx creates a task and binds its domain owner in one transaction. bind must perform database work only and tolerate a transaction retry.

func (*Service) Get 

func (s *Service) Get(ctx context.Context, id int64) (*model.Task, error)

func (*Service) InspectRetry 

func (s *Service) InspectRetry(ctx context.Context, task *model.Task) error

InspectRetry preserves domain rejection evidence for callers presenting an unavailable action. Retry rechecks eligibility inside its transaction before creating a successor.

func (*Service) InspectRetryWithRepositories 

func (s *Service) InspectRetryWithRepositories(ctx context.Context, repos *repository.Repositories, task *model.Task) error

func (*Service) Policy 

func (s *Service) Policy(taskType model.TaskType) (ExecutionPolicy, bool)

func (*Service) Retry 

func (s *Service) Retry(ctx context.Context, id int64) (*model.Task, error)

func (*Service) Retryable 

func (s *Service) Retryable(task *model.Task) bool

func (*Service) RetryableContext 

func (s *Service) RetryableContext(ctx context.Context, task *model.Task) (bool, error)

func (*Service) Scheduler 

func (s *Service) Scheduler(owner string, types ...model.TaskType) (*Scheduler, error)

func (*Service) WakeInTransaction 

func (s *Service) WakeInTransaction(ctx context.Context, txRepos *repository.Repositories, ids []int64) (int, error)

type Settlement 

type Settlement func(context.Context, *repository.Repositories) error

Settlement performs database-only domain changes in the task transition transaction. It must be safe to invoke again after a rollback.

type WakePendingFilter 

type WakePendingFilter struct {
	Types           []model.TaskType
	SkipWaitReasons []string
}

type WorkStartPolicy 

type WorkStartPolicy uint8

WorkStartPolicy identifies when a task begins its own work. 

const (
	// WorkStartOnHandler records work start before invoking the handler.
	WorkStartOnHandler WorkStartPolicy = iota + 1
	// WorkStartOnEffect leaves timing to WithCheckpointedEffect or validated
	// handler evidence saved with checkpoint or Result.WithWorkStartedAt settlement.
	WorkStartOnEffect
)

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.