sqlite

func (*Store) ArchiveFactsBefore ¶

func (s *Store) 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 (*Store) ArchiveSummary ¶

func (s *Store) 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 (*Store) Cancel ¶

func (s *Store) 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 (*Store) CancelOwned ¶

func (s *Store) 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 (*Store) CancelRequested ¶

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

CancelRequested reports whether a cooperative cancel request is pending.

func (*Store) CancelRunning ¶

func (s *Store) 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 (*Store) ClaimDue ¶

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

ClaimDue atomically claims one due task for owner.

func (*Store) Close ¶

func (s *Store) Close() error

Close releases the database connection.

func (*Store) Complete ¶

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

Complete marks a Running task Completed.

func (*Store) CountFacts ¶

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

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

func (*Store) CountTasks ¶

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

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

func (*Store) Enqueue ¶

func (s *Store) 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 (*Store) Facts ¶

func (s *Store) 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 (*Store) FactsForTask ¶

func (s *Store) 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 (*Store) Fail ¶

func (s *Store) 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 (*Store) FailPermanent ¶

func (s *Store) 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 (*Store) Get ¶

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

Get returns the current task record.

func (*Store) HeadSeq ¶

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

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

func (*Store) Heartbeat ¶

func (s *Store) 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 (*Store) LastFacts ¶

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

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

func (*Store) List ¶

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

func (*Store) ListWatermarks ¶

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

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

func (*Store) MarkOrphaned ¶

func (s *Store) 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 (*Store) ProjectCounts ¶

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

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

func (*Store) Requeue ¶

func (s *Store) 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 (*Store) RescueDead ¶

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

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

func (*Store) SaveWatermark ¶

func (s *Store) 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 (*Store) SetWatermark ¶

func (s *Store) 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 (*Store) StatusCounts ¶

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

StatusCounts counts tasks per status in one GROUP BY.

func (*Store) Watermark ¶

func (s *Store) 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").