postgresv4

package module
v0.4.1 Latest Latest
Warning

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

 Go to latest Published: Oct 31, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Overview

Package postgresv4 is the ADR-0019 S1 spike (postgres half): a tq queue.Store implemented over go-cqrs-lite queue/postgres/v4 (the same upstream engine family as the sqlitev4 spike, postgres backend) plus tq-side companion surfaces in the SAME database. The companion surfaces themselves live ONCE in internal/queue/companion (dedup ruling 2026-09-26: one home, no cross-backend mirrors); this adapter keeps the engine-backed lifecycle, the pool ownership rules, and one-line delegations behind the Postgres dialect.

Division of labor (mirrors the sqlitev4 spike):

  • Engine-backed (upstream queue/postgres/v4, the S1 target): Enqueue, Complete, Fail, FailPermanent, Heartbeat, Cancel, CancelRunning, CancelRequested, CancelOwned, MarkOrphaned, RescueDead, DismissDead, UpdatePendingPriority.
  • Companion-owned (one home in internal/queue/companion): ClaimDue (upstream has no project exclusivity — the per-repo agent-pool guarantee, D24), Requeue (tq's evidence carries resume_closeout), all reads, RecordAnswer, AppendFact (non-task facts), watermarks, priority scores.

Known divergences are catalogued in the S1 report (D1 finalizes, D2 enqueued detail, D3 heartbeat facts, D4 fact archive, D5 payload storage class; D6 exclusivity/resume-closeout are adapter-implemented, not divergences); postgres-specific deltas land in the spike's own report.

Index

Constants

This section is empty.

Variables

View Source 
var WithProjectExclusivity = companion.WithProjectExclusivity

WithProjectExclusivity mirrors internal/queue/postgres's option: ClaimDue will not claim a task whose project already has another running task (the upstream engine has no such predicate — the adapter owns ClaimDue entirely, divergence noted in the S1 report).

Functions

This section is empty.

Types

type ArchiveStats 

type ArchiveStats = companion.ArchiveStats

ArchiveStats aliases the companion archive-summary shape (conform suite surface; the tq sqlite store's summary shape).

type Store 

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

Store is the S1 spike store: tq's queue.Store contract over the upstream postgres engine + companion surfaces in the same database.

func Open 

func Open(ctx context.Context, dsn string, opts ...StoreOption) (*Store, error)

Open connects to dsn (e.g. "postgres://user:pass@host:5432/db"), applies the upstream engine schema, and returns a ready store.

func OpenWithPool 

func OpenWithPool(ctx context.Context, pool *pgxpool.Pool, opts ...StoreOption) (*Store, error)

OpenWithPool wraps a caller-owned pgx pool into a ready store, applying the engine schema plus the tq companion tables. The caller keeps pool ownership: Close tears down the engine handle and the adapter's companion database/sql handle, never the caller's pool (the legacy internal/queue/postgres ownership contract, preserved at the S1 flip).

func (*Store) AppendFact 

func (s *Store) AppendFact(ctx context.Context, f journal.Fact) error

AppendFact records a NON-task journal fact (session.opened / session.closed). Task facts are never written through it.

S2 (ADR-0019): the append rides the engine's FactTx sink — the same-tx fact-append capability is engine-enforced, not the adapter's hand INSERT. The tq journal.Fact maps onto facts.Fact verbatim (open string FactType + raw detail bytes).

func (*Store) ArchiveFactsBefore 

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

ArchiveFactsBefore is NOT implemented in the S1 spike: the fact archive (facts_archive / journal_meta) has no upstream counterpart. The stub exists so the copied sqlite conformance suite compiles; its archive test skips itself with the divergence note.

func (*Store) ArchiveSummary 

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

ArchiveSummary is NOT implemented in the S1 spike: see ArchiveFactsBefore.

func (*Store) Cancel 

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

Cancel withdraws a Pending task (engine; error vocabulary mapped).

func (*Store) CancelOwned 

func (s *Store) CancelOwned(ctx context.Context, id task.ID, claim queue.Claim) error

CancelOwned finalizes a cooperative cancel. tq's gate is status+owner (no live-lease requirement — a worker may legitimately finish the stop just after the lease lapsed but before a reclaim), so requireLive=false.

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.

func (*Store) ClaimCounts added in v0.4.0 

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

ClaimCounts counts task.claimed facts per task id in one GROUP BY.

func (*Store) ClaimDue 

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

ClaimDue atomically claims at most one due task for owner, with tq's project-exclusivity predicate (companion-owned; S1 divergence).

func (*Store) ClaimDueTypes added in v0.4.0 

func (s *Store) ClaimDueTypes(
	ctx context.Context,
	owner string,
	lease time.Duration,
	types []string,
) (task.Task, queue.Claim, error)

ClaimDueTypes claims with an executor-type filter (the bare-worker refusal): a sh-only pool never claims an agent task — it stays PENDING for a pool that can run it. nil/empty types claims any type.

func (*Store) Close 

func (s *Store) Close() error

Close releases the engine and the companion handle (never a caller-owned pool — see OpenWithPool).

art-dupl:accept two-handle teardown: this backend owns its engine and companion db handles; companion cannot close them.

func (*Store) Complete 

func (s *Store) Complete(ctx context.Context, id task.ID, claim queue.Claim, result jsontext.Value) error

Complete marks a Running task Completed (owner gate, engine finalize). tq parity (divergence D1): tq's postgres clears last_error in the same UPDATE that completes the task; the upstream engine leaves a failed attempt's error on the completed row, so the adapter clears it after the finalize. A completed task can never Fail again, so the follow-up UPDATE cannot race a new error onto the row.

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) DeletePriorityScores 

func (s *Store) DeletePriorityScores(ctx context.Context, itemKeys []string) (int64, error)

DeletePriorityScores removes the cached verdicts for the given item keys.

func (*Store) DismissDead 

func (s *Store) DismissDead(ctx context.Context, id task.ID, reason, by string) error

DismissDead cancels a Dead task with a recorded reason.

func (*Store) Enqueue 

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

Enqueue persists a new task via the upstream engine; the shared routing surface (validation, deps/notBefore mapping, wake, tq-field re-read) lives ONCE in companion (dedup ruling 2026-09-26).

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. S2 (ADR-0019): the ENGINE owns the journal read (queue.Store contract) — no mirrored companion SQL.

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. S2: engine-owned read (queue.Store contract).

func (*Store) FactsSince 

func (s *Store) FactsSince(
	ctx context.Context,
	ftype journal.FactType,
	since time.Time,
	limit int,
) ([]journal.Fact, error)

FactsSince returns facts of one type recorded at or after since.

func (*Store) Fail 

func (s *Store) Fail(
	ctx context.Context,
	id task.ID,
	claim queue.Claim,
	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,
	claim queue.Claim,
	errText string,
	evidence jsontext.Value,
) error

FailPermanent dead-letters immediately (permanent error class).

func (*Store) Get 

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

Get returns the current task record.

art-dupl:accept Store-interface delegator: the method must exist on each backend; its body is already the companion call.

func (*Store) HeadSeq 

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

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

func (*Store) Heartbeat 

func (s *Store) Heartbeat(ctx context.Context, id task.ID, claim queue.Claim, 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)

List returns tasks matching the filter.

func (*Store) ListWatermarks 

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

ListWatermarks returns every consumer cursor, by consumer name.

func (*Store) MarkOrphaned 

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

MarkOrphaned appends task.orphaned facts for expired-lease Running tasks.

func (*Store) Notify added in v0.3.3 

func (s *Store) Notify() <-chan struct{}

Notify implements queue.Waker (M7): the claim-wake channel. See the Waker contract in internal/queue — buffered 1, non-blocking, fires after commits that may have landed a task in PENDING.

func (*Store) Park added in v0.4.0 

func (s *Store) Park(ctx context.Context, id task.ID, reason string, until time.Time, by string) error

Park suppresses claims on a PENDING task until the given instant (tq park): NotBefore moves and the task.parked fact lands in the same transaction (companion-owned). A past until is the release.

func (*Store) PriorityScore 

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) PriorityScores 

func (s *Store) PriorityScores(ctx context.Context) ([]queue.PriorityScore, error)

PriorityScores returns every cached verdict, ordered by item key.

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) RecordAnswer 

func (s *Store) RecordAnswer(ctx context.Context, id task.ID, ans queue.AnswerRecord) error

RecordAnswer records an owner's decision for a parked task's question (companion-owned; same-tx payload injection + NotBefore clear + fact).

func (*Store) Requeue 

func (s *Store) Requeue(
	ctx context.Context,
	id task.ID,
	claim queue.Claim,
	errText string,
	delay time.Duration,
	resumeCloseout bool,
	class string,
	provider string,
) error

Requeue returns a claimed task to Pending without counting an attempt (companion-owned; tq's evidence carries the resume_closeout flag and the requeue class).

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.

func (*Store) SavePriorityScore 

func (s *Store) SavePriorityScore(ctx context.Context, score queue.PriorityScore) error

SavePriorityScore upserts one cached item score (ADR-0015 score cache).

func (*Store) SaveWatermark 

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

SaveWatermark checkpoints a consumer cursor as a monotonic upsert.

func (*Store) SetWatermark 

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

SetWatermark overwrites a consumer cursor unconditionally (ops rescue).

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) UpdatePendingPriority 

func (s *Store) UpdatePendingPriority(ctx context.Context, id task.ID, newPriority int, source, reason string) error

UpdatePendingPriority changes a PENDING task's priority (in-tx fact).

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.

type StoreOption 

type StoreOption = companion.StoreOption

StoreOption configures optional Store behavior.

art-dupl:accept public-surface alias: the option type must be restated in each backend's package so callers import the backend, not companion.

Source Files

View all Source files

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.