companion

package module
v0.4.1 Latest Latest
Warning

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

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

Documentation

Overview

Package companion is the single home for the tq companion surfaces the ADR-0019 spike adapters (sqlitev4, postgresv4) currently mirror: the facts/watermark/priority-score/status/project reads, listWhere/List, ClaimDue with project exclusivity, Requeue, and the RecordAnswer/question cluster, all written against a placeholder- dialable Runner so one implementation serves every backend dialect.

STATUS: extracted by the 2026-09-26 companion window (owner ruling "accepts ≈ 0"): the sqlitev4 and postgresv4 adapters delegate their companion surfaces here — the design in docs/planning/2026-09-26_companion-extraction-design.md is LANDED for production code (art-dupl: zero cross-backend adapter groups). The three mirrored conformance suites are the last mirrors; their consolidation into companion/conform is tracked in TODO_LIST and ratcheted by scripts/check-mirror-clones.sh (strict: baseline scripts/mirror-baseline.txt fails on any new cross-backend group).

Index

Constants

View Source 
const CompanionIndexes = `` /* 199-byte string literal not displayed */

CompanionIndexes extends the engine-owned tasks table with the claim path's hot probes (an index is an extension, not a redefinition). Both are PARTIAL over the running set only — a handful of rows — so they stay tiny no matter how deep the terminal history grows:

  • idx_tasks_project_running serves ClaimDue's project-exclusivity NOT EXISTS: with the engine's full idx_tasks_project the probe walks every historical row of the candidate's project (thousands after a harvest-heavy life); the partial index answers from the running set alone.
  • idx_tasks_lease_running serves the expired-lease reclaim arm (status='running' AND lease_expires <= ?) as a direct range probe instead of a scan of all running rows filtered row-by-row.

SQLite and PostgreSQL accept identical partial-index DDL, so the one const serves both dialects. The indexed columns predate the index (they are engine-schema columns), keeping the "indexes only after the column exists" migration rule satisfied by construction.

View Source 
const CompanionSchema = `` /* 460-byte string literal not displayed */

CompanionSchema is the tq-side table set the upstream engines do not carry (the tasks/facts/deps/watermarks tables come from the engines' own migrate — same-DB companion surfaces read and extend them but never redefine them). BIGINT has INTEGER affinity in SQLite, so the one DDL serves both engine families.

Variables

This section is empty.

Functions

func AppendFactTx 

func AppendFactTx(ctx context.Context, d Dialect, db Beginner, f journal.Fact) error

AppendFactTx records a NON-task journal fact in its own transaction (session.opened / session.closed). Task facts are never written through it — every task mutation appends its fact inside its own operation's transaction.

func ApplyProjectExclusivity 

func ApplyProjectExclusivity(opts []StoreOption) bool

ApplyProjectExclusivity resolves the option list into the boolean the ClaimDue predicate consumes.

func ClaimCounts added in v0.4.1 

func ClaimCounts(ctx context.Context, r Runner) (map[string]int, error)

ClaimCounts counts task.claimed facts per task id in one GROUP BY — the census pushdown behind loop-suspect detection: one grouped read over the journal instead of a per-task facts walk.

art-dupl:accept idiomatic SQL read: the rows/err/defer-Close prolog and; ruling: docs/status/2026-10-05_16-02_art-dupl-t3-tail-zero-dedup-sweep.md a)3 the scan-loop tail are the database/sql idiom; the defer must live in the consuming frame, so no seam can shorten this.

func ClaimDue 

func ClaimDue(
	ctx context.Context,
	d Dialect,
	db Beginner,
	projectExclusive bool,
	owner string,
	lease time.Duration,
	types []string,
) (task.Task, queue.Claim, error)

ClaimDue atomically claims at most one due task for owner, with tq's project-exclusivity predicate (the upstream engine's candidate query has no such clause — S1 divergence). The claim mints an upstream-format fencing token and stamps it on the row, so engine-backed finalizes keep working for the claimed task. A non-empty types list restricts candidates to those executor types (the bare-worker refusal: a sh-only pool never claims an agent task, so it stays PENDING for a pool that can run it — no claim, no task.claimed fact, no loop-suspect noise).

func CountFacts 

func CountFacts(ctx context.Context, r Runner, ftype journal.FactType, since time.Time) (int64, error)

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

func CountTasks 

func CountTasks(ctx context.Context, r Runner, f queue.Filter) (int, error)

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

func DeletePriorityScores 

func DeletePriorityScores(ctx context.Context, r Runner, itemKeys []string) (int64, error)

DeletePriorityScores removes the cached verdicts for the given item keys.

func Enqueue added in v0.4.1 

func Enqueue(ctx context.Context, engine EngineEnqueuer, cr Runner, wake func(), n task.New) (task.Task, error)

Enqueue routes a tq task.New through the upstream engine and re-reads the created task so the returned record carries every tq field (DedupKey included — the upstream task struct does not surface it). The engine owns the write; this is the ONE shared routing surface for both backend adapters (dedup ruling 2026-09-26: one home, no cross-backend mirrors). wake is the adapter's fireWake (may be nil).

func FactsSince 

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

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

func Get 

func Get(ctx context.Context, r Runner, id task.ID) (task.Task, error)

Get returns the current task record.

func HeadSeq 

func HeadSeq(ctx context.Context, r Runner) (int64, error)

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

func IdentityCodec 

func IdentityCodec() uqueue.Codec[[]byte]

IdentityCodec passes payloads through byte-for-byte: tq payloads are jsontext.Value, already JSON; the default JSONCodec would base64-encode the bytes. A nil value binds as the EMPTY blob, never SQL NULL — the upstream tasks.payload column is NOT NULL and tq's zero-value payloads are empty, not null (tq's own schema stores the empty string the same way).

func JournalFacts 

func JournalFacts(fs []ufacts.Fact) []journal.Fact

JournalFacts maps engine-read facts onto the tq journal vocabulary (ADR-0019 S2): the ENGINE owns the journal reads — Facts/FactsForTask come from the upstream store contract, never mirrored companion SQL. The alias flip made the vocabularies one type, so the copy is the identity — except Detail: the engine hands back empty non-nil blobs for detail-less facts, and an empty jsontext.Value fails payload marshaling with "unexpected EOF", so empty normalizes to nil. This empty-to-nil rule is the genuinely-needed residue of the pre-alias conversion (regression: TestProjectionHostTailsLiveFacts wedged on task.claimed when the identity briefly shipped).

func LastFacts 

func LastFacts(ctx context.Context, r Runner, limit int) ([]journal.Fact, error)

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

func List 

func List(ctx context.Context, r Runner, f queue.Filter) ([]task.Task, error)

List returns tasks matching the filter.

func ListWatermarks 

func ListWatermarks(ctx context.Context, r Runner) ([]queue.WatermarkEntry, error)

ListWatermarks returns every consumer cursor, by consumer name.

art-dupl:accept idiomatic SQL read: the rows/err/defer-Close prolog and; ruling: docs/status/2026-10-05_16-02_art-dupl-t3-tail-zero-dedup-sweep.md a)3 the scan-loop tail are the database/sql idiom; the defer must live in the consuming frame, so no seam can shorten this.

func MapErr 

func MapErr(err error) error

MapErr translates upstream engine errors onto tq's sentinels so callers program against one error vocabulary.

func Migrate 

func Migrate(ctx context.Context, r Runner) error

Migrate creates the companion tables on the shared database handle.

func MustJSON 

func MustJSON(v any) jsontext.Value

func Park added in v0.4.1 

func Park(
	ctx context.Context,
	d Dialect,
	db Beginner,
	id task.ID,
	reason string,
	until time.Time,
	by string,
) error

Park suppresses claims on a PENDING task until the given instant (tq park): the operator's reversible disposition, recorded as a fact. Until moves NotBefore (clamped to now — a past until is the release); the task.parked fact lands in the SAME transaction, so "parked but still claimable" is unrepresentable and the journal keeps every ruling. Running/terminal tasks refuse with task.ErrInvalidTransition, unknown ids with task.ErrNotFound.

func PriorityScore 

func PriorityScore(ctx context.Context, r Runner, itemKey string) (queue.PriorityScore, bool, error)

PriorityScore returns the cached verdict for an item key, if any.

func PriorityScores 

func PriorityScores(ctx context.Context, r Runner) ([]queue.PriorityScore, error)

PriorityScores returns every cached verdict, ordered by item key.

art-dupl:accept idiomatic SQL read: the rows/err/defer-Close prolog and; ruling: docs/status/2026-10-05_16-02_art-dupl-t3-tail-zero-dedup-sweep.md a)3 the scan-loop tail are the database/sql idiom; the defer must live in the consuming frame, so no seam can shorten this.

func ProjectCounts 

func ProjectCounts(ctx context.Context, r Runner) (map[string]map[task.Status]int, error)

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

art-dupl:accept idiomatic SQL read: the rows/err/defer-Close prolog and; ruling: docs/status/2026-10-05_16-02_art-dupl-t3-tail-zero-dedup-sweep.md a)3 the scan-loop tail are the database/sql idiom; the defer must live in the consuming frame, so no seam can shorten this.

func RecordAnswer 

func RecordAnswer(ctx context.Context, d Dialect, db Beginner, id task.ID, ans queue.AnswerRecord) error

RecordAnswer records an owner's decision for a parked task's question. Same-transaction payload injection + NotBefore clear + fact, idempotent per question ref — tq semantics over the shared tasks/facts tables.

func Requeue 

func Requeue(
	ctx context.Context,
	d Dialect,
	db Beginner,
	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. Adapter-side (not the engine's): tq's task.requeued evidence carries the resume_closeout flag and the requeue class, which the upstream RequeueEvidence lacks (S1 divergence). The claim token fences the requeue; the recorded fact's owner is the claim's OWNER (read from the row before the release), so the journal keeps speaking owner vocabulary.

func SavePriorityScore 

func SavePriorityScore(ctx context.Context, r Runner, score queue.PriorityScore) error

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

func SaveWatermark 

func SaveWatermark(ctx context.Context, r Runner, consumer string, seq int64) error

SaveWatermark checkpoints a consumer cursor as a monotonic upsert.

func SetWatermark 

func SetWatermark(ctx context.Context, r Runner, consumer string, seq int64) error

SetWatermark overwrites a consumer cursor unconditionally (ops rescue).

func StatusCounts 

func StatusCounts(ctx context.Context, r Runner) (map[task.Status]int, error)

StatusCounts counts tasks per status in one GROUP BY.

art-dupl:accept idiomatic SQL read: the rows/err/defer-Close prolog and; ruling: docs/status/2026-10-05_16-02_art-dupl-t3-tail-zero-dedup-sweep.md a)3 the scan-loop tail are the database/sql idiom; the defer must live in the consuming frame, so no seam can shorten this.

func TokenFor 

func TokenFor(ctx context.Context, r Runner, id task.ID, claim queue.Claim, requireLive bool) (string, error)

TokenFor enforces tq's token gate for a finalize: the caller must present the claim token minted at ClaimDue (theft detection lives in the store — ADR-0019 S1). With requireLive the lease must also be unexpired; CancelOwned deliberately skips liveness (a worker may finish a stop just after lease lapse, before reclaim).

func UpstreamFact 

func UpstreamFact(f journal.Fact) ufacts.Fact

UpstreamFact maps a tq journal fact onto the upstream facts.Fact. Type strings and detail bytes carry over verbatim; Seq is zeroed because the journal reassigns it on append. Since the S2 alias flip (ADR-0019) the two fact types ARE one type — the mapper is the identity modulo Seq, kept as the named seam for the engine adapters.

func Watermark 

func Watermark(ctx context.Context, r Runner, consumer string) (int64, bool, error)

Watermark returns the persisted read cursor for a journal consumer.

func WithToken added in v0.3.3 

func WithToken(
	ctx context.Context,
	r Runner,
	id task.ID,
	claim queue.Claim,
	requireLive bool,
	use func(token string) error,
) error

WithToken runs use with the claim token TokenFor validates: the token gate and the engine finalize it guards collapse onto one seam, so a backend adapter cannot order the finalize ahead of the gate. The TokenFor error is returned as-is (already tq vocabulary); everything use returns passes through untouched.

func WithTx 

func WithTx(ctx context.Context, db Beginner, d Dialect, fn func(Runner) error) error

WithTx runs fn inside one transaction, handing it the DIALED tx. A non-nil error rolls the tx back; otherwise it commits.

Types

type ArchiveStats 

type ArchiveStats struct {
	Hot       int64
	Archived  int64
	Watermark int64
}

ArchiveStats mirrors the tq sqlite store's archive summary shape — shared by every companion backend's ArchiveSummary surface.

type Beginner 

type Beginner interface {
	BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error)
}

Beginner is the transaction-opening surface (*sql.DB). Tx-bodied companion functions take it so the dialect applies inside the tx too.

type Dialect 

type Dialect func(query string) string

Dialect rewrites the sqlite-style "?" placeholder SQL every companion body is written in into the backend's native form. SQLite is the identity; Postgres maps "?" to "$n" ordinals. 

var Postgres Dialect = pgq

Postgres is the placeholder-rewriting dialect (pgq). 

var SQLite Dialect = func(query string) string { return query }

SQLite passes the shared SQL through unchanged.

type EngineEnqueuer added in v0.4.1 

type EngineEnqueuer interface {
	Enqueue(ctx context.Context, n utask.New[[]byte]) (utask.Task[[]byte], error)
}

EngineEnqueuer is the slice of the upstream engine the shared Enqueue path needs (both backend engines are instantiated at T=[]byte).

type Runner 

type Runner interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

Runner is the query surface every companion function operates on: a *sql.DB, a *sql.Tx, or the dialed wrapper For returns.

func For 

func For(d Dialect, r Runner) Runner

For bakes the dialect into a handle; adapters call it once per handle and companion functions keep their bodies dialect-free.

type StoreOption 

type StoreOption func(*storeOptions)

StoreOption configures optional store behavior (the adapters alias it so their public option surface stays put).

func WithProjectExclusivity 

func WithProjectExclusivity() StoreOption

WithProjectExclusivity mirrors internal/queue/sqlite's option: ClaimDue will not claim a task whose project already has another running task. The upstream engine has no such predicate, so ClaimDue is adapter-owned (divergence noted in the S1 report).

Source Files

View all Source files

Directories

Path Synopsis
Package conform holds the shared conformance suite for the S1 spike queue backends (sqlitev4, postgresv4).
 Package conform holds the shared conformance suite for the S1 spike queue backends (sqlitev4, postgresv4).

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.