Documentation
¶
Overview ¶
Package watermark holds the durable journal cursor shared by the journal-driven sweepers (review, dlqfix, status, prioritize): one row per consumer in the queue store's watermarks table, paged Facts reads, and a checkpoint AFTER each consumed page — never before, or a crash would silently skip the page's facts. A first use bootstraps at the journal head (eagerly persisted), so facts that predate a sweeper are not replayed.
Index ¶
Constants ¶
const DefaultPageSize = 500
DefaultPageSize bounds one Facts page per sweep iteration.
Variables ¶
This section is empty.
Functions ¶
func PermanentOutcome ¶ added in v0.4.0
PermanentOutcome reports that a store failure can never succeed on retry: the record is gone (task.ErrNotFound) or the transition already happened elsewhere (task.ErrInvalidTransition, the idempotent-replay class). Sweeper refusal gates checkpoint past permanent outcomes — one bad record must not wedge the cursor — and treat everything else as transient and refuse.
Types ¶
type Config ¶
type Config struct {
// Store pages the facts and owns the watermarks row.
Store Source
// Key is the cursor's identity in the watermarks table (the sweeper's
// ConsumerKey), shared by every sweeper over the same database.
Key string
// Domain names the owner in error messages ("review sweep"), keeping
// the wrapped errors grep-identical across the sweepers.
Domain string
// PageSize bounds one Facts page; 0 selects DefaultPageSize.
PageSize int
}
Config builds one Cursor.
type Cursor ¶
type Cursor struct {
// contains filtered or unexported fields
}
Cursor is a durable, page-checkpointing journal cursor. It is safe for concurrent use: sweeps serialize on an internal mutex, so ticks and the --once drain watcher may call Sweep from different goroutines.
func New ¶
New returns a cursor resuming from the persisted checkpoint — facts recorded while no sweeper was running are consumed on the next start. A first run bootstraps at the journal head and eagerly persists it (even 0), so a crash before the first sweep still resumes exactly here and never replays history. Rewind deliberately with `tq watermarks set <key> SEQ` — replay idempotency is the sweeper's concern (dedup keys and transition guards), not the cursor's.
func (*Cursor) Sweep ¶
Sweep consumes new facts since the last pass, handing each to handle in Seq order. Idempotent by the sweeper's dedup and transition guards: a replayed page (crash between consumption and checkpoint) re-hits dedup keys instead of duplicating work. The cursor checkpoints after each page; a failed checkpoint (or handler) stops the sweep — facts are never consumed past an unpersisted cursor. A handler rejection pulls the cursor back behind the rejected fact, so a retry sweep in the SAME process redelivers it too: pending checkpoints cover accepted facts only.
type Handler ¶
Handler reacts to one consumed fact. An error aborts the sweep before the page's checkpoint, so the fact redelivers on the next sweep (at-least-once).
type Source ¶
type Source interface {
Facts(ctx context.Context, after int64, limit int) ([]journal.Fact, error)
HeadSeq(ctx context.Context) (int64, error)
Watermark(ctx context.Context, consumer string) (int64, bool, error)
SaveWatermark(ctx context.Context, consumer string, seq int64) error
}
Source is the pull seam the cursor pages; queue.Store satisfies it.
Source Files