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

sqlite

package
v1.9.3 Latest Latest
Warning

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

 Go to latest Published: Oct 26, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package sqlite implements store.Store using the Grove ORM with SQLite. You can use it for embedded applications, CLI tools and standalone services. The caller owns the *grove.DB lifecycle; sqlite never closes it.

Construct the driver before passing the database to the store. Check errors from Open and Migrate in your application: 

import (
    "github.com/xraph/grove"
    "github.com/xraph/grove/driver"
    "github.com/xraph/grove/drivers/sqlitedriver"
    "github.com/xraph/dispatch/store/sqlite"
)

drv := sqlitedriver.New()
err := drv.Open(ctx, "file:dispatch.db", driver.WithPoolSize(10))
// Handle err before continuing.
db, err := grove.Open(drv)
// Handle err before continuing.
store := sqlite.New(db)
err = store.Migrate(ctx)

Write concurrency

SQLite permits one writer at a time per database. Job claims, lease renewals and workflow reopens all compete for that writer, even within one process. Grove's default SQLite profile enables WAL, uses up to ten pooled connections and leaves busy_timeout at zero. A competing writer can therefore return SQLITE_BUSY immediately.

Store operations that use withBusyRetry retry SQLITE_BUSY within a five-second window. Jittered exponential backoff keeps each pause below 48 milliseconds. Earlier caller cancellation prevents further retries. When the internal window expires, the operation returns its last busy error. Non-busy errors propagate without retry. This policy does not cover every database operation and does not guarantee that a contended write will succeed.

The window bounds retry issuance and waits, not the duration of a synchronous driver call. A call that ignores context can return later. If that call succeeds, the store preserves the accepted result even when the deadline has passed.

You can configure connection behavior when opening the driver, though Store does not expose the underlying sql.DB. driver.WithPoolSize sets the pool size. With Grove's SQLite driver, this optional DSN sets a five-second busy timeout on each connection: 

file:dispatch.db?_pragma=busy_timeout(5000)

Dispatch does not add this option by default. A single PRAGMA executed through a pooled handle only configures the connection that executes it. Use the DSN when you need the setting on every connection. Driver-level waits can extend a synchronous call beyond the store's retry window, so choose both settings with your request deadlines in mind. Reducing the pool can reduce contention within one handle; other handles and processes still compete for the writer. For sustained concurrent writes, consider the PostgreSQL store.

Index

Constants

This section is empty.

Variables

View Source 
var Migrations = migrate.NewGroup("dispatch")

Migrations is the grove migration group for the dispatch sqlite store.

Functions

This section is empty.

Types

type Option 

type Option func(*Store)

Option configures the Store.

func WithLogger 

func WithLogger(logger log.Logger) Option

WithLogger sets the logger for the store.

type Store 

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

Store is a grove ORM implementation of store.Store using SQLite dialect. The caller owns the *grove.DB lifecycle; Store never closes it.

func New 

func New(db *grove.DB, opts ...Option) *Store

New creates a new grove store. The caller owns the db lifecycle -- the Store will not close it on Close().

func (*Store) AckEvent 

func (s *Store) AckEvent(ctx context.Context, eventID id.EventID) error

AckEvent acknowledges an event, marking it as consumed.

func (*Store) AcquireCronLock 

func (s *Store) AcquireCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID, ttl time.Duration) (bool, error)

AcquireCronLock attempts to acquire a distributed lock for a cron entry. Uses row-level update to prevent race conditions.

func (*Store) AcquireLeadership 

func (s *Store) AcquireLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error)

AcquireLeadership attempts to become the cluster leader.

func (*Store) ClaimReplay 

func (s *Store) ClaimReplay(ctx context.Context, entryID id.DLQID, jobID id.JobID) error

ClaimReplay marks an unreplayed entry replayed by jobID.

func (*Store) Close 

func (s *Store) Close() error

Close is a no-op because the caller owns the *grove.DB lifecycle.

func (*Store) CountDLQ 

func (s *Store) CountDLQ(ctx context.Context) (int64, error)

CountDLQ returns the total number of entries in the dead letter queue.

func (*Store) CountDLQEntries 

func (s *Store) CountDLQEntries(ctx context.Context, opts dlq.CountOpts) (int64, error)

CountDLQEntries counts dead letter entries under the given filters. FailedBefore is bound as a time.Time, exactly as PurgeDLQ binds it, so the driver writes both sides of failed_at < ? in the same UTC text form and the count matches what a purge at that time would delete.

func (*Store) CountJobs 

func (s *Store) CountJobs(ctx context.Context, opts job.CountOpts) (int64, error)

CountJobs returns the number of jobs matching the given options.

func (*Store) CountRuns 

func (s *Store) CountRuns(ctx context.Context, opts workflow.CountRunsOpts) (int64, error)

CountRuns counts workflow runs by state and exact name.

func (*Store) CreateArtifact 

func (s *Store) CreateArtifact(ctx context.Context, a *artifact.Artifact, link *artifact.Link) error

CreateArtifact inserts an artifact and, when link is non-nil, its first link in a single transaction.

func (*Store) CreateRun 

func (s *Store) CreateRun(ctx context.Context, run *workflow.Run) error

CreateRun persists a new workflow run.

func (*Store) DB 

func (s *Store) DB() *grove.DB

DB returns the underlying *grove.DB for advanced usage.

func (*Store) DeleteCheckpointsAfter 

func (s *Store) DeleteCheckpointsAfter(ctx context.Context, runID id.RunID, afterStep string) error

DeleteCheckpointsAfter removes all checkpoints created after the given step name (by creation order). Used for workflow replay.

func (*Store) DeleteCron 

func (s *Store) DeleteCron(ctx context.Context, entryID id.CronID) error

DeleteCron removes a cron entry by ID.

func (*Store) DeleteDLQ 

func (s *Store) DeleteDLQ(ctx context.Context, entryID id.DLQID) error

DeleteDLQ removes one entry.

func (*Store) DeleteJob 

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

DeleteJob removes a job by ID.

func (*Store) DeleteStaleWorkers 

func (s *Store) DeleteStaleWorkers(ctx context.Context, threshold time.Duration) (int64, error)

DeleteStaleWorkers removes worker rows whose last-seen timestamp is older than the given threshold. Returns the number of rows deleted.

func (*Store) DequeueJobs 

func (s *Store) DequeueJobs(ctx context.Context, opts job.DequeueOpts) ([]*job.Job, error)

DequeueJobs atomically claims up to opts.Limit ready jobs from opts.Queues that fit opts, sets them to running, and returns them ordered by priority descending, then locality-preferred first, then RunAt ascending.

SQLite doesn't support FOR UPDATE SKIP LOCKED, so the claim is a single UPDATE ... WHERE id IN (SELECT ... ORDER BY ... LIMIT ?) RETURNING *: one statement, which SQLite runs inside an implicit immediate transaction with the database write lock held, so two concurrent claimers cannot select the same candidate. That mechanism is unchanged here — the fit predicate is simply another conjunct of the inner SELECT's WHERE, so a job that does not fit is never written to. It stays pending and untouched for the next worker that does have room.

When opts.Grants() the lease columns are additional assignments in that same UPDATE's SET clause, never a follow-up statement. ReclaimExpiredLeases requires lease_expires_at IS NOT NULL, so a row left running with a null expiry is not a row at risk of reclamation — it is one reclamation can never see. See job.DequeueOpts.LeaseUntil.

func (*Store) DeregisterWorker 

func (s *Store) DeregisterWorker(ctx context.Context, workerID id.WorkerID) error

DeregisterWorker removes a worker from the cluster registry.

func (*Store) EnqueueJob 

func (s *Store) EnqueueJob(ctx context.Context, j *job.Job) error

EnqueueJob persists a new job in pending state.

func (*Store) FindArtifactByKey 

func (s *Store) FindArtifactByKey(ctx context.Context, backend, bucket, key string) (*artifact.Artifact, error)

FindArtifactByKey retrieves a live artifact by its storage coordinates.

func (*Store) FindLinkByName 

func (s *Store) FindLinkByName(
	ctx context.Context,
	owner artifact.OwnerRef,
	name string,
) (*artifact.Link, error)

FindLinkByName returns the highest-attempt link for an owner and name, breaking ties by created_at descending — see the artifact.Store doc comment for why ties happen and what the tie-break does and does not fix.

func (*Store) GetArtifact 

func (s *Store) GetArtifact(ctx context.Context, artifactID id.ArtifactID) (*artifact.Artifact, error)

GetArtifact retrieves a live artifact by ID.

func (*Store) GetArtifactRecord 

func (s *Store) GetArtifactRecord(ctx context.Context, artifactID id.ArtifactID) (*artifact.Artifact, error)

GetArtifactRecord returns metadata even after soft deletion.

func (*Store) GetCheckpoint 

func (s *Store) GetCheckpoint(ctx context.Context, runID id.RunID, stepName string) ([]byte, error)

GetCheckpoint retrieves checkpoint data for a specific workflow step. Returns nil data if no checkpoint exists.

func (*Store) GetCron 

func (s *Store) GetCron(ctx context.Context, entryID id.CronID) (*cron.Entry, error)

GetCron retrieves a cron entry by ID.

func (*Store) GetDLQ 

func (s *Store) GetDLQ(ctx context.Context, entryID id.DLQID) (*dlq.Entry, error)

GetDLQ retrieves a DLQ entry by ID.

func (*Store) GetDLQByJobID 

func (s *Store) GetDLQByJobID(ctx context.Context, jobID id.JobID) (*dlq.Entry, error)

GetDLQByJobID returns the newest entry, by ID, for a failed job. idx_dispatch_dlq_job_id holds the entries for one job in ID order, so this is one index probe.

func (*Store) GetJob 

func (s *Store) GetJob(ctx context.Context, jobID id.JobID) (*job.Job, error)

GetJob retrieves a job by ID.

func (*Store) GetLeader 

func (s *Store) GetLeader(ctx context.Context) (*cluster.Worker, error)

GetLeader returns the current cluster leader, or nil if there is no leader.

func (*Store) GetRun 

func (s *Store) GetRun(ctx context.Context, runID id.RunID) (*workflow.Run, error)

GetRun retrieves a workflow run by ID.

func (*Store) GetWorker 

func (s *Store) GetWorker(ctx context.Context, workerID id.WorkerID) (*cluster.Worker, error)

GetWorker returns one registered worker.

func (*Store) HeartbeatJob 

func (s *Store) HeartbeatJob(ctx context.Context, jobID id.JobID, _ id.WorkerID) error

HeartbeatJob updates the heartbeat timestamp for a running job.

func (*Store) HeartbeatWorker 

func (s *Store) HeartbeatWorker(ctx context.Context, workerID id.WorkerID) error

HeartbeatWorker updates the last-seen timestamp for a worker.

func (*Store) LinkArtifact 

func (s *Store) LinkArtifact(ctx context.Context, link *artifact.Link) error

LinkArtifact records that an owner references an artifact, idempotently.

func (*Store) ListArtifacts 

func (s *Store) ListArtifacts(ctx context.Context, opts artifact.ListOpts) ([]*artifact.Artifact, error)

ListArtifacts returns artifacts matching the given options, newest first.

func (*Store) ListArtifactsByOwner 

func (s *Store) ListArtifactsByOwner(
	ctx context.Context,
	owner artifact.OwnerRef,
	role artifact.Role,
) ([]*artifact.Artifact, error)

ListArtifactsByOwner returns live artifacts linked to an owner.

func (*Store) ListArtifactsPage 

func (s *Store) ListArtifactsPage(ctx context.Context, opts artifact.PageOpts) (artifact.Page, error)

ListArtifactsPage returns artifacts newest first by ID, filtered and paged.

func (*Store) ListCheckpoints 

func (s *Store) ListCheckpoints(ctx context.Context, runID id.RunID) ([]*workflow.Checkpoint, error)

ListCheckpoints returns all checkpoints for a workflow run.

func (*Store) ListChildRuns 

func (s *Store) ListChildRuns(ctx context.Context, parentRunID id.RunID) ([]*workflow.Run, error)

ListChildRuns returns all child workflow runs for a parent.

func (*Store) ListCrons 

func (s *Store) ListCrons(ctx context.Context) ([]*cron.Entry, error)

ListCrons returns all cron entries.

func (*Store) ListDLQ 

func (s *Store) ListDLQ(ctx context.Context, opts dlq.ListOpts) ([]*dlq.Entry, error)

ListDLQ returns DLQ entries matching the given options.

func (*Store) ListDLQPage 

func (s *Store) ListDLQPage(ctx context.Context, opts dlq.PageOpts) (dlq.Page, error)

ListDLQPage returns dead letter entries newest first by ID, filtered and paged.

func (*Store) ListJobUsage 

func (s *Store) ListJobUsage(ctx context.Context, opts job.UsageListOpts) ([]*job.Usage, error)

ListJobUsage returns recorded attempts, most recent first.

func (*Store) ListJobs 

func (s *Store) ListJobs(ctx context.Context, opts job.ListJobsOpts) (job.Page, error)

ListJobs returns jobs newest first by ID, filtered and paged.

func (*Store) ListJobsByState 

func (s *Store) ListJobsByState(ctx context.Context, state job.State, opts job.ListOpts) ([]*job.Job, error)

ListJobsByState returns jobs matching the given state. 

func (s *Store) ListLinks(ctx context.Context, owner artifact.OwnerRef) ([]*artifact.Link, error)

ListLinks returns every link belonging to the given owner.

func (*Store) ListPurgeable 

func (s *Store) ListPurgeable(
	ctx context.Context,
	grace time.Duration,
	limit int,
) ([]*artifact.Artifact, error)

ListPurgeable returns soft-deleted artifacts older than grace.

func (*Store) ListRuns 

func (s *Store) ListRuns(ctx context.Context, opts workflow.ListOpts) ([]*workflow.Run, error)

ListRuns returns workflow runs matching the given options.

func (*Store) ListRunsPage 

func (s *Store) ListRunsPage(ctx context.Context, opts workflow.ListRunsPageOpts) (workflow.RunPage, error)

ListRunsPage returns workflow runs newest first by ID, filtered and paged.

func (*Store) ListWorkers 

func (s *Store) ListWorkers(ctx context.Context) ([]*cluster.Worker, error)

ListWorkers returns all registered workers.

func (*Store) Migrate 

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

Migrate runs programmatic migrations via the grove orchestrator.

func (*Store) Ping 

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

Ping checks database connectivity.

func (*Store) PublishEvent 

func (s *Store) PublishEvent(ctx context.Context, evt *event.Event) error

PublishEvent persists a new event and makes it available for subscribers.

func (*Store) PurgeArtifact 

func (s *Store) PurgeArtifact(ctx context.Context, artifactID id.ArtifactID) error

PurgeArtifact hard-deletes an artifact. Links cascade.

func (*Store) PurgeDLQ 

func (s *Store) PurgeDLQ(ctx context.Context, before time.Time) (int64, error)

PurgeDLQ removes DLQ entries with FailedAt before the given time. Returns the number of entries removed.

func (*Store) PurgeJobUsage 

func (s *Store) PurgeJobUsage(ctx context.Context, before time.Time, limit int) (int64, error)

PurgeJobUsage deletes records older than before, up to limit rows.

SQLite is usually built without DELETE ... LIMIT, so the victims are selected first and deleted by id. That also keeps the batching behaviour identical to Postgres rather than dialect-dependent.

func (*Store) PushDLQ 

func (s *Store) PushDLQ(ctx context.Context, entry *dlq.Entry) error

PushDLQ adds a failed job entry to the dead letter queue. An ID that is already there is refused with dispatch.ErrDLQAlreadyExists. The primary key is the only unique constraint on dispatch_dlq, so a UNIQUE failure here can only be the ID, and isDuplicateKey matches it the way RegisterCron and CreateRun match theirs.

func (*Store) ReapDeadWorkers 

func (s *Store) ReapDeadWorkers(ctx context.Context, threshold time.Duration) ([]*cluster.Worker, error)

ReapDeadWorkers returns workers whose last-seen timestamp is older than the given threshold.

func (*Store) ReapStaleJobs 

func (s *Store) ReapStaleJobs(ctx context.Context, threshold time.Duration) ([]*job.Job, error)

ReapStaleJobs returns running jobs whose last heartbeat — or, for jobs whose worker died before the first heartbeat, whose start time — is older than the given threshold.

func (*Store) ReclaimExpiredLeases 

func (s *Store) ReclaimExpiredLeases(ctx context.Context, limit int) ([]*job.Job, error)

ReclaimExpiredLeases returns expired-lease jobs to pending, fencing their previous holders.

func (*Store) RecordJobUsage 

func (s *Store) RecordJobUsage(ctx context.Context, u *job.Usage) error

RecordJobUsage persists one attempt's measurements.

func (*Store) RegisterCron 

func (s *Store) RegisterCron(ctx context.Context, entry *cron.Entry) error

RegisterCron persists a new cron entry. Returns an error if the name already exists.

func (*Store) RegisterWorker 

func (s *Store) RegisterWorker(ctx context.Context, w *cluster.Worker) error

RegisterWorker adds a new worker to the cluster registry. Uses ON CONFLICT to upsert if the worker already exists.

func (*Store) ReleaseCronLock 

func (s *Store) ReleaseCronLock(ctx context.Context, entryID id.CronID, workerID id.WorkerID) error

ReleaseCronLock releases the distributed lock for a cron entry.

func (*Store) ReleaseReplay 

func (s *Store) ReleaseReplay(ctx context.Context, entryID id.DLQID, jobID id.JobID) error

ReleaseReplay undoes a claim, but only the one jobID made.

func (*Store) RenewLeadership 

func (s *Store) RenewLeadership(ctx context.Context, workerID id.WorkerID, ttl time.Duration) (bool, error)

RenewLeadership extends the leader's hold.

func (*Store) RenewLease 

func (s *Store) RenewLease(
	ctx context.Context,
	jobID id.JobID,
	workerID id.WorkerID,
	epoch int,
	leaseUntil time.Time,
) error

RenewLease extends the lease only if the caller still holds it.

func (*Store) ReopenRun 

func (s *Store) ReopenRun(ctx context.Context, runID id.RunID, expectedGeneration int64) error

ReopenRun moves a finished run back to running, for exactly one caller.

func (*Store) ReplayDLQ 

func (s *Store) ReplayDLQ(ctx context.Context, entryID id.DLQID) error

ReplayDLQ marks a DLQ entry as replayed.

func (*Store) SaveCheckpoint 

func (s *Store) SaveCheckpoint(ctx context.Context, runID id.RunID, stepName string, data []byte) error

SaveCheckpoint persists checkpoint data for a workflow step. If a checkpoint already exists for the same run/step, it is replaced.

func (*Store) SetCronEnabled 

func (s *Store) SetCronEnabled(ctx context.Context, entryID id.CronID, enabled bool, nextRunAt *time.Time) error

SetCronEnabled sets enabled, and next_run_at when one is given.

func (*Store) SubscribeEvent 

func (s *Store) SubscribeEvent(ctx context.Context, name string, timeout time.Duration) (*event.Event, error)

SubscribeEvent waits for an unacked event matching the given name. Uses a polling approach with short intervals.

func (*Store) SweepEphemeral 

func (s *Store) SweepEphemeral(
	ctx context.Context,
	opts artifact.SweepOpts,
) ([]*artifact.Artifact, error)

SweepEphemeral marks eligible ephemeral artifacts as deleted.

func (*Store) SweepOrphans 

func (s *Store) SweepOrphans(
	ctx context.Context,
	cutoff time.Time,
	limit int,
) ([]*artifact.Artifact, error)

SweepOrphans marks link-less ephemeral artifacts created before cutoff.

func (*Store) UpdateArtifact 

func (s *Store) UpdateArtifact(ctx context.Context, a *artifact.Artifact) error

UpdateArtifact persists size, hash, content type, and expiry. Lifecycle, created_at, and deleted_at are deliberately not updatable here.

func (*Store) UpdateCronEntry 

func (s *Store) UpdateCronEntry(ctx context.Context, entry *cron.Entry) error

UpdateCronEntry updates a cron entry (Enabled, NextRunAt, etc.).

func (*Store) UpdateCronLastRun 

func (s *Store) UpdateCronLastRun(ctx context.Context, entryID id.CronID, at time.Time) error

UpdateCronLastRun records when a cron entry last fired.

func (*Store) UpdateCronNextRun 

func (s *Store) UpdateCronNextRun(ctx context.Context, entryID id.CronID, nextRunAt time.Time) error

UpdateCronNextRun sets next_run_at and nothing else the scheduler does not own. In particular it never writes enabled, which is what lets an operator's disable survive a fire that read the entry before it.

func (*Store) UpdateJob 

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

UpdateJob persists changes to an existing job.

func (*Store) UpdateLeasedJob 

func (s *Store) UpdateLeasedJob(ctx context.Context, j *job.Job, workerID id.WorkerID, epoch int) error

UpdateLeasedJob persists j only while the caller still holds the lease.

func (*Store) UpdateRun 

func (s *Store) UpdateRun(ctx context.Context, run *workflow.Run) error

UpdateRun persists changes to an existing workflow run.

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.