sqlite

package
v0.0.1-rc.1 Latest Latest
Warning

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

 Go to latest Published: Jun 25, 2026 License: AGPL-3.0 Imports: 12 Imported by: 0

Documentation

Overview

Package sqlite provides a SQLite-backed implementation of store.Store.

Usage 

opts := sqlite.DefaultOptions()
s, err := sqlite.Open(ctx, "/path/to/sqi.db", opts)
if err != nil {
    log.Fatal(err)
}
defer s.Close()

The returned *Store satisfies store.Store in full. It can be passed wherever a store.Store is expected, or used directly for the [Ping] method that is not part of the public interface (used by health checks).

Concurrency

The underlying sql.DB is opened with sql.DB.SetMaxOpenConns(1) which serializes all reads and writes through a single SQLite connection. This is the correct setting for SQLite in WAL mode when used from a single process: it prevents write-write conflicts and avoids the overhead of connection negotiation while still allowing the WAL reader snapshot semantics to function correctly within a single connection.

Migrations

When Options.AutoMigrate is true (the default), Open calls goose.Up against the embedded migration FS before returning. Set AutoMigrate to false in HA deployments where migrations are applied explicitly with "sqi-server migrate up" before rolling out new server instances.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Options 

type Options struct {
	// AutoMigrate controls whether Open runs goose.Up to apply any pending
	// migrations before returning. Default: true.
	//
	// Set to false in HA deployments where migrations are applied explicitly
	// via "sqi-server migrate up" before starting new server instances.
	AutoMigrate bool
}

Options configures Open.

func DefaultOptions 

func DefaultOptions() Options

DefaultOptions returns an Options with recommended defaults.

type Store 

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

Store is a SQLite-backed implementation of store.Store. Create one with Open; close it with [Close] when done.

func Open 

func Open(ctx context.Context, path string, opts Options) (*Store, error)

Open opens (or creates) the SQLite database at path, applies connection pragmas, optionally runs pending migrations, prepares all statements, and returns a ready-to-use *Store.

Open returns an error and releases all resources if any step fails.

func (*Store) ActiveClaimCount 

func (s *Store) ActiveClaimCount(ctx context.Context, poolID string) (int, error)

ActiveClaimCount implements store.UsageClaimStore.

func (*Store) AppendAuditEntry 

func (s *Store) AppendAuditEntry(ctx context.Context, entry store.AuditEntry) error

AppendAuditEntry implements store.AuditStore.

func (*Store) AssignTask 

func (s *Store) AssignTask(ctx context.Context, id, workerID string, assignedAt time.Time) error

AssignTask implements store.TaskStore.

func (*Store) Backup 

func (s *Store) Backup(ctx context.Context, destPath string) error

Backup creates a consistent, online copy of the database at destPath using SQLite's VACUUM INTO statement. The source database remains fully available for reads and writes during the operation.

VACUUM INTO is the SQL-level equivalent of the SQLite C online backup API (sqlite3_backup_*): it produces a clean, fully checkpointed snapshot of the live database in a single atomic operation. The modernc.org/sqlite driver does not expose the C backup API directly, so VACUUM INTO is the idiomatic replacement.

destPath must not already exist; SQLite will return an error if it does.

func (*Store) CancelJobAttempts 

func (s *Store) CancelJobAttempts(ctx context.Context, jobID string, endedAt time.Time) (int, error)

CancelJobAttempts implements store.TaskAttemptStore.

func (*Store) CancelJobStatus 

func (s *Store) CancelJobStatus(ctx context.Context, id string) error

CancelJobStatus implements store.JobStore.

It issues a conditional UPDATE that only fires when the job is not already in a terminal state. When zero rows are affected the job is either missing or already terminal; a follow-up GetJob disambiguates:

func (*Store) CancelJobTasks 

func (s *Store) CancelJobTasks(ctx context.Context, jobID string, now time.Time) ([]store.Task, error)

CancelJobTasks implements store.TaskStore.

The SELECT and UPDATE execute inside a single SQLite transaction so no concurrent scheduler tick can assign a task between observation and cancellation. The rows cursor is closed inside a helper closure before the UPDATE runs, which avoids any potential cursor/write contention on the single-connection pool.

func (*Store) Checkpoint 

func (s *Store) Checkpoint(ctx context.Context) (walPages, checkpointed int, err error)

Checkpoint runs a WAL checkpoint in TRUNCATE mode, which copies all committed WAL frames back into the main database file and then truncates the WAL to zero bytes. It returns the number of WAL log pages and the number successfully checkpointed.

TRUNCATE mode is appropriate for the periodic background case: because the store uses a single connection (SetMaxOpenConns(1)), there are no concurrent readers that could block the checkpoint, so all frames will always be transferred in one call.

func (*Store) Close 

func (s *Store) Close() error

Close closes all prepared statements and the underlying database connection. It is safe to call Close more than once.

func (*Store) CommittedCores 

func (s *Store) CommittedCores(ctx context.Context, workerID string, fullMachineCost int) (int, error)

CommittedCores implements store.TaskStore.

func (*Store) CountActiveTasksInFarm 

func (s *Store) CountActiveTasksInFarm(ctx context.Context, farmID string) (int, error)

CountActiveTasksInFarm implements store.TaskStore.

func (*Store) CountActiveTasksInQueue 

func (s *Store) CountActiveTasksInQueue(ctx context.Context, queueID string) (int, error)

CountActiveTasksInQueue implements store.TaskStore.

func (*Store) CountIdleWorkers 

func (s *Store) CountIdleWorkers(ctx context.Context, farmID string) (int, error)

CountIdleWorkers implements store.WorkerStore. When farmID is empty all farms are counted.

func (*Store) CountReadyTasksByQueue 

func (s *Store) CountReadyTasksByQueue(ctx context.Context, farmID string) (map[string]int, error)

CountReadyTasksByQueue implements store.TaskStore. Returns a map of queue ID → ready-task count for the given farm. Queues with zero ready tasks are omitted.

func (*Store) CountTasksByJob 

func (s *Store) CountTasksByJob(ctx context.Context, jobID string) (map[store.TaskStatus]int, error)

CountTasksByJob implements store.TaskStore. Returns the number of tasks for the given job keyed by status. Statuses with zero tasks are omitted from the returned map.

func (*Store) CreateClaim 

func (s *Store) CreateClaim(ctx context.Context, claim store.UsageClaim) (store.UsageClaim, error)

CreateClaim implements store.UsageClaimStore.

func (*Store) CreateFarm 

func (s *Store) CreateFarm(ctx context.Context, farm store.Farm) (store.Farm, error)

CreateFarm implements store.FarmStore.

func (*Store) CreateJob 

func (s *Store) CreateJob(ctx context.Context, job store.Job) (store.Job, error)

CreateJob implements store.JobStore.

func (*Store) CreateQueue 

func (s *Store) CreateQueue(ctx context.Context, queue store.Queue) (store.Queue, error)

CreateQueue implements store.QueueStore.

func (*Store) CreateStep 

func (s *Store) CreateStep(ctx context.Context, step store.Step) (store.Step, error)

CreateStep implements store.StepStore.

func (*Store) CreateStorageLocation 

func (s *Store) CreateStorageLocation(ctx context.Context, loc store.StorageLocation) (store.StorageLocation, error)

CreateStorageLocation implements store.StorageLocationStore.

func (*Store) CreateTask 

func (s *Store) CreateTask(ctx context.Context, task store.Task) (store.Task, error)

CreateTask implements store.TaskStore.

func (*Store) CreateTaskAttempt 

func (s *Store) CreateTaskAttempt(ctx context.Context, attempt store.TaskAttempt) (store.TaskAttempt, error)

CreateTaskAttempt implements store.TaskAttemptStore.

func (*Store) CreateTaskLog 

func (s *Store) CreateTaskLog(ctx context.Context, log store.TaskLog) (store.TaskLog, error)

CreateTaskLog implements store.TaskLogStore.

func (*Store) CreateUsagePool 

func (s *Store) CreateUsagePool(ctx context.Context, pool store.UsagePool) (store.UsagePool, error)

CreateUsagePool implements store.UsagePoolStore.

func (*Store) DeleteFarm 

func (s *Store) DeleteFarm(ctx context.Context, id string) error

DeleteFarm implements store.FarmStore.

func (*Store) DeleteJob 

func (s *Store) DeleteJob(ctx context.Context, id string) error

DeleteJob implements store.JobStore. The cascade runs in a single transaction; if the jobs-row delete affects zero rows the job did not exist and the transaction is rolled back with store.ErrNotFound.

func (*Store) DeleteOfflineWorkersBefore 

func (s *Store) DeleteOfflineWorkersBefore(ctx context.Context, cutoff time.Time) ([]store.Worker, error)

DeleteOfflineWorkersBefore implements store.WorkerStore.

func (*Store) DeleteQueue 

func (s *Store) DeleteQueue(ctx context.Context, id string) error

DeleteQueue implements store.QueueStore.

func (*Store) DeleteStorageLocation 

func (s *Store) DeleteStorageLocation(ctx context.Context, id string) error

DeleteStorageLocation implements store.StorageLocationStore.

func (*Store) DeleteTerminalJobsBefore 

func (s *Store) DeleteTerminalJobsBefore(
	ctx context.Context, cutoff time.Time, includeFailed bool,
) ([]store.DeletedJob, error)

DeleteTerminalJobsBefore implements store.JobStore. It selects the eligible job IDs first, then deletes each via the shared cascade, all inside one transaction so a partially-applied sweep can never leave orphaned child rows.

func (*Store) DeleteUsagePool 

func (s *Store) DeleteUsagePool(ctx context.Context, id string) error

DeleteUsagePool implements store.UsagePoolStore.

func (*Store) DeleteWorker 

func (s *Store) DeleteWorker(ctx context.Context, id string) error

DeleteWorker implements store.WorkerStore.

func (*Store) DemoteStalledJobs 

func (s *Store) DemoteStalledJobs(ctx context.Context, now time.Time) ([]string, error)

DemoteStalledJobs implements store.JobStore.

The single UPDATE ... RETURNING statement is atomic: a concurrent task transition to assigned/running cannot slip between the EXISTS check and the write, so a job that has just been re-activated is never spuriously demoted.

func (*Store) GetFarm 

func (s *Store) GetFarm(ctx context.Context, id string) (store.Farm, error)

GetFarm implements store.FarmStore.

func (*Store) GetJob 

func (s *Store) GetJob(ctx context.Context, id string) (store.Job, error)

GetJob implements store.JobStore.

func (*Store) GetQueue 

func (s *Store) GetQueue(ctx context.Context, id string) (store.Queue, error)

GetQueue implements store.QueueStore.

func (*Store) GetStep 

func (s *Store) GetStep(ctx context.Context, id string) (store.Step, error)

GetStep implements store.StepStore.

func (*Store) GetStorageLocation 

func (s *Store) GetStorageLocation(ctx context.Context, id string) (store.StorageLocation, error)

GetStorageLocation implements store.StorageLocationStore.

func (*Store) GetStorageLocationByName 

func (s *Store) GetStorageLocationByName(ctx context.Context, name string) (store.StorageLocation, error)

GetStorageLocationByName implements store.StorageLocationStore.

func (*Store) GetTask 

func (s *Store) GetTask(ctx context.Context, id string) (store.Task, error)

GetTask implements store.TaskStore.

func (*Store) GetTaskAttempt 

func (s *Store) GetTaskAttempt(ctx context.Context, id string) (store.TaskAttempt, error)

GetTaskAttempt implements store.TaskAttemptStore.

func (*Store) GetUsagePool 

func (s *Store) GetUsagePool(ctx context.Context, id string) (store.UsagePool, error)

GetUsagePool implements store.UsagePoolStore.

func (*Store) GetWorker 

func (s *Store) GetWorker(ctx context.Context, id string) (store.Worker, error)

GetWorker implements store.WorkerStore.

func (*Store) LatestTaskAttempt 

func (s *Store) LatestTaskAttempt(ctx context.Context, taskID string) (store.TaskAttempt, error)

LatestTaskAttempt implements store.TaskAttemptStore.

func (*Store) LeaseReadyTask 

func (s *Store) LeaseReadyTask(ctx context.Context, taskID, workerID string, now time.Time) (bool, error)

LeaseReadyTask implements store.TaskStore.

func (*Store) ListAuditEntries 

func (s *Store) ListAuditEntries(ctx context.Context, entityType, entityID string) ([]store.AuditEntry, error)

ListAuditEntries implements store.AuditStore. Passing empty strings for both entityType and entityID returns all entries.

func (*Store) ListFarms 

func (s *Store) ListFarms(ctx context.Context) ([]store.Farm, error)

ListFarms implements store.FarmStore.

func (*Store) ListJobs 

func (s *Store) ListJobs(ctx context.Context, opts store.ListJobsOptions) (store.Page[store.Job], error)

ListJobs implements store.JobStore. Dynamic filters use parameterised ad-hoc queries due to the variable WHERE clause. The sort column is looked up from a hard-coded allow-list before being interpolated into the SQL to prevent injection.

func (*Store) ListQueues 

func (s *Store) ListQueues(ctx context.Context, opts store.ListQueuesOptions) (store.Page[store.Queue], error)

ListQueues implements store.QueueStore. Dynamic filters use parameterised ad-hoc queries; the sort column is looked up from a hard-coded allow-list to prevent injection.

func (*Store) ListReadyTasks 

func (s *Store) ListReadyTasks(ctx context.Context, farmID string, limit int) ([]store.Task, error)

ListReadyTasks implements store.TaskStore.

func (*Store) ListStaleWorkers 

func (s *Store) ListStaleWorkers(ctx context.Context, before time.Time) ([]store.Worker, error)

ListStaleWorkers implements store.WorkerStore.

func (*Store) ListSteps 

func (s *Store) ListSteps(ctx context.Context, jobID string) ([]store.Step, error)

ListSteps implements store.StepStore.

func (*Store) ListStorageLocations 

func (s *Store) ListStorageLocations(ctx context.Context) ([]store.StorageLocation, error)

ListStorageLocations implements store.StorageLocationStore.

func (*Store) ListTaskAttempts 

func (s *Store) ListTaskAttempts(ctx context.Context, taskID string) ([]store.TaskAttempt, error)

ListTaskAttempts implements store.TaskAttemptStore.

func (*Store) ListTaskLogs 

func (s *Store) ListTaskLogs(ctx context.Context, attemptID string, afterNATSSeq int64, limit int) ([]store.TaskLog, error)

ListTaskLogs implements store.TaskLogStore.

func (*Store) ListTasks 

func (s *Store) ListTasks(ctx context.Context, opts store.ListTasksOptions) (store.Page[store.Task], error)

ListTasks implements store.TaskStore. Dynamic filters use parameterised ad-hoc queries due to the variable WHERE clause. The sort column is looked up from a hard-coded allow-list.

func (*Store) ListUsagePoolUtilization 

func (s *Store) ListUsagePoolUtilization(ctx context.Context) ([]store.UsagePoolUtilization, error)

ListUsagePoolUtilization implements store.UsagePoolStore.

func (*Store) ListUsagePools 

func (s *Store) ListUsagePools(ctx context.Context) ([]store.UsagePool, error)

ListUsagePools implements store.UsagePoolStore.

func (*Store) ListWorkers 

func (s *Store) ListWorkers(ctx context.Context, opts store.ListWorkersOptions) (store.Page[store.Worker], error)

ListWorkers implements store.WorkerStore. Dynamic filters are applied via parameterised ad-hoc queries because the WHERE clause varies; this is safe as all values are bound via placeholders. The sort column is looked up from a hard-coded allow-list.

func (*Store) Ping 

func (s *Store) Ping(ctx context.Context) error

Ping verifies the database connection is still alive. Used by health checks.

func (*Store) ReclaimStaleAssignedTasks 

func (s *Store) ReclaimStaleAssignedTasks(ctx context.Context, cutoff time.Time) ([]store.Task, error)

ReclaimStaleAssignedTasks implements store.TaskStore.

The SELECT and UPDATE run inside a single SQLite transaction so a concurrent "running" status update cannot slip a task out of 'assigned' between the observation and the reset; the UPDATE's status guard keeps the returned set and the reclaimed set identical. Mirrors Store.CancelJobTasks.

func (*Store) ReclaimWorkerTasks 

func (s *Store) ReclaimWorkerTasks(ctx context.Context, workerID string) (int, error)

ReclaimWorkerTasks implements store.TaskStore.

func (*Store) RegisterWorker 

func (s *Store) RegisterWorker(ctx context.Context, worker store.Worker) (store.Worker, error)

RegisterWorker implements store.WorkerStore.

func (*Store) ReleaseAttemptClaims 

func (s *Store) ReleaseAttemptClaims(ctx context.Context, taskAttemptID string, releasedAt time.Time) (int, error)

ReleaseAttemptClaims implements store.UsageClaimStore.

func (*Store) ReleaseClaim 

func (s *Store) ReleaseClaim(ctx context.Context, id string, releasedAt time.Time) error

ReleaseClaim implements store.UsageClaimStore.

func (*Store) ReleaseJobClaims 

func (s *Store) ReleaseJobClaims(ctx context.Context, jobID string, releasedAt time.Time) (int, error)

ReleaseJobClaims implements store.UsageClaimStore.

func (*Store) RetryTasks 

func (s *Store) RetryTasks(ctx context.Context, jobID string, taskIDs []string, now time.Time) ([]store.Task, error)

RetryTasks implements store.TaskStore.

func (*Store) StartCheckpointer 

func (s *Store) StartCheckpointer(ctx context.Context, interval time.Duration, logger *slog.Logger)

StartCheckpointer launches a background goroutine that calls [Checkpoint] every interval. The goroutine exits when ctx is canceled, running one final checkpoint before returning so the WAL is fully flushed before the store is closed.

Errors are logged but do not stop the loop; transient checkpoint failures are not fatal. The caller does not need to wrap this in a goroutine — it manages its own.

func (*Store) TerminateWorkerAttempts 

func (s *Store) TerminateWorkerAttempts(ctx context.Context, workerID string, status store.AttemptStatus, endedAt time.Time) (int, error)

TerminateWorkerAttempts implements store.TaskAttemptStore.

func (*Store) TransitionStepPendingTasks 

func (s *Store) TransitionStepPendingTasks(ctx context.Context, stepID string, to store.TaskStatus) ([]store.Task, error)

TransitionStepPendingTasks implements store.TaskStore.

The UPDATE ... RETURNING runs as one statement so the transition is atomic and covers every pending task of the step regardless of count.

func (*Store) TryClaimSlots 

func (s *Store) TryClaimSlots(
	ctx context.Context,
	taskAttemptID string,
	claims []store.UsagePoolClaim,
	claimedAt time.Time,
) error

TryClaimSlots implements store.UsageClaimStore.

It opens a transaction, counts active claims for each pool in claims, and either inserts all claim rows (all pools have capacity) or rolls back and returns store.ErrUsageAtCapacity (at least one pool is saturated).

The transaction is serialized by the single-connection pool (SetMaxOpenConns(1)) so no other goroutine can modify claim counts between the count check and the inserts.

func (*Store) UpdateFarm 

func (s *Store) UpdateFarm(ctx context.Context, farm store.Farm) (store.Farm, error)

UpdateFarm implements store.FarmStore.

func (*Store) UpdateJob 

func (s *Store) UpdateJob(ctx context.Context, job store.Job) (store.Job, error)

UpdateJob implements store.JobStore. Only mutable user-settable fields are written (farm_id, queue_id, name, owner, submitter, priority, project, raw_template, template_format). status, started_at, and completed_at are never touched here; use UpdateJobStatus or CancelJobStatus for those.

func (*Store) UpdateJobStatus 

func (s *Store) UpdateJobStatus(ctx context.Context, id string, status store.JobStatus) error

UpdateJobStatus implements store.JobStore.

func (*Store) UpdateQueue 

func (s *Store) UpdateQueue(ctx context.Context, queue store.Queue) (store.Queue, error)

UpdateQueue implements store.QueueStore.

func (*Store) UpdateStepStatus 

func (s *Store) UpdateStepStatus(ctx context.Context, id string, status store.StepStatus) error

UpdateStepStatus implements store.StepStore.

func (*Store) UpdateStorageLocation 

func (s *Store) UpdateStorageLocation(ctx context.Context, loc store.StorageLocation) (store.StorageLocation, error)

UpdateStorageLocation implements store.StorageLocationStore.

func (*Store) UpdateTaskAttempt 

func (s *Store) UpdateTaskAttempt(ctx context.Context, attempt store.TaskAttempt) (store.TaskAttempt, error)

UpdateTaskAttempt implements store.TaskAttemptStore. If attempt.SessionID is non-empty it is written to the record; an empty value is treated as "no change" via COALESCE so callers that do not have a session ID (e.g. the cancellation path) do not overwrite an existing one.

func (*Store) UpdateTaskStatus 

func (s *Store) UpdateTaskStatus(ctx context.Context, id string, status store.TaskStatus) error

UpdateTaskStatus implements store.TaskStore.

func (*Store) UpdateUsagePool 

func (s *Store) UpdateUsagePool(ctx context.Context, pool store.UsagePool) (store.UsagePool, error)

UpdateUsagePool implements store.UsagePoolStore.

func (*Store) UpdateWorker 

func (s *Store) UpdateWorker(ctx context.Context, worker store.Worker) (store.Worker, error)

UpdateWorker implements store.WorkerStore.

func (*Store) UpdateWorkerHeartbeat 

func (s *Store) UpdateWorkerHeartbeat(ctx context.Context, id string, at time.Time) error

UpdateWorkerHeartbeat implements store.WorkerStore.

func (*Store) UpdateWorkerStatus 

func (s *Store) UpdateWorkerStatus(ctx context.Context, id string, status store.WorkerStatus) error

UpdateWorkerStatus implements store.WorkerStore.

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.