azyncpgx

func (*Store) Ack ¶

func (s *Store) Ack(ctx context.Context, id, leaseToken uuid.UUID) error

Ack completes an active job, retaining it as StateSucceeded. Clearing the lease and the partial idempotency index excluding 'succeeded' frees the live-job idempotency key exactly as a delete would.

func (*Store) AckTaskResult ¶

func (s *Store) AckTaskResult(ctx context.Context, id, leaseToken uuid.UUID, result json.RawMessage) error

AckTaskResult completes an active task exactly like Ack and additionally persists result as the task's durable output, atomically. Same lease-token fencing: a stale token that no longer owns an active row is a not-found error.

func (*Store) AcquireLeadership ¶

func (s *Store) AcquireLeadership(ctx context.Context, name string) (func(), bool, error)

AcquireLeadership tries to take the named leadership via a PostgreSQL session-scoped advisory lock held on a dedicated connection retained for the lease. release is idempotent: it unlocks the advisory lock and closes the connection. acquired=false means another instance leads.

AcquireLeadership cannot detect a lost leadership (the session dying frees the lock server-side with no signal to the caller); prefer AcquireLeadershipLease, which exposes a Valid check for exactly that.

func (*Store) AcquireLeadershipLease ¶ added in v0.0.4

func (s *Store) AcquireLeadershipLease(ctx context.Context, name string) (driver.LeadershipLease, bool, error)

AcquireLeadershipLease tries to take the named leadership via a PostgreSQL session-scoped advisory lock held on a dedicated connection retained for the lease. acquired=false means another instance leads.

func (*Store) AllDaily ¶

func (s *Store) AllDaily(ctx context.Context, source driver.Source) ([]driver.DailyCount, error)

AllDaily returns the daily throughput window summed across every kind of the source, oldest day first.

func (*Store) AppendHistory ¶

func (s *Store) AppendHistory(ctx context.Context, workflowID uuid.UUID, typ string, payload json.RawMessage) (int64, error)

AppendHistory appends one durable history record with the next monotonic sequence number for the workflow. For an OperationCompleted/OperationFailed payload carrying an execution_key already recorded, it is idempotent: it returns the existing record's seq instead of appending a duplicate (see insertHistorySQL).

func (*Store) ApplyFailurePolicy ¶

func (s *Store) ApplyFailurePolicy(ctx context.Context) ([]driver.DAGFailure, error)

ApplyFailurePolicy applies each running workflow's OnFailure policy when it has at least one triggering dead task, in one transaction. This runs after PromoteUnblocked/CompleteDueSleeps and before CompleteDAGs on the worker's tick: a workflow whose dead task triggers is moved out of 'running' here, so only tolerated deaths remain for CompleteDAGs to settle.

Concurrency: triggeringDeadTasksSQL reads without a lock, so two concurrent ticks (or a tick racing a manual CompensateDAG/CancelDAG) can both observe the same workflow as 'running' under READ COMMITTED. The cancel/compensate branch below re-acquires the workflow row with lockDAGForUpdate and re-checks its state before touching any task rows: the loser of the race sees a state that already left 'running' and skips cleanly instead of racing insertCompensations's guard, which would otherwise violate the (dag_id, task_key) unique index.

func (*Store) ArchiveJob ¶

func (s *Store) ArchiveJob(ctx context.Context, source driver.Source, id uuid.UUID) error

ArchiveJob force-fails a pending or scheduled job of the source to dead.

func (*Store) CancelDAG ¶

func (s *Store) CancelDAG(ctx context.Context, id uuid.UUID) error

CancelDAG cancels a non-terminal workflow without compensating: its non-terminal tasks are cancelled and the workflow becomes cancelled, except a compensating workflow, which keeps compensating until CompleteDAGs lands it on cancelled.

Concurrency: the initial state read takes the row lock (FOR UPDATE), so a cancel racing a scheduler pass (or another verb) that is settling the same workflow blocks until the winner commits and then observes the committed state — a workflow that just reached a terminal state is a not-found here, never flipped to cancelled after the fact.

func (*Store) CancelWorkflowExecution ¶

func (s *Store) CancelWorkflowExecution(ctx context.Context, id uuid.UUID) error

CancelWorkflowExecution settles a non-terminal workflow as cancelled.

func (*Store) Changes ¶ added in v0.0.8

func (s *Store) Changes(ctx context.Context) (<-chan driver.Change, error)

Changes implements the optional driver.ChangeNotifier capability over a second, lazily-started LISTEN connection. It is separate from the wakeup listener on purpose: every worker subscribes to wakeups, and one shared connection would push the full change firehose at all of them; this way only processes that actually watch pay for the stream. A poll-only store returns the nil-channel contract signal.

func (*Store) Close ¶

func (s *Store) Close(_ context.Context) error

Close stops both listeners and closes the pool when the driver owns it. A pool passed to New is left for the caller to close.

func (*Store) CompensateDAG ¶

func (s *Store) CompensateDAG(ctx context.Context, id uuid.UUID) error

CompensateDAG manually triggers compensation on a running or suspended workflow, exactly like the OnFailureCancel policy.

Concurrency: the initial state read takes the row lock (FOR UPDATE) so it serializes against a concurrent ApplyFailurePolicy tick or another CompensateDAG call on the same workflow id. The loser blocks until the winner commits, then observes the post-commit state: no longer running/suspended, so it returns the same not-found this call already returns for a workflow that never qualified, instead of racing insertCompensations's guard into a unique-violation.

func (*Store) CompleteDAGs ¶

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

CompleteDAGs settles dags whose work is finished, running the four disjoint transitions in one transaction (suspend-on-dead-compensation before the terminal compensation settle so the two never overlap).

func (*Store) CompleteDueSleeps ¶

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

CompleteDueSleeps marks every scheduled $sleep timer of a running workflow whose run_at is due as succeeded, without running any handler.

func (*Store) CompleteWorkflow ¶

func (s *Store) CompleteWorkflow(ctx context.Context, id uuid.UUID, result json.RawMessage) error

CompleteWorkflow settles a workflow as succeeded, persisting the result.

func (*Store) CreateDAG ¶

func (s *Store) CreateDAG(ctx context.Context, p driver.DAGParams) (bool, uuid.UUID, error)

CreateDAG atomically inserts the workflow header, its tasks and its dependency edges, and signals workers for immediately runnable tasks, in one transaction. It deduplicates by (Name, IdempotencyKey) against live executions, returning (false, existingID) without inserting anything when a live execution already holds the key.

func (*Store) CreateDAGTx ¶

func (s *Store) CreateDAGTx(ctx context.Context, tx pgx.Tx, p driver.DAGParams) (bool, uuid.UUID, error)

CreateDAGTx performs CreateDAG within the caller's transaction so the workflow commits atomically with the caller's own writes.

func (*Store) DAGDeps ¶ added in v0.0.7

func (s *Store) DAGDeps(ctx context.Context, id uuid.UUID) ([]driver.DAGDep, error)

DAGDeps returns every dependency edge of the workflow, compensation-chain links included. An unknown id yields no rows, not an error.

func (*Store) DAGNameStateCounts ¶ added in v0.0.7

func (s *Store) DAGNameStateCounts(ctx context.Context) (map[string]map[driver.DAGState]int64, error)

DAGNameStateCounts returns how many dags sit in each state, per definition.

func (*Store) DAGTaskCounts ¶ added in v0.0.7

func (s *Store) DAGTaskCounts(ctx context.Context, ids []uuid.UUID) (map[uuid.UUID]map[driver.JobState]int64, error)

DAGTaskCounts returns each requested dag's task breakdown by state. Ids with no tasks are absent from the result; an empty ids slice queries nothing.

func (*Store) DAGTasks ¶

func (s *Store) DAGTasks(ctx context.Context, id uuid.UUID) ([]driver.Job, error)

DAGTasks returns every task job of the workflow (compensation tasks included) ordered by created_at then id — tasks inserted in the same atomic batch share created_at, so their relative order is stable, not the declaration order. It returns a not-found error when the workflow does not exist.

func (*Store) Dead ¶

func (s *Store) Dead(ctx context.Context, id, leaseToken uuid.UUID, lastError string) error

Dead moves a failed active job to StateDead and records the final attempt.

func (*Store) DeleteAll ¶

func (s *Store) DeleteAll(ctx context.Context, source driver.Source, kind string, state driver.JobState) (int64, error)

DeleteAll deletes every job of (source, kind) in the given state, looping in maintenanceBatch-sized deletes until a batch removes fewer than that many, and returns the total removed. An empty kind targets all kinds of the source.

func (*Store) DeleteJob ¶

func (s *Store) DeleteJob(ctx context.Context, source driver.Source, id uuid.UUID, state driver.JobState) error

DeleteJob deletes a job of the source in the given state.

func (*Store) DeleteSubscriber ¶ added in v0.0.4

func (s *Store) DeleteSubscriber(ctx context.Context, name, eventType string) (int64, error)

DeleteSubscriber removes the (name, eventType) registration, or every registration of name when eventType is empty.

func (*Store) DeliverBufferedSignals ¶ added in v0.0.6

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

DeliverBufferedSignals hands buffered inbox signals to tasks that have become deliverable since the signal arrived, oldest first per task, and returns the count. Set-based and idempotent; called on the scheduler tick right after PromoteUnblocked.

func (*Store) DequeueBatch ¶

func (s *Store) DequeueBatch(ctx context.Context, source driver.Source, p driver.DequeueParams) ([]driver.Job, error)

DequeueBatch leases up to p.Limit due pending jobs of (source, p.Kind).

func (*Store) Enqueue ¶

func (s *Store) Enqueue(ctx context.Context, p driver.EnqueueParams) (bool, error)

Enqueue durably inserts one queue job in its own short transaction and signals workers with a best-effort notify after commit (see notifyAfterCommit): a slow or failing notify can never abort the enqueue.

func (*Store) EnqueueTx ¶

func (s *Store) EnqueueTx(ctx context.Context, tx pgx.Tx, p driver.EnqueueParams) (bool, error)

EnqueueTx performs Enqueue within the caller's transaction so the outbox commits atomically with the caller's own writes. The notify stays inside the caller's transaction here — that is the outbox contract: a wakeup is only ever sent if the caller's own transaction commits, and Postgres fires it exactly on that commit.

func (*Store) ExtendLease ¶

func (s *Store) ExtendLease(ctx context.Context, id, leaseToken uuid.UUID, lease time.Duration) error

ExtendLease renews an active job's lease. Fenced by lease token.

func (*Store) FailWorkflow ¶

func (s *Store) FailWorkflow(ctx context.Context, id uuid.UUID, reason string) error

FailWorkflow settles a workflow as failed, recording the reason.

func (*Store) FindDAGByKey ¶ added in v0.0.6

func (s *Store) FindDAGByKey(ctx context.Context, name, idempotencyKey string) (uuid.UUID, error)

FindDAGByKey resolves the live workflow holding (name, idempotencyKey) — the business-key lookup a webhook handler needs. Served by the partial unique dedupe index, so at most one row can match.

func (*Store) GetDAG ¶

func (s *Store) GetDAG(ctx context.Context, id uuid.UUID) (*driver.DAGView, error)

GetDAG returns one workflow header by id, or a not-found error.

func (*Store) GetEvent ¶

func (s *Store) GetEvent(ctx context.Context, id uuid.UUID) (*driver.EventAdminRow, error)

GetEvent returns a single ledger event by id, or a not-found error.

func (*Store) GetJob ¶

func (s *Store) GetJob(ctx context.Context, source driver.Source, id uuid.UUID) (*driver.Job, error)

GetJob returns a single job of the source by id, or a not-found error.

func (*Store) GetWorkflowExecution ¶

func (s *Store) GetWorkflowExecution(ctx context.Context, id uuid.UUID) (driver.WorkflowExecutionView, error)

GetWorkflowExecution returns one execution header by id, or a not-found error when it does not exist.

func (*Store) JobAttempts ¶

func (s *Store) JobAttempts(ctx context.Context, source driver.Source, id uuid.UUID) ([]driver.AttemptError, error)

JobAttempts returns a job's failure history, oldest attempt first.

func (*Store) KindDepths ¶

func (s *Store) KindDepths(ctx context.Context, source driver.Source) (map[string]driver.Depths, error)

KindDepths returns per-kind instantaneous state counters of the source, including the oldest pending job's age (see driver.Depths.OldestPendingAge).

func (*Store) ListDAGs ¶

func (s *Store) ListDAGs(ctx context.Context, filter driver.DAGFilter, offset, limit int) ([]driver.DAGView, int64, error)

ListDAGs lists dags matching filter, newest first (created_at then id descending), paginated, with the total matching count.

func (*Store) ListEvents ¶

func (s *Store) ListEvents(ctx context.Context, filter driver.EventFilter, offset, limit int) ([]driver.EventAdminRow, int64, error)

ListEvents lists ledger events matching filter, newest first, paginated.

func (*Store) ListHistory ¶

func (s *Store) ListHistory(ctx context.Context, workflowID uuid.UUID) ([]driver.HistoryEvent, error)

ListHistory returns the workflow's history events in sequence order.

func (*Store) ListJobs ¶

func (s *Store) ListJobs(ctx context.Context, source driver.Source, filter driver.JobFilter, offset, limit int) ([]driver.Job, int64, error)

ListJobs lists jobs of the source matching filter, paginated. Ordering depends on filter.State (see the driver.Store contract); a limit <= 0 means unbounded.

func (*Store) ListKinds ¶

func (s *Store) ListKinds(ctx context.Context, source driver.Source) ([]string, error)

ListKinds returns the distinct kinds of the source (from live jobs and stat history), sorted.

func (*Store) ListStalledWorkflows ¶ added in v0.0.4

func (s *Store) ListStalledWorkflows(ctx context.Context, olderThan time.Duration, limit int) ([]driver.StalledWorkflow, error)

ListStalledWorkflows returns up to limit running executions with no live task, updated at least olderThan ago.

func (*Store) ListSubscriberViews ¶

func (s *Store) ListSubscriberViews(ctx context.Context, eventType string) ([]driver.SubscriberView, error)

ListSubscriberViews returns subscriber registrations, ordered by event type then name. An empty eventType returns all.

func (*Store) MarkUncertain ¶

func (s *Store) MarkUncertain(ctx context.Context, operationJobID, leaseToken uuid.UUID, reason string) error

MarkUncertain moves an active Operation to uncertain and suspends the run.

func (*Store) Migrate ¶

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

Migrate brings the backend schema up to date via goose. Open never migrates. When a schema is configured it is created if absent and the migrations run there; the version-tracking table is the configured MigrationsTable (default azync_migrations).

func (*Store) NukeAll ¶

func (s *Store) NukeAll(ctx context.Context, source driver.Source) (driver.NukeReport, error)

NukeAll deletes all jobs, stats and idempotency keys of the source (a dev reset) and reports the counts. The event ledger is left intact; deleting the jobs cascades to their attempt history.

func (*Store) OpsStats ¶

func (s *Store) OpsStats(ctx context.Context) (driver.OpsStats, error)

OpsStats returns the event ledger admin summary.

func (*Store) PauseDAG ¶ added in v0.0.6

func (s *Store) PauseDAG(ctx context.Context, id uuid.UUID, reason string) error

PauseDAG freezes a running workflow: header to paused (reason recorded in failure_reason — the "why is this stopped" column for both suspensions and pauses), pending/scheduled tasks to paused, one transaction. Blocked and waiting tasks keep their states: with the header out of running/compensating, PromoteUnblocked and DeliverBufferedSignals skip the workflow, and incoming signals buffer for delivery after RetryDAG resumes it. The row lock serializes against the scheduler's policy/cancel passes.

func (*Store) PauseJob ¶

func (s *Store) PauseJob(ctx context.Context, source driver.Source, id uuid.UUID) error

PauseJob holds a pending or scheduled job of the source out of the ready set.

func (*Store) PromoteDue ¶

func (s *Store) PromoteDue(ctx context.Context, source driver.Source, kinds []string) (int64, error)

PromoteDue moves due scheduled jobs of the given kinds to pending, looping in maintenanceBatch-sized steps until a batch promotes fewer than that many (see WithMaintenanceBatch).

func (*Store) PromoteUnblocked ¶

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

PromoteUnblocked moves every blocked task whose dependencies are all satisfied to its runnable state, waking workers for the newly pending tasks.

func (*Store) Publish ¶

func (s *Store) Publish(ctx context.Context, p driver.PublishParams) (int, error)

Publish atomically appends one event and fans out one pending delivery per matching subscriber in a single transaction. The wakeup notify fires best-effort after commit (see notifyAfterCommit): wakeups are lossy by contract, and firing pg_notify outside the transaction keeps a slow or failing notify from ever aborting the publish.

func (*Store) PublishTx ¶

func (s *Store) PublishTx(ctx context.Context, tx pgx.Tx, p driver.PublishParams) (int, error)

PublishTx performs Publish within the caller's transaction. The notify stays inside the caller's transaction here — that is the outbox contract: a wakeup is only ever sent if the caller's own transaction commits.

func (*Store) ReapExpired ¶

func (s *Store) ReapExpired(ctx context.Context, source driver.Source, kinds []string, maxReaps int) (int64, int64, error)

ReapExpired reclaims active jobs of the given kinds whose lease expired, returning the number reaped and the subset killed. It loops in maintenanceBatch-sized transactions (see WithMaintenanceBatch) until a batch reclaims fewer than that many, so a mass lease expiry is reclaimed in bounded steps with short-lived locks rather than one long transaction holding every expired row.

func (*Store) RegisterSubscriber ¶

func (s *Store) RegisterSubscriber(ctx context.Context, sub driver.Subscriber) error

RegisterSubscriber upserts a subscriber keyed by (Name, EventType).

func (*Store) Release ¶

func (s *Store) Release(ctx context.Context, id, leaseToken uuid.UUID) error

Release returns a leased job to StatePending, decrementing the attempt it did not really spend, without recording an attempt. Fenced by lease token.

func (*Store) Replay ¶

func (s *Store) Replay(ctx context.Context, filter driver.ReplayFilter) (int64, error)

Replay re-fans-out ledger events matching filter into fresh pending deliveries flagged Replay, and returns the number created. It owns its own transaction (there is no ReplayTx / outbox variant), so its wakeups fire best-effort after commit like Enqueue and Publish.

func (*Store) Reschedule ¶

func (s *Store) Reschedule(ctx context.Context, id, leaseToken uuid.UUID, delay time.Duration, lastError string) error

Reschedule parks a failed active job as StateScheduled and records the attempt.

func (*Store) ResolveUncertain ¶

func (s *Store) ResolveUncertain(ctx context.Context, operationJobID uuid.UUID, decision string, result json.RawMessage) (uuid.UUID, string, error)

ResolveUncertain applies complete/fail/retry to an uncertain Operation. History append is the caller's responsibility (so payloads stay typed).

func (*Store) ResumeJob ¶

func (s *Store) ResumeJob(ctx context.Context, source driver.Source, id uuid.UUID) error

ResumeJob returns a paused job of the source to pending or scheduled per its run_at.

func (*Store) ResumeWorkflow ¶

func (s *Store) ResumeWorkflow(ctx context.Context, id uuid.UUID) error

ResumeWorkflow moves a suspended execution back to running.

func (*Store) Retain ¶

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

Retain deletes up to limit ledger events before the cutoff whose deliveries are all terminal, cascading to those deliveries, and returns the count.

func (*Store) RetryAllDead ¶

func (s *Store) RetryAllDead(ctx context.Context, source driver.Source, kind string) (int64, error)

RetryAllDead resets every dead job of (source, kind) to pending, looping in maintenanceBatch-sized updates until a batch touches fewer than that many. An empty kind targets all kinds of the source.

func (*Store) RetryDAG ¶

func (s *Store) RetryDAG(ctx context.Context, id uuid.UUID) error

RetryDAG resumes a non-terminal workflow after failures or an operator pause: dead tasks reset (or only dead compensation tasks once a chain exists), paused tasks return to the ready set, and a suspended or paused workflow resumes — to running, or back to compensating. Both paths clear the tasks' stamped snooze deadline, so a resumed wait starts with a fresh budget.

func (*Store) RetryJob ¶

func (s *Store) RetryJob(ctx context.Context, source driver.Source, id uuid.UUID) error

RetryJob resets a dead job of the source to pending for immediate retry.

func (*Store) RunNow ¶ added in v0.0.6

func (s *Store) RunNow(ctx context.Context, source driver.Source, id uuid.UUID) error

RunNow expedites a scheduled job to pending with run_at = now, then wakes workers post-commit — the early-wake verb for a snoozed poll.

func (*Store) ScheduleOperation ¶

func (s *Store) ScheduleOperation(ctx context.Context, p driver.ScheduleOperationParams) (uuid.UUID, error)

ScheduleOperation inserts one Operation task job, deduping by ExecutionKey.

func (*Store) ScheduleTask ¶

func (s *Store) ScheduleTask(ctx context.Context, workflowID uuid.UUID, kind string, runAt time.Time) error

ScheduleTask durably inserts one workflow-task job (Source SourceWorkflow, RunID = workflowID), born pending when runAt is due, scheduled otherwise, and signals workers for an immediately-runnable job.

func (*Store) Signal ¶

func (s *Store) Signal(ctx context.Context, p driver.DAGSignalParams) (int64, bool, error)

Signal delivers (or buffers) one named signal on a live workflow. One transaction: the header lock serializes against ApplyFailurePolicy / CancelDAG / CompensateDAG (which take the same FOR UPDATE), the inbox insert dedupes by MessageID, and the immediate-delivery attempt completes a waiting $signal (payload as result) or wakes a scheduled $sleep. A signal nothing was waiting for stays buffered, unconsumed, for DeliverBufferedSignals — never lost. No notify: $signal/$sleep have no handler; dependents are promoted (and notified) by the scheduler tick.

func (*Store) SignalWorkflow ¶

func (s *Store) SignalWorkflow(ctx context.Context, p driver.SignalParams) (bool, error)

SignalWorkflow atomically appends the delivery to the inbox (deduped by MessageID), the SignalReceived history record, and — unless the execution is already terminal — a wake workflow-task job, all in one transaction: a newly delivered signal is never left recorded with no live task able to act on it, closing the crash window between the separate calls Client.Signal used to make. The row lock (FOR UPDATE) also serializes the history sequence number against a concurrent AppendHistory (e.g. a workflow-task replay appending its own event at the same moment).

func (*Store) Skip ¶ added in v0.0.6

func (s *Store) Skip(ctx context.Context, id, leaseToken uuid.UUID, reason string) error

Skip settles an active job as StateSkipped with the reason retained. Fenced by lease token.

func (*Store) Snooze ¶

func (s *Store) Snooze(ctx context.Context, id, leaseToken uuid.UUID, delay time.Duration, deadlineError string) (bool, error)

Snooze parks an active job as StateScheduled with run_at now()+delay without consuming the retry budget and without recording an attempt — unless the job's stamped snooze deadline has passed, in which case it dead-letters the job atomically (deadlined=true) with deadlineError as its final attempt. Fenced by lease token.

func (*Store) StartWorkflow ¶

func (s *Store) StartWorkflow(ctx context.Context, p driver.WorkflowStartParams) (bool, uuid.UUID, error)

StartWorkflow atomically inserts one workflow-as-code execution header, deduplicating by (Name, BusinessIdempotencyKey) against live (running or suspended) executions, and — for a newly inserted execution — records the WorkflowStarted history event and schedules the first workflow-task job in the same transaction: a caller can never observe a newly started execution with no history or task, closing the crash window between the three separate calls Client.Start used to make.

func (*Store) Stats ¶

func (s *Store) Stats(ctx context.Context, source driver.Source, kind string) (driver.Depths, []driver.DailyCount, error)

Stats returns one kind's instantaneous depths and its daily throughput window, oldest day first.

func (*Store) Subscribers ¶

func (s *Store) Subscribers(ctx context.Context, eventType string) ([]driver.Subscriber, error)

Subscribers returns the registrations for an event type, ordered by name.

func (*Store) SuspendWorkflow ¶

func (s *Store) SuspendWorkflow(ctx context.Context, id uuid.UUID, reason string) error

SuspendWorkflow parks a running workflow for a manual decision.

func (*Store) TaskResults ¶

func (s *Store) TaskResults(ctx context.Context, dagID uuid.UUID, keys []string) (map[string]driver.TaskResult, error)

TaskResults returns the settled outcomes of the workflow's succeeded and skipped tasks, keyed by task key, restricted to keys when non-empty. A succeeded task without a result maps to an entry with a nil Result; a skipped one to Skipped=true — distinguishable, never a silent zero value.

func (*Store) VacuumCompleted ¶

func (s *Store) VacuumCompleted(ctx context.Context, source driver.Source, retention time.Duration) (int64, error)

VacuumCompleted trims succeeded jobs of the source completed before retention ago, looping in maintenanceBatch-sized deletes until a batch removes fewer than that many. A retention <= 0 removes nothing.

func (*Store) VacuumDAGs ¶

func (s *Store) VacuumDAGs(ctx context.Context, retention time.Duration) (int64, error)

VacuumDAGs deletes terminal dags completed before retention ago, cascading (via the FKs) to their task jobs and dependency edges. A retention <= 0 removes nothing.

func (*Store) VacuumDead ¶

func (s *Store) VacuumDead(ctx context.Context, source driver.Source, kind string, olderThan time.Duration) (int64, error)

VacuumDead deletes dead jobs of (source, kind) enqueued before olderThan ago, looping in maintenanceBatch-sized deletes until a batch removes fewer than that many, and returns the total removed. An empty kind targets all kinds of the source.

func (*Store) VacuumIdempotency ¶

func (s *Store) VacuumIdempotency(ctx context.Context, source driver.Source) (int64, error)

VacuumIdempotency trims expired time-window dedupe keys of the source.

func (*Store) VacuumStats ¶

func (s *Store) VacuumStats(ctx context.Context, source driver.Source, retention time.Duration) (int64, error)

VacuumStats trims daily stat counters of the source older than retention. A retention <= 0 removes nothing.

func (*Store) VacuumWorkflows ¶

func (s *Store) VacuumWorkflows(ctx context.Context, retention time.Duration) (int64, error)

VacuumWorkflows deletes terminal workflow-as-code executions completed before retention ago, cascading (via FKs) to history, signals, timers and jobs linked by run_id. A retention <= 0 removes nothing.

func (*Store) Wake ¶

func (s *Store) Wake(ctx context.Context) (<-chan driver.Wake, error)

Wake returns a channel of wakeups signaled by enqueues and publishes. It may be called several times; each caller gets its own channel, closed when its ctx ends. A poll-only listener returns a nil channel and nil error.