readmodel

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: 32 Imported by: 0

Documentation

Overview

Package readmodel is the ADR-0019 S3 read model: tq's task ledger and status counts as go-cqrs-lite metaengine Store collections (planned tables), folded from the fact journal, with metaengine Watcher/ServeSSE as the live-fragment mechanism that replaces the hand journal-tailer → hub fan-out (ADR-0003 Phase D read path) once the S3 flip lands.

Viewpoint — Consumer: the live dashboard and any tq reader that wants the task ledger + status counts without folding the journal itself; Concern: the S3 projection — planned metaengine collections folded from journal facts into a disposable side DB (<db>.readmodel.db) under the single-writer fold-lock election; Non-goals: being a second source of truth (the journal is truth; the projection rebuilds from it) and any write path that is not journal-first feeding.

Cross-package map (lifecycle × concern): docs/architecture-understanding/2026-10-23_lifecycle-concern-grid.md

The Model opens its OWN sqlite database beside the queue DB (path decision: separate file — the projection is disposable and rebuildable from the journal, and a second writer connection on the queue file would contend with the store's single serialized writer). Feeding is journal-first: every lifecycle fact maps to an event applied through metaengine Store.ApplyRecord, so the collections are a fold projection of the journal exactly like every other tq consumer (ADR-0001).

THIN-ENQUEUE SIDE CHANNEL (interim): the upstream engine's task.enqueued fact carries only {project, type} today — priority, dedup key and the creation timestamp are not in the journal (S1 divergence, M4 ratification memo pending). While that holds, the Projector fills those fields from the store row on first sight of a task (RowSource; nil disables). The side channel dies with the upstream detail growth: the fold for task.enqueued then reads everything from the fact.

The package is read-only over the queue: it never mutates task state.

Index

Constants

View Source 
const (
	// DefaultPoll is the journal tail interval (webui parity).
	DefaultPoll = 500 * time.Millisecond
	// DefaultBatch is the per-poll fact batch cap (webui tailBatchLimit
	// parity): the pump only needs to converge, so a longer burst
	// finishes over the next ticks.
	DefaultBatch = 1000
	// DefaultReplayCapacity bounds the SSE replay journal (Last-Event-ID
	// reconnection window).
	DefaultReplayCapacity = 1024
	// SSEHeartbeat is the keepalive cadence for the events stream.
	SSEHeartbeat = 15 * time.Second
	// SSETimeout caps one stream's lifetime so server shutdowns and
	// stuck clients reclaim their goroutines.
	SSETimeout = 30 * time.Minute
	// CursorConsumer is the watermarks-table key the durable cursor
	// checkpoints under (queue.Store.SaveWatermark): the projection
	// resumes from it instead of replaying the whole journal on every
	// open. One key per queue db — the projection file is PathFor-derived,
	// so one store has exactly one durable projection.
	CursorConsumer = "readmodel"
	// DefaultTaskPageSize is the TasksPage size when the caller passes
	// no limit: the row-rich fragment page, sized to render in one pass
	// while the cursor keeps deeper pages one read away.
	DefaultTaskPageSize = 200
)

Defaults for the projection pump and the SSE surface.

View Source 
const FoldLockSuffix = ".fold-lock"

FoldLockSuffix names the fold-lock sidecar beside a projection home.

Variables

View Source 
var ErrFoldLocked = errors.New("readmodel: projection fold lock held by another folder")

ErrFoldLocked reports the projection home's fold lock is held by another folder: the caller lost the single-writer election and must mount read-only.

View Source 
var ErrNoSource = errors.New("readmodel: nil journal source")

ErrNoSource reports an Open call without a journal source: the model is a projection, and without a journal to fold there is nothing to serve.

View Source 
var ProjectionHomeCallerPragmas = projectionHomeCallerPragmas()

Functions

func ConvergeProjectionHome added in v0.3.4 

func ConvergeProjectionHome(ctx context.Context, path string) error

ConvergeProjectionHome reconciles an on-disk projection home with the current TaskRow model BEFORE any engine opens the file: a projection written by an older model (a planned-table column the current fold inserts does not exist) makes every task.enqueued fold fail, the metaengine's CREATE TABLE IF NOT EXISTS never alters an existing table, and the projection worker crash-loops on the first fact until it exhausts its restart budget — the Oct 2026 production outage where meta_planned_tasks predated not_before.

The projection is DISPOSABLE by contract (delete the file to force a full journal replay), so a divergent home is snapshotted aside — renamed, never deleted — and the next open recreates it from the current model; WithDurableCursor already replays from zero exactly once when the surviving watermark points past an empty projection. A healthy or absent home is left untouched.

func DLQPathFor added in v0.3.3 

func DLQPathFor(modelPath string) string

DLQPathFor derives the poison-fact sidecar path beside a projection database: one DLQ per projection home, discovered by both the serve wiring (OpenDeadLetters) and the doctor's projection section.

func EventIDToSeq added in v0.3.3 

func EventIDToSeq(eventID id.EventID) (int64, bool)

EventIDToSeq decodes a sequence-derived event ID back to its journal sequence; ok is false for any ID not minted by this layout (a random or foreign cursor), which callers must treat as "no position".

func FoldLockPath added in v0.4.1 

func FoldLockPath(home string) string

FoldLockPath derives the fold-lock sidecar path for a projection home (the PathFor file).

func Lag added in v0.3.3 

func Lag(h *projectionhost.Host) time.Duration

Lag reports the host's aggregate fold lag — the doctor/health surface (M09) reads it through the host the caller owns.

func PathFor 

func PathFor(dbPath string) string

PathFor derives the projection database path beside a queue database: one shared projection file per queue db. Sharing is REAL multi-process: the sqliteengine opens the file WAL + busy_timeout, and the fold-lock protocol (FoldLockPath) elects exactly ONE folder per home — a second folding Open mounts read-only (store-backed reads, no fold, no checkpoint) instead of double-folding the non-idempotent counters.

func SeqToEventID added in v0.3.3 

func SeqToEventID(seq int64) (id.EventID, error)

SeqToEventID encodes a journal sequence as the synthetic event ID the cqrs journal stamps on every fact event. Exported for the checkpoint adapter's consumers (doctor/health surfaces, composition root).

func StatusCounts added in v0.3.3 

func StatusCounts(ctx context.Context, model *Model, store queue.Store) (map[task.Status]int, error)

StatusCounts is the one seam both servers read status counts through: from the model when one is supplied, from the queue store otherwise. The dashboard server (internal/webui) and the machine API (internal/httpapi) shared this exact branch pair before it landed here.

func StringPtr 

func StringPtr(s string) *string

StringPtr is a small helper for building a TaskFilter inline.

func WorkerFailedMarkerPath added in v0.4.1 

func WorkerFailedMarkerPath(dbPath string) string

WorkerFailedMarkerPath derives the marker sidecar's path for a queue database: beside the projection db PathFor stamps.

func WriteWorkerFailedMarker added in v0.4.1 

func WriteWorkerFailedMarker(dbPath string, marker WorkerFailedMarker) error

WriteWorkerFailedMarker stamps the marker for a queue database (atomic temp+rename: the write itself happens on a crash path). This is the exported write point for tests and smokes; the live host writes through the same single function via its recorder.

Types

type DeadLetters added in v0.3.3 

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

DeadLetters is the projection host's poison-fact sidecar over one sqlite file beside the projection database (DLQPathFor). Serve opens it for the fold host's DeadLetterStore; doctor surfaces read counts and recent entries from the same file. Open creates the file and the dead-letter table; Close releases the handle.

func OpenDeadLetters added in v0.3.3 

func OpenDeadLetters(ctx context.Context, d config.Deployment) (*DeadLetters, error)

OpenDeadLetters opens (creating if needed) the DLQ sidecar for the projection home described by d (ADR-0022: the deployment is the ONE pragma source — the sidecar follows the same sync tier as both homes). The handle follows the queue's sqlite posture: WAL, busy timeout, MaxOpenConns(1).

func (*DeadLetters) Close added in v0.3.3 

func (d *DeadLetters) Close() error

Close releases the sidecar's handle.

func (*DeadLetters) Count added in v0.3.3 

func (d *DeadLetters) Count(ctx context.Context) (int64, error)

Count reports the stored poison facts across every projection.

func (*DeadLetters) Recent added in v0.3.3 

func (d *DeadLetters) Recent(ctx context.Context, limit int) ([]projectionhost.DeadLetterEntry, error)

Recent returns up to limit dead-letter entries, newest failure first.

func (*DeadLetters) Store added in v0.3.3 

func (d *DeadLetters) Store() *projectionhost.SQLiteDeadLetterStore

Store returns the platform DLQ for the projection host's DeadLetterStore option; the concrete store satisfies the interface.

type EngineHealthRow added in v0.4.1 

type EngineHealthRow struct {
	Name                string `json:"name"`
	State               string `json:"state"`
	ConsecutiveFailures int    `json:"consecutiveFailures,omitempty"`
	LastError           string `json:"lastError,omitempty"`
}

EngineHealthRow is one projection engine's health snapshot — the tq-side JSON shape over metaengine's EngineHealth (no upstream types on the wire).

type FailureText 

type FailureText string

FailureText is a failed attempt's error text — a named string so event structs carry at most one bare string (the fold key).

type FoldLock added in v0.4.1 

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

FoldLock is one folder's held lease on a projection home. The model owns the lease from adoption on (WithFoldLock, or the internal acquire in Open) and releases it on Close.

func AcquireFoldLock added in v0.4.1 

func AcquireFoldLock(home string, wait time.Duration) (*FoldLock, error)

AcquireFoldLock takes a projection home's fold lock: one non-blocking exclusive acquire for wait <= 0, otherwise the acquire retried every foldLockPoll until wait has elapsed (the composition runtime's bounded wait rides out a transient holder — a stats/doctor catchup — instead of demoting the long-lived serve folder on a startup overlap).

ErrFoldLocked reports the election lost (mount read-only); any other error is an I/O failure, not contention. The lock file itself is never unlinked: unlinking under a held lock lets a third folder lock a fresh inode while the second still holds the old one — two folders, the exact corruption the lock exists to prevent. A stale zero-byte sidecar beside the projection home is the steady state, not a leak.

func (*FoldLock) Release added in v0.4.1 

func (l *FoldLock) Release() error

Release drops the lease. Safe to call twice (the second is a no-op), so a caller-side cleanup may run after an Open-internal release.

type FoldProjection added in v0.3.3 

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

FoldProjection feeds platform-delivered fact events through the same fold the hand pump uses (Model.apply) — one fold, two drivers.

func NewFoldProjection added in v0.3.3 

func NewFoldProjection(m *Model) FoldProjection

NewFoldProjection wraps the model's fold as a platform projection.

func (FoldProjection) EventTypes added in v0.3.3 

func (p FoldProjection) EventTypes() []event.Type

EventTypes delivers every fact family, matching the hand pump's deliver-everything behavior exactly: the fold's default arm skips the no-ledger families, and a future family must reach Handle without an edit here.

func (FoldProjection) Handle added in v0.3.3 

func (p FoldProjection) Handle(ctx context.Context, evt event.Event) error

Handle reconstructs the fact from the platform event and folds it. The payload was stamped by the cqrs journal's factEvent (event.New with the JSON codec), so DecodePayloadAuto is the matching decode path.

func (FoldProjection) Name added in v0.3.3 

func (p FoldProjection) Name() string

Name is the projection identity and the watermarks consumer.

type FromProject added in v0.4.1 

type FromProject string

FromStatus is a transition's previous folded status and FromProject its previous folded project — named strings so the engine's key inference never reads them as fold keys. The map folds ignore both; the Delta counter folds (status_counts, project_status_counts) consume them to decrement the from-side exactly. The model fills them at apply time with an O(1) keyed read of the row the map fold is about to update — the from-status-ambiguous transitions (lease reclaims, dead rescues, dismissals) carry no statically knowable from-status.

type FromStatus added in v0.4.1 

type FromStatus string

FromStatus is a transition's previous folded status and FromProject its previous folded project — named strings so the engine's key inference never reads them as fold keys. The map folds ignore both; the Delta counter folds (status_counts, project_status_counts) consume them to decrement the from-side exactly. The model fills them at apply time with an O(1) keyed read of the row the map fold is about to update — the from-status-ambiguous transitions (lease reclaims, dead rescues, dismissals) carry no statically knowable from-status.

type Model 

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

Model is the metaengine-backed read model over one queue's fact journal: the tasks collection (planned table) plus the Watcher/ServeSSE live surface. Create with Open, pump with Run (or CatchUp for one pass), read with Tasks/StatusCounts, stream with EventsHandler.

func Open 

func Open(ctx context.Context, path string, src queue.Store, opts ...Option) (*Model, error)

Open creates the Model over its own sqlite database file (the projection is disposable; delete the file to force a full journal replay on next open). src is the journal source — the same queue.Store the dashboards read; it is used read-only (Facts, HeadSeq, Get). ctx covers the durable-cursor load (watermark read + replay guard), not the pump (Run carries its own).

Every Open participates in the projection home's single-writer election: a plain Open acquires the fold lock without blocking, and a lost election mounts the model read-only (store-backed reads, no fold, no checkpoint) so the holder stays the one folder — the Delta counter folds are not idempotent, and a second folder would double-count every transition. WithFoldLock callers elected earlier (their lease must cover the converge + engine open too); WithReadOnly callers never contend.

func (*Model) CatchUp 

func (m *Model) CatchUp(ctx context.Context) error

CatchUp applies every journal fact up to the current head once.

func (*Model) Close 

func (m *Model) Close() error

Close releases the watcher, the projection database, and the fold lease (a folder mount). The journal source is owned by the caller.

func (*Model) EventsHandler 

func (m *Model) EventsHandler() http.Handler

EventsHandler serves the live ledger as Server-Sent Events: every fold update streams as one JSON TaskRow, with `id: <seq>` for Last-Event-ID reconnection (metaengine's replay journal). This is the Watcher/ServeSSE mechanism the S3 flip installs behind the dashboard fragments.

func (*Model) JournalCursor 

func (m *Model) JournalCursor() int64

JournalCursor reports the applied journal watermark: the seq of the last fact folded into the collections (0 before the first apply). Live consumers use it as the change-notification sequence, so a notification carries the same journal watermark the hand tailer used to report.

func (*Model) ProjectCounts 

func (m *Model) ProjectCounts(ctx context.Context) (map[string]map[string]int, error)

ProjectCounts counts the ledger rows per project per status — the same matrix the store-side ProjectCounts GROUP BY serves (queue.Store vocabulary parity for the projection).

func (*Model) ProjectionHealth added in v0.4.1 

func (m *Model) ProjectionHealth(ctx context.Context) ProjectionHealth

ProjectionHealth reads the live projection surfaces. Head errors surface as lag -1 with the error — a health endpoint must answer, not fail.

func (*Model) ReadOnly added in v0.4.1 

func (m *Model) ReadOnly() bool

ReadOnly reports the model mounted without the fold role: the fold-lock election was lost (another folder owns the projection home) or the caller forced WithReadOnly. Reads are store-backed as-is; folding calls are no-ops; a pump host should NOT be built over it (the lock holder does the folding — a read-only host would only re-checkpoint the shared watermark without moving any row).

func (*Model) Run 

func (m *Model) Run(ctx context.Context) error

Run pumps the journal until ctx is cancelled: catch up to the head, then tail for new facts. Without WithDurableCursor the cursor is in-process — a restarted model replays the journal from the beginning, which the folds converge on (every fold is an upsert keyed by task id); with it, the model resumes from the persisted checkpoint instead. A read-only mount has no fold to pump: it parks until ctx is done, so a pump owner sees a healthy long-running Run instead of an exit to restart-loop on.

func (*Model) Stats 

func (m *Model) Stats(
	ctx context.Context,
	filter TaskFilter,
) (map[string]int, map[string]map[string]int, error)

Stats reads the status counts and the per-project-per-status counts from the Delta counter collections (upstream metaengine COOKBOOK § Status Count Dashboard) — O(1) counter reads, zero rows loaded. The counters fold the same fact stream the tasks map folds, with the from-side resolved by the apply-time prev read, so they stay exact across the from-status-ambiguous transitions (lease reclaims, dead rescues, dismissals).

filter narrows both matrices exactly like a Tasks read would: a Status filter leaves only that status's counts, a Project filter only that project's. A project-scoped byStatus derives from the project|status counters — the global status counter cannot narrow by project.

func (*Model) StatusCounts 

func (m *Model) StatusCounts(ctx context.Context) (map[string]int, error)

StatusCounts counts the ledger rows per status — the Delta counter read behind the stats surfaces (dashboard chips, /api/stats, tq stats).

func (*Model) Tasks 

func (m *Model) Tasks(ctx context.Context, filter TaskFilter) ([]TaskRow, error)

Tasks reads the ledger under the filter, newest creation first — the planned-table pushdown scan. UNBOUNDED on purpose: the export path that genuinely wants every row. Row-rich views page through TasksPage instead. Filters ride the documented TypedReader surface (conditional WithFilter options): the query-input dispatch (Store.ExecuteCtx) cannot express optional filters — a nil *string input field binds as a typed-nil interface and lands in SQL as `= NULL` (metaengine v4.14.0, extractValueByName), so every read here goes through the reader.

func (*Model) TasksPage added in v0.4.1 

func (m *Model) TasksPage(ctx context.Context, filter TaskFilter, page TaskPage) ([]TaskRow, string, error)

TasksPage reads one page of the ledger under the filter, newest creation first, through the TypedReader keyset cursor round-trip (metaengine ScanPage + WithCursor, the upstream COOKBOOK §Cursor Pagination shape): each read returns the rows plus the HTTP-safe encoded continuation cursor (empty = last page). The sqlite engine's compound sort-key cursor keeps the created_at ties (unix-milli timestamps collide constantly) drop- and duplicate-free across pages. Row-rich views — the board/table fragments — read through this; Tasks stays the unbounded export path.

func (*Model) Watch 

func (m *Model) Watch(ctx context.Context) <-chan TaskRow

Watch subscribes to live ledger changes: every fold update arrives as the folded TaskRow (buffered, drop-oldest on a slow consumer). The subscription ends with ctx. This is the Watcher half of the S3 live fragments; EventsHandler is its SSE transport.

func (*Model) WatchSeq 

func (m *Model) WatchSeq(ctx context.Context) <-chan metaengine.SeqValue[TaskRow]

WatchSeq is Watch with the projection write sequence attached. The dashboard's notification pump consumes this: each delivery replaces one journal-tailer poll cycle — the change signal, not the payload.

type Option 

type Option func(*Model)

Option configures the Model.

func WithBatch 

func WithBatch(n int) Option

WithBatch overrides the per-poll fact batch cap.

func WithDurableCursor 

func WithDurableCursor() Option

WithDurableCursor checkpoints the projection cursor into the queue db's watermarks table after every applied batch and resumes from it on open: a restarted model folds only the facts appended since its last checkpoint instead of replaying the whole journal (the 9.5k-fact restart replay the adoption review measured). An empty projection under a nonzero checkpoint replays from zero once, so the documented "delete the file to force a replay" escape hatch keeps working. Checkpoint writes are monotonic upserts; a failed write is logged and retried on the next batch — a missed checkpoint only costs replay on the next restart, never a skipped fact.

func WithEngine added in v0.3.4 

func WithEngine(eng metaengine.Engine) Option

WithEngine adopts a caller-built engine instead of constructing one: the composition root opens the projection-home engine ONCE and hands it over (the single tq-owned constructor call per serve run). From adoption on the model owns the engine exactly as for a self-opened one — the FromDSN constructors mark their engine as the DB owner, so Close tears the connection down with the model.

func WithFoldLock added in v0.4.1 

func WithFoldLock(lock *FoldLock) Option

func WithPoll 

func WithPoll(d time.Duration) Option

WithPoll overrides the journal tail interval.

func WithReadOnly added in v0.4.1 

func WithReadOnly() Option

WithReadOnly mounts the model WITHOUT the fold role: store-backed reads only — no fold-lock acquisition, no fold, no checkpoint. The explicit reader's option (tq doctor's projection check): unlike a lost fold-lock election it never contends with a running folder and never logs. Overrides WithFoldLock (a read-only mount never holds the lease).

func WithRowSource 

func WithRowSource(rs RowSource) Option

WithRowSource overrides the thin-enqueue side channel (nil disables it). The default adapts the journal source's Get.

type ProjectionHealth added in v0.4.1 

type ProjectionHealth struct {
	// Cursor is the applied journal watermark (JournalCursor).
	Cursor int64 `json:"cursor"`
	// Head is the journal's highest seq at read time.
	Head int64 `json:"head"`
	// Lag is Head - Cursor; 0 means the fold is at the head.
	Lag int64 `json:"lag"`
	// Checkpointed reports whether a durable cursor checkpoint exists in
	// the queue db's watermarks table.
	Checkpointed bool `json:"checkpointed"`
	// Checkpoint is the persisted checkpoint seq (0 when none).
	Checkpoint int64 `json:"checkpointSeq,omitempty"`
	// Engines is the projection store's per-engine health snapshot.
	Engines []EngineHealthRow `json:"engines"`
}

ProjectionHealth is the token-gated /health projection section: the fold's cursor lag against the journal head, the durable checkpoint state, and the projection engines' health (ADR-0019 S3 observability — the same surfaces `tq doctor`'s projection section reads, live).

type ProjectionHost added in v0.3.3 

type ProjectionHost struct {
	*projectionhost.Host
	// contains filtered or unexported fields
}

ProjectionHost pairs the platform host with the live tails' lifecycle: the platform never closes its subscriber (the Bus doc leaves cleanup to the owner via io.Closer), so Close here is the tail-ownership half and Stop stays the platform half. Embedding keeps Start, Stop, and Register on the wrapper.

func NewProjectionHost added in v0.3.3 

func NewProjectionHost(src queue.Store, m *Model, opts ProjectionHostOptions) (*ProjectionHost, error)

NewProjectionHost builds the managed host over the fact journal with the model's fold registered as its projection. Start/Stop stay with the caller (the composition root owns the lifetime); the journal is the queue store itself, read-only (Facts). The live tail after the initial drain polls the journal at DefaultPoll — the queue journal has no bus, and the webui contract is a tail, not a one-shot drain.

func (*ProjectionHost) Close added in v0.3.3 

func (h *ProjectionHost) Close() error

Close stops every tail goroutine the subscriber spawned (one per worker generation: a worker restart re-subscribes) and waits for their exit, so a closed journal never sees a late poll.

type ProjectionHostOptions added in v0.3.3 

type ProjectionHostOptions struct {
	// CheckpointEvery batches live-phase checkpoint saves; 0 keeps the
	// platform default.
	CheckpointEvery int
	// BatchSize caps the journal read per drain loop; 0 keeps the
	// platform default.
	BatchSize int
	// MaxRestarts bounds the crash-restart budget per worker; 0 keeps
	// the platform default (5).
	MaxRestarts int
	// DeadLetterStore captures poison facts after Threshold failures;
	// nil disables the DLQ.
	DeadLetterStore projectionhost.DeadLetterStore
	// DeadLetterThreshold is the failure count that routes a fact to the
	// DLQ; 0 keeps the platform default when a store is configured.
	DeadLetterThreshold int
}

ProjectionHostOptions are the platform knobs the composition root may tune; zero values take the projectionhost defaults except where the queue's operating posture demands one (checkpoint batching on).

type RequeueClass 

type RequeueClass string

RequeueClass is a requeue evidence class — a named string so the engine's type-based key inference keeps ID unambiguous (same rule as FailureText).

type RowSnapshot 

type RowSnapshot struct {
	Project   string
	Type      string
	Priority  int
	DedupKey  string
	CreatedAt int64
}

RowSnapshot is the enqueue-time slice of a task row.

type RowSource 

type RowSource interface {
	// EnqueueRow reports the enqueue-time projection of one task; a
	// miss (task already compacted away) returns ok=false and the fold
	// keeps the fact-detail fallbacks.
	EnqueueRow(ctx context.Context, id string) (row RowSnapshot, ok bool, err error)
}

RowSource supplies the enqueue-time fields the current engine's thin task.enqueued fact omits (project/type fall back to the fact detail; priority, dedup key and creation time exist only on the row). It dies with the upstream detail growth (M4 memo) — the fold then reads everything from the fact.

type StoreRows 

type StoreRows struct {
	Store queue.Store
}

StoreRows adapts a queue.Store into a RowSource via Get.

func (StoreRows) EnqueueRow 

func (s StoreRows) EnqueueRow(ctx context.Context, id string) (RowSnapshot, bool, error)

EnqueueRow implements RowSource.

type TaskFilter 

type TaskFilter struct {
	Status  *string
	Project *string
}

TaskFilter selects the Tasks read. Nil fields mean "no filter"; the values are the raw queue vocabulary (task.Status string, project name). Deliberately a separate type from TaskList (queries.go): TaskList is the metaengine query-input declaration carrier and MUST NOT dispatch reads through it (typed-nil binding hazard), while TaskFilter is the caller-facing filter the Model reads satisfy.

type TaskList 

type TaskList struct {
	Status  *string
	Project *string
}

TaskList is the query's declared input type — a declaration carrier only. Reads MUST NOT dispatch through it (Store.ExecuteCtx binds nil pointer fields as typed-nil interfaces → `= NULL`, and empty strings as real filters): Model.Tasks reads via the TypedReader instead.

type TaskPage added in v0.4.1 

type TaskPage struct {
	Limit  int
	Cursor string
}

TaskPage bounds and continues a TasksPage read: Limit is the page size (<= 0 takes DefaultTaskPageSize), Cursor the opaque continuation from the previous page's next-cursor ("" = first page). The cursor is the HTTP-safe base64 encoding metaengine's Cursor.Encode mints — pass it back exactly as received; a malformed cursor fails the read instead of silently restarting pagination.

type TaskRow 

type TaskRow struct {
	ID string `json:"id"`
	// Project is the task's repo/project scope (empty = unscoped).
	Project string `json:"project"`
	// Type is the executor task type (sh, agent, review, ...).
	Type string `json:"type"`
	// Status is the task.Status value (pending, running, completed,
	// dead, cancelled).
	Status string `json:"status"`
	// Priority is the STORED priority (ADR-0015); aging is scheduling
	// and never mutates it.
	Priority int `json:"priority"`
	// Attempts counts claimed leases — every claim is one attempt start.
	Attempts int `json:"attempts"`
	// CreatedAt/UpdatedAt are unix millis — the task-row storage format.
	CreatedAt int64 `json:"created_at"`
	UpdatedAt int64 `json:"updated_at"`
	// DedupKey is the harvest/enqueue identity (empty for ad-hoc tasks).
	DedupKey string `json:"dedup_key"`
	// LastError is the most recent failure text (empty when none).
	LastError string `json:"last_error"`
	// NotBefore is the earliest claim time (unix millis) of the latest
	// CLASS park (from the requeue evidence), zero when none is in
	// force. The store's backoff-after-failure parks are deliberately
	// NOT mirrored: task.failed facts carry no class or window, so the
	// projection records only what the journal proves.
	NotBefore int64 `json:"not_before"`
	// ParkedBy is the latest requeue class (queue.RequeueClass*;
	// "unknown" for legacy facts, empty until the first requeue). A
	// budget-parked task is Status==pending with ParkedBy=="budget" and
	// a future NotBefore. Cleared on claim and on failure (a burned
	// attempt is a retry, not a park).
	ParkedBy string `json:"parked_by"`
}

TaskRow is one ledger row in the tasks collection: the fields the queue views (board, table, stats) render, folded from the journal.

type WatermarkCheckpoints added in v0.3.3 

type WatermarkCheckpoints struct {
	Src queue.Store
}

WatermarkCheckpoints adapts the queue's watermarks table (one monotonic integer cursor per consumer) to the platform's event.CheckpointStore. The projection's Name IS the watermarks consumer, so the host-managed pump and the in-model durable cursor (WithDurableCursor) share one checkpoint slot by construction.

func (WatermarkCheckpoints) Load added in v0.3.3 

func (w WatermarkCheckpoints) Load(ctx context.Context, name string) (event.Checkpoint, error)

Load returns the stored checkpoint; no row (or a zero seq) is the zero Checkpoint — the platform's "no prior progress" shape, replaying from the journal start.

func (WatermarkCheckpoints) Save added in v0.3.3 

func (w WatermarkCheckpoints) Save(ctx context.Context, name string, cp event.Checkpoint) error

Save stores the checkpoint's event ID as the consumer's seq. A non-sequence-derived ID cannot come from this journal's events and is refused instead of silently wedging the cursor.

type WorkerFailedMarker added in v0.4.1 

type WorkerFailedMarker struct {
	Projection string    `json:"projection"`
	FailedAt   time.Time `json:"failedAt"`
}

WorkerFailedMarker is the sidecar's JSON shape: which projection's worker died and when (last failure wins — one terminal marker, no history).

func ReadWorkerFailedMarker added in v0.4.1 

func ReadWorkerFailedMarker(dbPath string) (marker WorkerFailedMarker, exists bool, err error)

ReadWorkerFailedMarker reads the marker for a queue database; exists is false when no marker stands (the healthy case).

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.