Documentation
¶
Index ¶
- type Store
- func (s *Store) Cancel(ctx context.Context, id task.ID, reason string) error
- func (s *Store) CancelOwned(ctx context.Context, id task.ID, owner string) error
- func (s *Store) CancelRequested(ctx context.Context, id task.ID) (bool, error)
- func (s *Store) CancelRunning(ctx context.Context, id task.ID, reason string) error
- func (s *Store) ClaimDue(ctx context.Context, owner string, lease time.Duration) (task.Task, error)
- func (s *Store) Close() error
- func (s *Store) Complete(ctx context.Context, id task.ID, owner string, result jsontext.Value) error
- func (s *Store) CountFacts(ctx context.Context, ftype journal.FactType, since time.Time) (int64, error)
- func (s *Store) CountTasks(ctx context.Context, f queue.Filter) (int, error)
- func (s *Store) DismissDead(ctx context.Context, id task.ID, reason, by string) error
- func (s *Store) Enqueue(ctx context.Context, n task.New) (task.Task, error)
- func (s *Store) Facts(ctx context.Context, after int64, limit int) ([]journal.Fact, error)
- func (s *Store) FactsForTask(ctx context.Context, id string, limit int) ([]journal.Fact, error)
- func (s *Store) Fail(ctx context.Context, id task.ID, owner string, errText string, ...) error
- func (s *Store) FailPermanent(ctx context.Context, id task.ID, owner string, errText string, ...) error
- func (s *Store) Get(ctx context.Context, id task.ID) (task.Task, error)
- func (s *Store) HeadSeq(ctx context.Context) (int64, error)
- func (s *Store) Heartbeat(ctx context.Context, id task.ID, owner string, extend time.Duration) error
- func (s *Store) LastFacts(ctx context.Context, limit int) ([]journal.Fact, error)
- func (s *Store) List(ctx context.Context, f queue.Filter) ([]task.Task, error)
- func (s *Store) ListWatermarks(ctx context.Context) ([]queue.WatermarkEntry, error)
- func (s *Store) MarkOrphaned(ctx context.Context, cutoff time.Time) (int, error)
- func (s *Store) PriorityScore(ctx context.Context, itemKey string) (queue.PriorityScore, bool, error)
- func (s *Store) ProjectCounts(ctx context.Context) (map[string]map[task.Status]int, error)
- func (s *Store) Requeue(ctx context.Context, id task.ID, owner string, errText string, ...) error
- func (s *Store) RescueDead(ctx context.Context, id task.ID, maxAttempts int) error
- func (s *Store) SavePriorityScore(ctx context.Context, score queue.PriorityScore) error
- func (s *Store) SaveWatermark(ctx context.Context, consumer string, seq int64) error
- func (s *Store) SetWatermark(ctx context.Context, consumer string, seq int64) error
- func (s *Store) StatusCounts(ctx context.Context) (map[task.Status]int, error)
- func (s *Store) UpdatePendingPriority(ctx context.Context, id task.ID, newPriority int, source, reason string) error
- func (s *Store) Watermark(ctx context.Context, consumer string) (int64, bool, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the networked Store twin of sqlite.Store (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 Open ¶
Open 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 OpenWithPool ¶ added in v0.3.0
OpenWithPool wraps a caller-owned pool into a ready store, applying the schema on it. The caller keeps pool ownership: Store.Close does NOT close a caller-owned pool — shut the pool down yourself once the whole application is done with it. Consumers with an existing pool pass THEIR pool in instead of opening a second one via Open.
func (*Store) Cancel ¶
Cancel withdraws a Pending task. A non-empty reason is stored in the task.cancelled fact detail ("reason" key).
func (*Store) CancelOwned ¶
CancelOwned finalizes a cooperative cancel (lease holder). The operator's reason (from the cancel-requested fact) is carried onto the cancelled fact.
func (*Store) CancelRequested ¶
CancelRequested reports a pending cooperative cancel request.
func (*Store) CancelRunning ¶
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 (*Store) ClaimDue ¶
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 sqlite.Store: deps gate, expired-lease reclaim with task.released, pending cooperative cancels finalized at reclaim.
func (*Store) Close ¶
Close releases the store's resources. A pool the store opened via Open is closed; a caller-owned pool passed to OpenWithPool is left running — its lifecycle belongs to the caller.
func (*Store) Complete ¶
func (s *Store) Complete(ctx context.Context, id task.ID, owner string, result jsontext.Value) 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 since a time.
func (*Store) CountTasks ¶
CountTasks counts tasks matching the filter.
func (*Store) DismissDead ¶ added in v0.3.0
DismissDead cancels a Dead task with a recorded reason (DLQ dismiss): the autopsy verdict "unfixable" or an operator's ruling. The task.cancelled fact's detail carries the reason and by ("dlqfix-sweeper" or "operator"), so the journal keeps the death evidence AND the why of the withdrawal.
func (*Store) Enqueue ¶
Enqueue persists a new task and records task.enqueued; dedup keys make it idempotent (the partial unique index is the arbiter).
func (*Store) FactsForTask ¶
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 (*Store) Fail ¶
func (s *Store) Fail( ctx context.Context, id task.ID, owner string, errText string, backoff time.Duration, evidence jsontext.Value, ) 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 jsontext.Value, ) error
FailPermanent dead-letters regardless of the attempt budget.
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) ListWatermarks ¶
ListWatermarks returns every consumer cursor, by consumer name — the admin read behind `tq watermarks show`.
func (*Store) MarkOrphaned ¶
MarkOrphaned records stranded expired-lease Running tasks (idempotent).
func (*Store) PriorityScore ¶ added in v0.3.0
func (s *Store) PriorityScore(ctx context.Context, itemKey string) (queue.PriorityScore, bool, error)
PriorityScore returns the cached verdict for an item key, if any.
func (*Store) ProjectCounts ¶
ProjectCounts counts tasks per project per status.
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.
func (*Store) RescueDead ¶
RescueDead re-queues a Dead task with a fresh attempt budget.
func (*Store) SavePriorityScore ¶ added in v0.3.0
SavePriorityScore upserts one cached item score (ADR-0015 score cache). Mirror of the sqlite store's method (ADR-0007 conformance twin).
func (*Store) SaveWatermark ¶
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 (*Store) SetWatermark ¶
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 (*Store) StatusCounts ¶
StatusCounts counts tasks per status.
func (*Store) UpdatePendingPriority ¶ added in v0.3.0
func (s *Store) UpdatePendingPriority(ctx context.Context, id task.ID, newPriority int, source, reason string) error
UpdatePendingPriority changes a PENDING task's priority (ADR-0015 §5). The task.reprioritized fact — old/new priority, source, reason — is appended IN THE SAME transaction; a same-value update appends nothing. Mirror of the sqlite store's method (ADR-0007 conformance twin).
Source Files