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 json.RawMessage) 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) 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) 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) 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) 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 (*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) 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 since a time.
func (*Store) CountTasks ¶
CountTasks counts tasks matching the filter.
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 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 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) 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) 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.
Source Files