Documentation
¶
Index ¶
- Variables
- type Engine
- type ResultStore
- type SQLiteStore
- func (s *SQLiteStore) Close() error
- func (s *SQLiteStore) Delete(id string) (bool, error)
- func (s *SQLiteStore) Get(id string) (*TaskResult, error)
- func (s *SQLiteStore) List(limit int) ([]TaskResult, error)
- func (s *SQLiteStore) ListStaleTasks() ([]TaskResult, error)
- func (s *SQLiteStore) Save(r *TaskResult) error
- type StatusResponse
- type Task
- type TaskHandler
- type TaskResult
- type TaskStatus
- type TaskType
Constants ¶
This section is empty.
Variables ¶
var ErrInvalidTaskID = fmt.Errorf("task ID must be a valid UUID")
ErrInvalidTaskID is returned when a caller-provided task ID is not a valid UUID.
Functions ¶
This section is empty.
Types ¶
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine is the task executor. Every submitted task runs in its own goroutine. The store is the single source of truth for all task state. The engine context propagates to all handlers — on SIGTERM the context is cancelled and handlers observe ctx.Done() to stop gracefully.
func NewEngine ¶
func NewEngine(ctx context.Context, handlers map[TaskType]TaskHandler, store ResultStore) *Engine
NewEngine creates a new Engine. The engine runs until ctx is cancelled. On startup, any tasks left in "running" state from a previous process are re-executed.
func (*Engine) GetResult ¶
func (e *Engine) GetResult(id string) *TaskResult
GetResult returns a task by ID, or nil if not found.
func (*Engine) RecentResults ¶
func (e *Engine) RecentResults() []TaskResult
RecentResults returns the most recent task results across all states.
func (*Engine) RemoveResult ¶
RemoveResult removes a task by ID. Returns true if found.
func (*Engine) Status ¶
func (e *Engine) Status() StatusResponse
Status returns the engine's current state.
func (*Engine) Submit ¶
Submit starts a task in its own goroutine and returns its ID. When task.ID is set, it becomes the canonical identifier (enabling deterministic IDs from the controller). When empty, a random UUID is generated. If a task with the same ID already exists, the existing ID is returned without re-submitting.
type ResultStore ¶ added in v0.0.25
type ResultStore interface {
// Save persists a TaskResult. If a result with the same ID already
// exists, it is overwritten (upsert).
Save(r *TaskResult) error
// Get returns a result by ID, or (nil, nil) when not found.
Get(id string) (*TaskResult, error)
// List returns the most recent results, newest first, up to limit.
List(limit int) ([]TaskResult, error)
// ListStaleTasks returns tasks left in "running" state from a
// previous process that exited without completing them.
ListStaleTasks() ([]TaskResult, error)
// Delete removes a result by ID. Returns true if it existed.
Delete(id string) (bool, error)
// Close releases underlying resources.
Close() error
}
ResultStore persists task results across all lifecycle states. Implementations must be safe for concurrent use.
type SQLiteStore ¶ added in v0.0.25
type SQLiteStore struct {
// contains filtered or unexported fields
}
SQLiteStore persists task results in a SQLite database.
func NewMemoryStore ¶ added in v0.0.25
func NewMemoryStore() (*SQLiteStore, error)
NewMemoryStore returns a SQLiteStore backed by an in-memory SQLite database (the ":memory:" DSN). The database exists only for the lifetime of the returned store — nothing is written to disk. Useful for tests and non-sidecar CLI commands.
func NewSQLiteStore ¶ added in v0.0.25
func NewSQLiteStore(dbPath string) (*SQLiteStore, error)
NewSQLiteStore opens (or creates) a SQLite database at dbPath and runs any pending schema migrations. The file is opened in WAL mode with pragmas tuned for a single-writer sidecar workload.
The database file must reside on a local or block-device-backed filesystem (e.g. EBS, GCE PD, local SSD). WAL mode is unsafe on NFS-backed volumes (EFS, Azure Files, CephFS over NFS) because they do not support the POSIX byte-range locks that SQLite requires for the shared-memory (-shm) file.
func (*SQLiteStore) Close ¶ added in v0.0.25
func (s *SQLiteStore) Close() error
func (*SQLiteStore) Get ¶ added in v0.0.25
func (s *SQLiteStore) Get(id string) (*TaskResult, error)
func (*SQLiteStore) List ¶ added in v0.0.25
func (s *SQLiteStore) List(limit int) ([]TaskResult, error)
func (*SQLiteStore) ListStaleTasks ¶ added in v0.0.25
func (s *SQLiteStore) ListStaleTasks() ([]TaskResult, error)
func (*SQLiteStore) Save ¶ added in v0.0.25
func (s *SQLiteStore) Save(r *TaskResult) error
type StatusResponse ¶
type StatusResponse struct {
Status string `json:"status"`
}
StatusResponse is the shape returned by the status endpoint.
type Task ¶
type Task struct {
ID string `json:"id,omitempty"`
Type TaskType `json:"type"`
Params map[string]any `json:"params,omitempty"`
}
Task is a unit of work submitted by the controller. When ID is set, the engine uses it as the canonical task identifier (enabling deterministic IDs from the controller). When empty, the engine generates a random UUID.
type TaskHandler ¶
TaskHandler executes a specific task type. Handlers MUST be idempotent: the engine may re-execute a handler after a crash recovery.
func TypedHandler ¶ added in v0.0.26
func TypedHandler[T any](fn func(ctx context.Context, params T) error) TaskHandler
TypedHandler wraps a function that accepts typed params into a TaskHandler. The map[string]any params are marshaled to JSON and unmarshaled into the typed struct T, giving handlers compile-time type safety without changing the engine's dispatch mechanism.
type TaskResult ¶
type TaskResult struct {
ID string `json:"id"`
Type string `json:"type"`
Status TaskStatus `json:"status"`
Params map[string]any `json:"params,omitempty"`
Error string `json:"error,omitempty"`
SubmittedAt time.Time `json:"submittedAt"`
CompletedAt *time.Time `json:"completedAt,omitempty"`
}
TaskResult records a task and its outcome.
type TaskStatus ¶ added in v0.0.15
type TaskStatus string
TaskStatus represents the lifecycle state of a task.
const ( TaskStatusRunning TaskStatus = "running" TaskStatusCompleted TaskStatus = "completed" TaskStatusFailed TaskStatus = "failed" )
type TaskType ¶
type TaskType string
TaskType identifies the kind of task to execute.
const ( TaskSnapshotRestore TaskType = "snapshot-restore" TaskDiscoverPeers TaskType = "discover-peers" TaskConfigPatch TaskType = "config-patch" TaskConfigApply TaskType = "config-apply" TaskConfigValidate TaskType = "config-validate" TaskConfigReload TaskType = "config-reload" TaskMarkReady TaskType = "mark-ready" TaskConfigureGenesis TaskType = "configure-genesis" TaskConfigureStateSync TaskType = "configure-state-sync" TaskSnapshotUpload TaskType = "snapshot-upload" TaskResultExport TaskType = "result-export" TaskAwaitCondition TaskType = "await-condition" TaskGenerateIdentity TaskType = "generate-identity" TaskGenerateGentx TaskType = "generate-gentx" TaskUploadGenesisArtifacts TaskType = "upload-genesis-artifacts" TaskAssembleAndUploadGenesis TaskType = "assemble-and-upload-genesis" TaskSetGenesisPeers TaskType = "set-genesis-peers" TaskAssembleGenesisFork TaskType = "assemble-genesis-fork" TaskExportState TaskType = "export-state" )
Source Files