engine

package
v0.0.66 Latest Latest
Warning

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

 Go to latest Published: Jul 13, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	TaskSnapshotRestore          = wire.TaskSnapshotRestore
	TaskConfigPatch              = wire.TaskConfigPatch
	TaskConfigApply              = wire.TaskConfigApply
	TaskConfigValidate           = wire.TaskConfigValidate
	TaskConfigReload             = wire.TaskConfigReload
	TaskMarkReady                = wire.TaskMarkReady
	TaskRestartSeid              = wire.TaskRestartSeid
	TaskConfigureGenesis         = wire.TaskConfigureGenesis
	TaskConfigureStateSync       = wire.TaskConfigureStateSync
	TaskSnapshotUpload           = wire.TaskSnapshotUpload
	TaskSnapshotUploadOnce       = wire.TaskSnapshotUploadOnce
	TaskResultExport             = wire.TaskResultExport
	TaskAwaitCondition           = wire.TaskAwaitCondition
	TaskGenerateIdentity         = wire.TaskGenerateIdentity
	TaskGenerateGentx            = wire.TaskGenerateGentx
	TaskUploadGenesisArtifacts   = wire.TaskUploadGenesisArtifacts
	TaskAssembleAndUploadGenesis = wire.TaskAssembleAndUploadGenesis
	TaskSetGenesisPeers          = wire.TaskSetGenesisPeers
	TaskGovVote                  = wire.TaskGovVote
	TaskGovSoftwareUpgrade       = wire.TaskGovSoftwareUpgrade
	TaskGovParamChange           = wire.TaskGovParamChange
	TaskEvmLogicalDigest         = wire.TaskEvmLogicalDigest
	TaskMarkNotReady             = wire.TaskMarkNotReady
	TaskStopSeid                 = wire.TaskStopSeid
	TaskResetData                = wire.TaskResetData
)

Variables

View Source 
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

func TaskIDFromContext added in v0.0.50 

func TaskIDFromContext(ctx context.Context) string

TaskIDFromContext returns the engine-assigned task ID for the current handler, or "" when ctx is not engine-produced. Sign-tx handlers use this to derive the per-task memo tag.

func WithTaskID added in v0.0.50 

func WithTaskID(ctx context.Context, id string) context.Context

WithTaskID attaches a task ID to ctx for handler consumption. The engine calls this in newTaskContext; tests use it to bypass Submit.

Types

type Checkpointer added in v0.0.61 

type Checkpointer interface {
	SaveTxMarker(m *TxMarker) error
	// GetTxMarker returns (nil, nil) when no marker exists for taskID.
	GetTxMarker(taskID string) (*TxMarker, error)
}

Checkpointer persists a TxMarker durably before broadcast and retrieves it on re-execution, so a crash between broadcast and result-persist re-adopts the in-flight tx instead of signing a second one. Nil when no durable store is configured (handlers then broadcast without the guard, and log it).

type Engine 

type Engine struct {

	// Config is set once during single-threaded startup before Submit
	// is reachable; read-only thereafter. No synchronization.
	Config ExecutionConfig
	// 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. Callers MUST install handler dependencies on e.Config before RehydrateStaleTasks, else a rehydrated handler races with the write.

func (*Engine) GetResult 

func (e *Engine) GetResult(id string) *TaskResult

GetResult returns a task by ID, or nil if not found.

func (*Engine) Healthz 

func (e *Engine) Healthz() bool

Healthz returns true after the engine has been marked ready. Use as a readiness check.

func (*Engine) Livez added in v0.0.31 

func (e *Engine) Livez() error

Livez returns nil when the engine's backing store is responsive. Use as a liveness check — a non-nil error means the process is wedged (e.g., SQLite WAL corruption, PVC read-only).

func (*Engine) RecentResults 

func (e *Engine) RecentResults() []TaskResult

RecentResults returns the most recent task results across all states.

func (*Engine) RehydrateStaleTasks added in v0.0.50 

func (e *Engine) RehydrateStaleTasks()

RehydrateStaleTasks re-executes tasks left in "running" state by a previous process that exited before completing them. Run count is NOT incremented — rehydration is crash recovery of an incomplete run, not a new run. Must be called only after Config is installed.

func (*Engine) RemoveResult 

func (e *Engine) RemoveResult(id string) (bool, error)

RemoveResult removes a task by ID. It returns (found, err): found reports whether a row existed, and a non-nil err means the store delete failed and the caller should retry the DELETE.

If the task is currently running, its context is cancelled first so the goroutine stops rather than being orphaned by the row deletion. Cancelling and clearing the registry entry are separate steps: the task is always cancelled to stop the work, but its entry is only removed once store.Delete confirms the row is gone. If store.Delete fails the row stays 'running' and the failure is surfaced — the caller can retry the DELETE, and failing that a process restart's RehydrateStaleTasks resumes the row. Either path recovers it, so a failed delete never strands a task in 'running' with nothing able to act on it. An already-terminal task has no registered cancel func, so this is a plain delete.

func (*Engine) Status 

func (e *Engine) Status() StatusResponse

Status returns the engine's current state.

func (*Engine) Submit 

func (e *Engine) Submit(task Task) (string, error)

Submit starts a task in its own goroutine and returns its ID.

The engine follows a cloud-API model for task lifecycle:

  • If no task with this ID exists, create and execute it (run 1).
  • If the task is running or completed, return its ID (idempotent no-op).
  • If the task failed, re-execute it with an incremented run counter.

The caller submits a stable key and the engine owns the execution lifecycle.

type ExecutionConfig added in v0.0.50 

type ExecutionConfig struct {
	// Keyring is opened from SEI_KEYRING_BACKEND. Nil when unset;
	// sign-tx handlers report a clear error rather than panic.
	Keyring keyring.Keyring

	// RPC talks to the co-located seid CometBFT RPC. Sign-tx handlers
	// use it for the chain-confusion guard and inclusion polling.
	RPC *rpc.Client

	// Checkpointer persists a pre-broadcast TxMarker so a crashed sign-tx
	// task re-adopts its in-flight tx on re-run rather than re-signing.
	// Nil when no durable store is configured.
	Checkpointer Checkpointer
}

ExecutionConfig carries process-wide deps the engine exposes to handlers. Fields are nil when the corresponding subsystem is not configured.

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)

	// DeleteByType removes all results of the given task type and returns
	// how many rows were removed. Used by mark-not-ready to purge recorded
	// mark-ready results so a stranded running one cannot rehydrate and
	// release a node hold after a data wipe.
	DeleteByType(taskType string) (int, error)

	// LatestByType returns the most recently submitted result of the given
	// task type (any status), or (nil, nil) when none exists. Rehydration
	// uses it to detect a hold that supersedes a stranded mark-ready even
	// after the hold's own record has gone terminal (e.g. a failed purge
	// persisted it Failed), which a stale-only scan would miss.
	LatestByType(taskType string) (*TaskResult, error)

	// Ping verifies the store is responsive. Used by liveness checks.
	Ping() 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) Delete added in v0.0.25 

func (s *SQLiteStore) Delete(id string) (bool, error)

func (*SQLiteStore) DeleteByType added in v0.0.64 

func (s *SQLiteStore) DeleteByType(taskType string) (int, error)

func (*SQLiteStore) Get added in v0.0.25 

func (s *SQLiteStore) Get(id string) (*TaskResult, error)

func (*SQLiteStore) GetTxMarker added in v0.0.61 

func (s *SQLiteStore) GetTxMarker(taskID string) (*TxMarker, error)

func (*SQLiteStore) LatestByType added in v0.0.64 

func (s *SQLiteStore) LatestByType(taskType 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) Ping added in v0.0.31 

func (s *SQLiteStore) Ping() error

func (*SQLiteStore) Save added in v0.0.25 

func (s *SQLiteStore) Save(r *TaskResult) error

func (*SQLiteStore) SaveTxMarker added in v0.0.61 

func (s *SQLiteStore) SaveTxMarker(m *TxMarker) error

SaveTxMarker persists a pre-broadcast marker and fsyncs it (via checkpoint, since the store runs synchronous=NORMAL) before returning, so it survives a crash. Callers MUST let it return before broadcasting.

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 TaskError added in v0.0.31 

type TaskError struct {
	Task      string `json:"task"`
	Operation string `json:"operation"`
	Message   string `json:"message"`
	Hint      string `json:"hint,omitempty"`
	Retryable bool   `json:"retryable"`
	Cause     string `json:"cause,omitempty"`
}

TaskError is a structured error that includes operator-actionable context. Task handlers return this to provide rich error detail beyond a plain string.

func (*TaskError) Error added in v0.0.31 

func (e *TaskError) Error() string

type TaskHandler 

type TaskHandler func(ctx context.Context, params map[string]any) (json.RawMessage, error)

TaskHandler executes a specific task type. Handlers MUST be idempotent: the engine may re-execute a handler after a crash recovery. The returned json.RawMessage is the handler's optional structured result, persisted on TaskResult.Result and surfaced over GET /v0/tasks/{id}; handlers with no result return nil. The engine stamps the result on both the success and error paths (a handler returning an error may still carry a result, e.g. a tx hash for an inclusion-undetermined gov submit).

func TypedHandler added in v0.0.26 

func TypedHandler[T any](fn func(ctx context.Context, params T) error) TaskHandler

TypedHandler wraps a result-less typed handler 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. Handlers that produce a structured result use TypedHandlerWithResult instead.

func TypedHandlerWithResult added in v0.0.61 

func TypedHandlerWithResult[T, R any](fn func(ctx context.Context, params T) (R, error)) TaskHandler

TypedHandlerWithResult wraps a typed handler that returns a structured result into a TaskHandler. R is marshaled to json.RawMessage and returned alongside the error, so the engine persists it on both the success and error paths (an error return may still carry a meaningful R). A nil/zero R that marshals to "null" is treated as no result.

type TaskResult 

type TaskResult struct {
	ID          string          `json:"id"`
	Type        string          `json:"type"`
	Status      TaskStatus      `json:"status"`
	Run         int             `json:"run"`
	Params      map[string]any  `json:"params,omitempty"`
	Result      json.RawMessage `json:"result,omitempty"`
	Error       string          `json:"error,omitempty"`
	SubmittedAt time.Time       `json:"submittedAt"`
	CompletedAt *time.Time      `json:"completedAt,omitempty"`
}

TaskResult records a task and its outcome.

Result carries a handler's structured output (e.g. assemble-genesis emits {"genesisHash":"<bare-hex>"}). It is optional and additive: handlers that emit nothing leave it nil and it is omitted from the wire, so the currently-deployed controller is unaffected. This in-band channel — read by the controller over the trusted GET /v0/tasks/{id} path — is the authenticated alternative to publishing results through attacker-writable shared storage.

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 = wire.TaskType

TaskType identifies the kind of task to execute. TaskType aliases wire.TaskType so engine call sites and the dispatch-map key type are unchanged; the consts below re-export wire.Task* likewise.

type TxMarker added in v0.0.61 

type TxMarker struct {
	TaskID        string
	TxHash        string
	TxBytes       []byte
	AccountNumber uint64
	Sequence      uint64
	ChainID       string
}

TxMarker is the pre-broadcast idempotency record for a sign-tx task — engine-owned metadata read during crash recovery, distinct from a handler's TaskResult.Result. It carries the signed tx bytes so a re-run re-broadcasts the identical tx rather than re-signing (which risks a double submit).

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.