sqlitev4

package module
v0.4.0 Latest Latest
Warning

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

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

Documentation

Overview

Package sqlitev4 is the ADR-0019 S1 spike: a tq queue.Store implemented over go-cqrs-lite queue/sqlite/v4 (the upstream engine transcribed from this repo's contract) plus tq-side companion surfaces over 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 single-writer handle, and one-line delegations.

Division of labor:

  • Engine-backed (upstream queue/sqlite/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 docs/status/2026-09-23_16-28_adr0019-s1-sqlitev4-conformance-divergence-report.md (D1 finalizes, D2 enqueued detail, D3 heartbeat facts, D4 fact archive, D5 payload storage class; D6 exclusivity/resume-closeout are adapter-implemented, not divergences).

Index

Constants

This section is empty.

Variables

View Source 
var WithProjectExclusivity = companion.WithProjectExclusivity

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 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 engine + companion surfaces on the same SQLite file.

func Open 

func Open(path string, opts ...StoreOption) (*Store, error)

Open opens (creating if needed) the queue database at path. The sync tier comes from the TQ_SQLITE_SYNC environment, exactly as before this signature existed — the deployment-struct entry point is OpenWithDeployment, which takes the tier resolved.

func OpenWithDeployment added in v0.3.4 

func OpenWithDeployment(d config.Deployment, opts ...StoreOption) (*Store, error)

OpenWithDeployment opens the store from a deployment struct (ADR-0022's deployment lane): the struct's pragma builder renders the connection DSN — the ONE pragma literal lives in internal/config — and the sync tier arrives resolved, never read from the environment.

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 shared pool. The engine and the companion runner ride ONE database pool (openSharedDB): the engine's Close already closes it, and the second Close on a *sql.DB is a documented idempotent no-op — the double call keeps this robust against either handle shape.

art-dupl:accept two-handle teardown: each 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 sqlite 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

Directories

Path Synopsis
Package migration implements the ADR-0019 S1 data migration and the endgame P1 on-open auto-upgrade: it replays or converges a tq fact journal — the hand-rolled store's database — onto the go-cqrs-lite engine store and verifies projection equality (StatusCounts, ProjectCounts, the full fact stream, head seq, per-task fact tails, DLQ contents, watermarks, priority scores).
 Package migration implements the ADR-0019 S1 data migration and the endgame P1 on-open auto-upgrade: it replays or converges a tq fact journal — the hand-rolled store's database — onto the go-cqrs-lite engine store and verifies projection equality (StatusCounts, ProjectCounts, the full fact stream, head seq, per-task fact tails, DLQ contents, watermarks, priority scores).
Command replay is the ADR-0019 S1 data-migration CLI: it replays a tq fact journal — the hand-rolled store's database — into a fresh go-cqrs-lite engine store and verifies projection equality.
 Command replay is the ADR-0019 S1 data-migration CLI: it replays a tq fact journal — the hand-rolled store's database — into a fresh go-cqrs-lite engine store and verifies projection equality.

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.