queue

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Sep 9, 2026 License: MIT Imports: 13 Imported by: 0

Documentation

Overview

Package queue provides the durable task store and the Queue facade.

The Store interface is the persistence boundary; the sqlite Store is the embedded default. Every mutating operation also appends a fact to the journal in the same transaction, so the journal is always a complete, consistent history of the queue.

Index

Constants

This section is empty.

Variables

View Source 
var ErrEmptyType = errors.New("queue: task type must not be empty")

ErrEmptyType is returned by Enqueue when New.Type is empty (Normalize cannot invent a type; callers must choose an executor).

View Source 
var ErrNoTaskDue = errors.New("queue: no due task")

ErrNoTaskDue is returned by ClaimDue when nothing is claimable right now.

Functions

This section is empty.

Types

type ArchiveStats 

type ArchiveStats struct {
	Hot       int64 // facts in the hot table
	Archived  int64 // facts in facts_archive
	Watermark int64 // highest seq ever archived (-1 = never)
}

ArchiveStats reports hot vs archived fact counts and the compaction watermark (highest archived seq; -1 when nothing was archived yet).

type Filter 

type Filter struct {
	Project *string
	Status  *task.Status
	Type    *string
	// Query is a case-insensitive substring search over id, type, project,
	// payload, lease owner and last error — pushed into SQL LIKE, not a
	// post-filter.
	Query string
	Limit int
	// Offset skips the first Offset matches (pagination); applied after
	// ordering. Meaningful together with Limit.
	Offset int
	// SeverityOrder orders by display severity (dead, running, pending,
	// cancelled, completed), newest first within a status — the dashboard
	// table order, stable across pages. Default order stays priority then
	// age (the queue's fairness order).
	SeverityOrder bool
	// Sort overrides the ordering with an allowlisted column sort for the
	// dashboard's sortable headers: "age-asc", "age-desc",
	// "priority-asc", "priority-desc", "attempts-asc", "attempts-desc".
	// Unknown values fall back to the default order (never interpolated
	// into SQL).
	Sort string
	// Since restricts the listing to tasks created at or after this time
	// (inclusive) — pushed into SQL as a created_at comparison, not a
	// post-filter.
	Since *time.Time
}

Filter selects tasks for List.

type PostgresStore 

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

PostgresStore is the networked Store twin of SQLiteStore (ADR-0007): the same facts-first semantics over PostgreSQL, for deployments where many producers/workers share a queue across machines. Claims use SELECT ... FOR UPDATE SKIP LOCKED instead of SQLite's single serialized writer: competing workers lock disjoint rows instead of queueing behind one connection.

Storage mapping mirrors SQLite exactly (unix-milli BIGINT timestamps, deps table, partial unique dedup index) so the projections and the journal remain byte-compatible across backends.

func OpenPostgres 

func OpenPostgres(ctx context.Context, dsn string, maxConns int32) (*PostgresStore, error)

OpenPostgres connects to dsn (e.g. "postgres://user:pass@host:5432/db"), applies the schema, and returns a ready store. maxConns bounds the pool (0 = pgx default).

func (*PostgresStore) Cancel 

func (s *PostgresStore) Cancel(ctx context.Context, id task.ID, reason string) error

Cancel withdraws a Pending task. A non-empty reason is stored in the task.cancelled fact detail ("reason" key).

func (*PostgresStore) CancelOwned 

func (s *PostgresStore) CancelOwned(ctx context.Context, id task.ID, owner string) error

CancelOwned finalizes a cooperative cancel (lease holder). The operator's reason (from the cancel-requested fact) is carried onto the cancelled fact.

func (*PostgresStore) CancelRequested 

func (s *PostgresStore) CancelRequested(ctx context.Context, id task.ID) (bool, error)

CancelRequested reports a pending cooperative cancel request.

func (*PostgresStore) CancelRunning 

func (s *PostgresStore) CancelRunning(ctx context.Context, id task.ID, reason string) error

CancelRunning records the cooperative cancel request (idempotent fact). A non-empty reason rides the request fact's detail and is carried onto the final task.cancelled fact by CancelOwned / the reclaim finalize.

func (*PostgresStore) ClaimDue 

func (s *PostgresStore) ClaimDue(ctx context.Context, owner string, lease time.Duration) (task.Task, error)

ClaimDue atomically claims one due task for owner: FOR UPDATE SKIP LOCKED keeps competing workers on disjoint rows (the multi-machine replacement for SQLite's single serialized writer). Semantics otherwise match SQLiteStore: deps gate, expired-lease reclaim with task.released, pending cooperative cancels finalized at reclaim.

func (*PostgresStore) Close 

func (s *PostgresStore) Close() error

Close releases the pool.

func (*PostgresStore) Complete 

func (s *PostgresStore) Complete(ctx context.Context, id task.ID, owner string, result json.RawMessage) error

Complete marks a Running task Completed.

func (*PostgresStore) CountFacts 

func (s *PostgresStore) CountFacts(ctx context.Context, ftype journal.FactType, since time.Time) (int64, error)

CountFacts counts facts of one type since a time.

func (*PostgresStore) CountTasks 

func (s *PostgresStore) CountTasks(ctx context.Context, f Filter) (int, error)

CountTasks counts tasks matching the filter.

func (*PostgresStore) Enqueue 

func (s *PostgresStore) Enqueue(ctx context.Context, n task.New) (task.Task, error)

Enqueue persists a new task and records task.enqueued; dedup keys make it idempotent (the partial unique index is the arbiter).

func (*PostgresStore) Facts 

func (s *PostgresStore) Facts(ctx context.Context, after int64, limit int) ([]journal.Fact, error)

Facts exposes the journal in Seq order after the cursor.

func (*PostgresStore) FactsForTask 

func (s *PostgresStore) FactsForTask(ctx context.Context, id string, limit int) ([]journal.Fact, error)

FactsForTask returns one task's facts in Seq order; limit > 0 bounds to the MOST RECENT n facts (same contract as the SQLite store).

func (*PostgresStore) Fail 

func (s *PostgresStore) Fail(
	ctx context.Context,
	id task.ID,
	owner string,
	errText string,
	backoff time.Duration,
	evidence json.RawMessage,
) error

Fail records a failed attempt: retry with backoff or dead-letter.

func (*PostgresStore) FailPermanent 

func (s *PostgresStore) FailPermanent(
	ctx context.Context,
	id task.ID,
	owner string,
	errText string,
	evidence json.RawMessage,
) error

FailPermanent dead-letters regardless of the attempt budget.

func (*PostgresStore) Get 

func (s *PostgresStore) Get(ctx context.Context, id task.ID) (task.Task, error)

Get returns the current task record.

func (*PostgresStore) HeadSeq 

func (s *PostgresStore) HeadSeq(ctx context.Context) (int64, error)

HeadSeq returns the highest fact seq (0 when empty).

func (*PostgresStore) Heartbeat 

func (s *PostgresStore) Heartbeat(ctx context.Context, id task.ID, owner string, extend time.Duration) error

Heartbeat extends the lease of a Running task held by owner.

func (*PostgresStore) LastFacts 

func (s *PostgresStore) LastFacts(ctx context.Context, limit int) ([]journal.Fact, error)

LastFacts returns the most recent facts in ascending order.

func (*PostgresStore) List 

func (s *PostgresStore) List(ctx context.Context, f Filter) ([]task.Task, error)

List returns tasks matching the filter.

func (*PostgresStore) ListWatermarks 

func (s *PostgresStore) ListWatermarks(ctx context.Context) ([]WatermarkEntry, error)

ListWatermarks returns every consumer cursor, by consumer name — the admin read behind `tq watermarks show`.

func (*PostgresStore) MarkOrphaned 

func (s *PostgresStore) MarkOrphaned(ctx context.Context, cutoff time.Time) (int, error)

MarkOrphaned records stranded expired-lease Running tasks (idempotent).

func (*PostgresStore) ProjectCounts 

func (s *PostgresStore) ProjectCounts(ctx context.Context) (map[string]map[task.Status]int, error)

ProjectCounts counts tasks per project per status.

func (*PostgresStore) Requeue 

func (s *PostgresStore) Requeue(
	ctx context.Context,
	id task.ID,
	owner string,
	errText string,
	delay time.Duration,
) error

Requeue returns a claimed task to Pending without counting an attempt.

func (*PostgresStore) RescueDead 

func (s *PostgresStore) RescueDead(ctx context.Context, id task.ID, maxAttempts int) error

RescueDead re-queues a Dead task with a fresh attempt budget.

func (*PostgresStore) SaveWatermark 

func (s *PostgresStore) SaveWatermark(ctx context.Context, consumer string, seq int64) error

SaveWatermark checkpoints a consumer cursor as a monotonic upsert: the stored seq never regresses. Consumer progress, not task state, so no fact is appended.

func (*PostgresStore) SetWatermark 

func (s *PostgresStore) SetWatermark(ctx context.Context, consumer string, seq int64) error

SetWatermark overwrites a consumer cursor unconditionally — the ops rewind hatch (`tq watermarks set`); deliberately bypasses the monotonic guard because a rewind is an intentional force-replay.

func (*PostgresStore) StatusCounts 

func (s *PostgresStore) StatusCounts(ctx context.Context) (map[task.Status]int, error)

StatusCounts counts tasks per status.

func (*PostgresStore) Watermark 

func (s *PostgresStore) Watermark(ctx context.Context, consumer string) (int64, bool, error)

Watermark returns the persisted read cursor for a journal consumer and whether it ever checkpointed. seq 0 with exists=true is a valid cursor.

type Queue 

type Queue struct {
	Store
}

Queue is the facade most consumers use: a Store plus convenience methods.

func New 

func New(s Store) *Queue

New wraps a Store.

func (*Queue) Enqueue 

func (q *Queue) Enqueue(ctx context.Context, n task.New) (task.Task, error)

Enqueue normalized-and-enqueues a task.

type RequeueEvidence 

type RequeueEvidence struct {
	Reason  string `json:"reason"`
	RetryIn int64  `json:"retry_in_ms"`
}

RequeueEvidence is the structured detail on task.requeued facts: why the executor refused to start and how long the task waits before it becomes claimable again (01:48 report f2: the reason was a plain error string).

type SQLiteStore 

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

SQLiteStore is the embedded, durable Store. One queue per database file.

Concurrency model: a single serialized write connection (MaxOpenConns(1)) plus WAL journal mode. All task mutations and their journal facts happen in one transaction, so the journal can never disagree with the task table. Multiple processes may open the same file; busy_timeout + WAL serialize cross-process writers.

func OpenSQLite 

func OpenSQLite(path string, opts ...StoreOption) (*SQLiteStore, error)

OpenSQLite opens (creating if needed) the queue database at path.

func (*SQLiteStore) ArchiveFactsBefore 

func (s *SQLiteStore) ArchiveFactsBefore(ctx context.Context, cutoff int64) (int64, error)

ArchiveFactsBefore moves the facts of TERMINAL tasks (completed, dead, cancelled) whose highest fact seq is below cutoff into facts_archive — the hot-cold compaction prototype (ADR-0006). One transaction, one statement pair per task set; projections (tasks table) are untouched. Returns how many facts moved. NOT on the Store interface: an admin operation, not a queue operation.

func (*SQLiteStore) ArchiveSummary 

func (s *SQLiteStore) ArchiveSummary(ctx context.Context) (ArchiveStats, error)

ArchiveSummary reports hot/archived fact counts and the compaction watermark (highest archived seq; -1 when nothing was archived yet).

func (*SQLiteStore) Cancel 

func (s *SQLiteStore) Cancel(ctx context.Context, id task.ID, reason string) error

Cancel withdraws a Pending task. A non-empty reason is stored in the task.cancelled fact detail ("reason" key).

func (*SQLiteStore) CancelOwned 

func (s *SQLiteStore) CancelOwned(ctx context.Context, id task.ID, owner string) error

CancelOwned finalizes a cooperative cancel: Running -> Cancelled, written by the lease-holding worker after it stopped the execution. The operator's reason (from the cancel-requested fact) is carried onto the cancelled fact.

func (*SQLiteStore) CancelRequested 

func (s *SQLiteStore) CancelRequested(ctx context.Context, id task.ID) (bool, error)

CancelRequested reports whether a cooperative cancel request is pending.

func (*SQLiteStore) CancelRunning 

func (s *SQLiteStore) CancelRunning(ctx context.Context, id task.ID, reason string) error

CancelRunning records a cooperative cancel request for a Running task. The task.cancel-requested fact IS the flag — no task-row column mirrors it (facts-first). A non-empty reason rides the request fact's detail and is carried onto the final task.cancelled fact by CancelOwned / the reclaim finalize. Idempotent: a second request appends nothing.

func (*SQLiteStore) ClaimDue 

func (s *SQLiteStore) ClaimDue(ctx context.Context, owner string, lease time.Duration) (task.Task, error)

ClaimDue atomically claims one due task for owner.

func (*SQLiteStore) Close 

func (s *SQLiteStore) Close() error

Close releases the database connection.

func (*SQLiteStore) Complete 

func (s *SQLiteStore) Complete(ctx context.Context, id task.ID, owner string, result json.RawMessage) error

Complete marks a Running task Completed.

func (*SQLiteStore) CountFacts 

func (s *SQLiteStore) CountFacts(ctx context.Context, ftype journal.FactType, since time.Time) (int64, error)

CountFacts counts facts of one type recorded at or after since.

func (*SQLiteStore) CountTasks 

func (s *SQLiteStore) CountTasks(ctx context.Context, f Filter) (int, error)

CountTasks counts the tasks matching the filter (COUNT(*) pushdown).

func (*SQLiteStore) Enqueue 

func (s *SQLiteStore) Enqueue(ctx context.Context, n task.New) (task.Task, error)

Enqueue persists a new task and records task.enqueued. When New.DedupKey is set and a task with that key already exists, the stored task is returned unchanged — no duplicate row, no duplicate fact (idempotent enqueue).

func (*SQLiteStore) Facts 

func (s *SQLiteStore) Facts(ctx context.Context, after int64, limit int) ([]journal.Fact, error)

Facts returns journal facts with Seq > after, ascending, bounded to the most recent limit when > 0. The seq primary key makes the cursor scan O(limit) regardless of journal size.

func (*SQLiteStore) FactsForTask 

func (s *SQLiteStore) FactsForTask(ctx context.Context, id string, limit int) ([]journal.Fact, error)

FactsForTask returns one task's facts in Seq order, bounded to the most recent limit when > 0. Served by idx_facts_task (task_id, seq).

func (*SQLiteStore) Fail 

func (s *SQLiteStore) Fail(
	ctx context.Context,
	id task.ID,
	owner string,
	errText string,
	backoff time.Duration,
	evidence json.RawMessage,
) error

Fail records a failed attempt: retry with backoff or dead-letter.

func (*SQLiteStore) FailPermanent 

func (s *SQLiteStore) FailPermanent(
	ctx context.Context,
	id task.ID,
	owner string,
	errText string,
	evidence json.RawMessage,
) error

FailPermanent dead-letters immediately: a permanent error means the identical retry would fail identically, so the remaining attempt budget is worthless (and, for agent tasks, expensive). The failing attempt is still counted. Facts: task.failed + task.dead-lettered with class "permanent".

func (*SQLiteStore) Get 

func (s *SQLiteStore) Get(ctx context.Context, id task.ID) (task.Task, error)

Get returns the current task record.

func (*SQLiteStore) HeadSeq 

func (s *SQLiteStore) HeadSeq(ctx context.Context) (int64, error)

HeadSeq returns the current highest fact Seq (0 when empty).

func (*SQLiteStore) Heartbeat 

func (s *SQLiteStore) Heartbeat(ctx context.Context, id task.ID, owner string, extend time.Duration) error

Heartbeat extends the lease of a Running task held by owner.

func (*SQLiteStore) LastFacts 

func (s *SQLiteStore) LastFacts(ctx context.Context, limit int) ([]journal.Fact, error)

LastFacts returns the most recent limit facts in ascending Seq order.

func (*SQLiteStore) List 

func (s *SQLiteStore) List(ctx context.Context, f Filter) ([]task.Task, error)

func (*SQLiteStore) ListWatermarks 

func (s *SQLiteStore) ListWatermarks(ctx context.Context) ([]WatermarkEntry, error)

ListWatermarks returns every consumer cursor, by consumer name — the admin read behind `tq watermarks show`.

func (*SQLiteStore) MarkOrphaned 

func (s *SQLiteStore) MarkOrphaned(ctx context.Context, cutoff time.Time) (int, error)

MarkOrphaned appends one task.orphaned fact per stranded Running task (lease expired before the cutoff, no orphaned fact yet). Observation only: the task stays Running until a reclaim; the fact explains why it is stranded (worker died, pool down). Idempotent per task.

func (*SQLiteStore) ProjectCounts 

func (s *SQLiteStore) ProjectCounts(ctx context.Context) (map[string]map[task.Status]int, error)

ProjectCounts counts tasks per project per status in one GROUP BY.

func (*SQLiteStore) Requeue 

func (s *SQLiteStore) Requeue(
	ctx context.Context,
	id task.ID,
	owner string,
	errText string,
	delay time.Duration,
) error

Requeue returns a claimed task to Pending without counting an attempt: the executor refused to start (preflight), so the task itself is fine and the environment is expected to become ready later. Claimable again after delay. Fact: task.requeued.

func (*SQLiteStore) RescueDead 

func (s *SQLiteStore) RescueDead(ctx context.Context, id task.ID, maxAttempts int) error

RescueDead re-queues a Dead task with a fresh attempt budget (DLQ rescue).

func (*SQLiteStore) SaveWatermark 

func (s *SQLiteStore) SaveWatermark(ctx context.Context, consumer string, seq int64) error

SaveWatermark checkpoints a consumer cursor as a monotonic upsert: the stored seq never regresses, so a lagging or misconfigured second process cannot drag a consumer backwards. Checkpointing is consumer progress, not task state, so no fact is appended.

func (*SQLiteStore) SetWatermark 

func (s *SQLiteStore) SetWatermark(ctx context.Context, consumer string, seq int64) error

SetWatermark overwrites a consumer cursor unconditionally — the ops rescue hatch behind `tq watermarks set`. Unlike SaveWatermark it MAY move the cursor backwards: a rewind forces replay, and downstream idempotency keys (seq-derived) make replay safe. It deliberately bypasses the monotonic runtime guard; use it knowing that.

func (*SQLiteStore) StatusCounts 

func (s *SQLiteStore) StatusCounts(ctx context.Context) (map[task.Status]int, error)

StatusCounts counts tasks per status in one GROUP BY.

func (*SQLiteStore) Watermark 

func (s *SQLiteStore) Watermark(ctx context.Context, consumer string) (int64, bool, error)

Watermark returns the persisted read cursor for a journal consumer and whether it ever checkpointed — the resume point for bridges and sweepers. seq 0 with exists=true is a valid cursor ("consumed nothing yet").

type Store 

type Store interface {
	// Enqueue persists a new task (ID and defaults assigned here) and records
	// the task.enqueued fact.
	Enqueue(ctx context.Context, n task.New) (task.Task, error)
	// ClaimDue atomically claims at most one due task for owner: pending,
	// NotBefore passed, all deps completed. Sets Running + lease. Returns
	// ErrNoTaskDue when nothing is claimable.
	ClaimDue(ctx context.Context, owner string, lease time.Duration) (task.Task, error)
	// Complete marks a Running task Completed (lease must be held) and records
	// the task.completed fact.
	Complete(ctx context.Context, id task.ID, owner string, result json.RawMessage) error
	// Fail records a failed attempt. When attempts remain the task returns to
	// Pending with NotBefore = now + backoff(attempt); otherwise it is
	// Dead-lettered. Facts: task.failed (+ task.dead-lettered). evidence,
	// when non-empty (executor.FailureEvidence JSON), lands on the
	// task.failed fact's detail — the forensics (exit code, output tail)
	// that make a failed attempt debuggable from the journal alone.
	Fail(
		ctx context.Context,
		id task.ID,
		owner string,
		errText string,
		backoff time.Duration,
		evidence json.RawMessage,
	) error
	// FailPermanent dead-letters a Running task immediately, regardless of
	// the attempt budget: the error class makes retrying pointless. The
	// attempt is still counted. Facts: task.failed (carrying evidence)
	// + task.dead-lettered (class "permanent").
	FailPermanent(ctx context.Context, id task.ID, owner string, errText string, evidence json.RawMessage) error
	// Requeue returns a claimed task to Pending WITHOUT counting an
	// attempt; it becomes claimable again after delay. For preflight
	// refusals: the environment was not ready, not the task. Facts:
	// task.requeued.
	Requeue(ctx context.Context, id task.ID, owner string, errText string, delay time.Duration) error
	// Heartbeat extends the lease of a Running task held by owner.
	Heartbeat(ctx context.Context, id task.ID, owner string, extend time.Duration) error
	// Cancel withdraws a Pending task. A non-empty reason is stored in the
	// task.cancelled fact's detail ("reason" key) so the journal records WHY
	// the task was withdrawn.
	Cancel(ctx context.Context, id task.ID, reason string) error
	// CancelRunning records a cooperative cancel request for a Running
	// task: the task.cancel-requested fact is the flag. The executing
	// worker observes it at its next heartbeat, stops the execution, and
	// finalizes with CancelOwned; an expired lease finalizes it at
	// reclaim. A non-empty reason is stored on the request fact's detail
	// and carried onto the final task.cancelled fact. Idempotent — a
	// second request appends nothing.
	CancelRunning(ctx context.Context, id task.ID, reason string) error
	// CancelRequested reports whether a cooperative cancel request is
	// pending for the task — the worker's heartbeat observation query.
	CancelRequested(ctx context.Context, id task.ID) (bool, error)
	// CancelOwned finalizes a cooperative cancel: Running -> Cancelled,
	// recorded by the lease-holding worker after it stopped the execution.
	CancelOwned(ctx context.Context, id task.ID, owner string) error
	// MarkOrphaned appends a task.orphaned fact for every Running task
	// whose lease expired before the cutoff and that has no orphaned fact
	// yet (idempotent). It changes no state — orphans stay Running until a
	// reclaim — it records WHY the task is stranded so the journal can
	// explain it. Returns how many facts were appended.
	MarkOrphaned(ctx context.Context, cutoff time.Time) (int, error)
	// RescueDead re-queues a Dead task with a fresh attempt budget.
	RescueDead(ctx context.Context, id task.ID, maxAttempts int) error
	// Get returns the current task record.
	Get(ctx context.Context, id task.ID) (task.Task, error)
	// List returns tasks matching the filter.
	List(ctx context.Context, f Filter) ([]task.Task, error)
	// Facts exposes the journal (same store, same transaction domain):
	// facts with Seq strictly greater than after, in Seq order. limit
	// bounds the result when > 0; 0 means unbounded (bulk exports).
	Facts(ctx context.Context, after int64, limit int) ([]journal.Fact, error)
	// LastFacts returns the most recent limit facts in ascending Seq
	// order — the bounded read behind feed-style renders. limit <= 0
	// returns the whole journal.
	LastFacts(ctx context.Context, limit int) ([]journal.Fact, error)
	// HeadSeq returns the current highest fact Seq (0 when the journal is
	// empty): the O(1) watermark for tailers, bridges and resume points.
	HeadSeq(ctx context.Context) (int64, error)
	// Watermark returns the persisted read cursor for a journal consumer
	// and whether the consumer ever checkpointed (seq 0 is a valid cursor:
	// "consumed nothing yet"): the resume point for bridges and sweepers
	// after a restart.
	Watermark(ctx context.Context, consumer string) (seq int64, exists bool, err error)
	// SaveWatermark checkpoints a consumer cursor as a monotonic upsert
	// (never regresses). It records consumer progress, not task state, so
	// no fact is appended.
	SaveWatermark(ctx context.Context, consumer string, seq int64) error
	// FactsForTask returns one task's facts in Seq order, bounded to the
	// most recent limit when > 0 (0 = unbounded).
	FactsForTask(ctx context.Context, id string, limit int) ([]journal.Fact, error)
	// CountFacts counts facts of one type recorded at or after since —
	// the SQL pushdown behind spend projections and stats.
	CountFacts(ctx context.Context, ftype journal.FactType, since time.Time) (int64, error)
	// StatusCounts counts tasks per status — the GROUP BY behind dashboard
	// counters: O(statuses) work instead of a full task scan.
	StatusCounts(ctx context.Context) (map[task.Status]int, error)
	// ProjectCounts counts tasks per project per status — the GROUP BY
	// behind the overview chips and per-project views.
	ProjectCounts(ctx context.Context) (map[string]map[task.Status]int, error)
	// CountTasks counts the tasks matching the filter — the COUNT(*)
	// pushdown behind pagination ("page 2 of 14").
	CountTasks(ctx context.Context, f Filter) (int, error)
	// Close releases resources.
	Close() error
}

Store is the persistence boundary for tasks and facts.

type StoreOption 

type StoreOption func(*storeOptions)

StoreOption configures optional SQLiteStore behavior.

func WithProjectExclusivity 

func WithProjectExclusivity() StoreOption

WithProjectExclusivity turns on store-level per-project serialization: ClaimDue will not claim a task whose project already has another running task — across ALL pools and processes sharing the same database file, not just within one pool. This is the per-repo guarantee for agent pools: two agents never work the same repo simultaneously. Every pool sharing the DB must opt in; pools that do not opt in ignore the guard. Tasks with an empty project are exempt (they are not tied to a repo).

type WatermarkEntry 

type WatermarkEntry struct {
	Consumer  string
	Seq       int64
	UpdatedAt int64 // unix millis
}

WatermarkEntry is one consumer cursor row (tq watermarks show).

Source Files

View all Source files

Directories

Path Synopsis
companion module
postgres module
postgresv4 module
sqlite module
sqlitev4 module

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.