README
¶
watch (package)
Import: github.com/kausys/azync/watch
User guide: ../watch.md · GoDoc: package docs via go doc / pkg.go.dev.
Role
Change-hint observer over the Core: streams best-effort, at-most-once row-change hints (jobs, DAG headers, ledger events) to external consumers — ops UIs, SSE bridges — that would otherwise poll. It runs no jobs, so unlike the four job-running runtimes it does not embed azync.Defaults; its only knob is the per-subscription buffer.
Source layout
| File / area | Responsibility |
|---|---|
watch.go |
New / Open, Watcher.Watch, Filter, options |
doc.go |
Package model: hints, resets, bulk |
Driver surface
Requires the optional driver.ChangeNotifier capability (azyncpgx: migration 00011's triggers on azync_jobs / azync_dags / azync_events NOTIFYing the fixed azync_changes channel; a second, lazily-opened LISTEN connection).
Public surface (summary)
New(core)/Open(dsn)— capability-asserting compositionWatcher.Watch(ctx, Filter)— filtered subscription; first delivery is always anEntityResetFilter{Entities, Sources, Kinds, DAGID}— resets always pass; bulk hints pass the kind/DAG boundsChange/Entity/Source— aliases of the driver typesErrPollOnly— Watch's sentinel for a driver that cannot pushWithBuffer,WithCoreOptions
Boundaries
- No import of
queue/event/dag/workflow(or vice versa); composes only through the Core. - Hints carry identifiers, kind/name, state and a timestamp — never payloads, results or meta (PII rule; full rows are read through the Managers behind the caller's authz).
- Not a durable feed: no replay, no cross-entity ordering; every possible gap is announced as an in-band reset.
Tests
go test ./watch/... · ChangeNotifier conformance in driver/drivertest.
Documentation
¶
Overview ¶
Package watch streams row-change hints from the backend so external observers — ops UIs, SSE/WebSocket bridges, cache invalidators — can react to state transitions without polling.
Model ¶
The stream carries hints, not data: each Change names the entity that changed (a job, a DAG header, a ledger event), its identifiers and its new state — never payloads, results or metadata, which routinely carry PII. Delivery is best-effort and at-most-once. Every gap a subscription can suffer — its own startup, a backend reconnect, a full buffer — is announced in-band as an EntityReset change, and the first delivery on every subscription is one. That yields a single consumer rule:
On a reset — and you always receive one first — refetch everything you care about through the Manager surfaces.
A statement that changes more rows than the backend's per-statement cap is coalesced into one Bulk hint per partition; treat it as "refetch this partition". The stream is not a durable feed: there is no replay, no ordering guarantee across entities, and the authoritative state always lives behind the Managers.
Filtering ¶
Watcher.Watch takes a Filter bounding entities, job sources, kinds and a single DAG. Resets always pass a filter; bulk hints pass the kind and DAG bounds (they carry neither).
Compose a Watcher over a shared Core with New, or standalone with Open; the driver must implement the change-notification capability (driver.ChangeNotifier).
Index ¶
Constants ¶
const ( // EntityJob marks a change to a job row (any Source). EntityJob = driver.ChangeJob // EntityDAG marks a change to a DAG header row. EntityDAG = driver.ChangeDAG // EntityEvent marks an append to the event ledger. EntityEvent = driver.ChangeEvent // EntityReset signals a possible delivery gap: refetch. Every // subscription receives one reset before any other change. EntityReset = driver.ChangeReset )
Re-exported entity constants, so callers filter without importing driver.
Variables ¶
var ErrPollOnly = errors.New("watch: driver is poll-only; change notifications are unavailable")
ErrPollOnly reports that the driver cannot push change notifications (the backend runs poll-only, e.g. azync.PollOnly or a PgBouncer transaction pool). Test with errors.Is and branch: a caller bridging Watch to a live endpoint typically maps it to "stream unavailable" rather than retrying.
Functions ¶
This section is empty.
Types ¶
type Change ¶
Change is one best-effort, at-most-once row-change hint. See driver.Change for the field contract; the one rule that matters to a consumer is: on an EntityReset change, refetch everything you care about through the Manager surfaces — the stream never replays what a gap lost.
type Filter ¶
type Filter struct {
// Entities admits only these entity kinds when non-empty.
Entities []Entity
// Sources admits only job changes of these runtimes when non-empty; it
// never excludes dag-header or ledger-event changes.
Sources []Source
// Kinds admits only these job kinds / event types / DAG definition names
// when non-empty.
Kinds []string
// DAGID, when non-zero, admits only that DAG's header change and its
// task-job changes.
DAGID uuid.UUID
}
Filter selects the changes a Watch subscription receives. A zero field means "no bound". Two kinds of change bypass parts of a filter by design: reset changes always pass (a gap concerns every consumer), and bulk changes pass the Kinds and DAGID bounds (a coalesced hint carries no kind or ids — Entities and Sources still apply).
type Option ¶
type Option func(*config) error
Option configures a Watcher. Options compose; later options win.
func WithBuffer ¶
WithBuffer overrides the per-subscription delivery buffer (default 256). A larger buffer tolerates slower consumers before hints collapse into a reset. Must be positive.
func WithCoreOptions ¶
WithCoreOptions forwards options to the Core that Open builds internally (schema, logger, notify channel, shared defaults...). Valid only with Open; New rejects it because the Core is already constructed.
type Watcher ¶
type Watcher struct {
// contains filtered or unexported fields
}
Watcher is the change-hint subscription surface over one azync Core. It is an observer: it runs no jobs and owns no worker — it only fans the driver's driver.ChangeNotifier stream out to filtered subscriptions. Pure library, no auth: gate it behind your own authorization like the Managers.
func New ¶
New composes a Watcher over a shared Core. It fails when the Core's driver does not implement driver.ChangeNotifier.
func Open ¶
Open builds a standalone Watcher that owns a private Core opened from dsn (pass Core options through WithCoreOptions). Close closes the owned Core. Open never migrates; call Migrate before using a fresh schema.
func (*Watcher) Close ¶
Close releases the private Core when the Watcher was built with Open; composed over a shared Core it is a no-op.
func (*Watcher) Migrate ¶
Migrate brings the backend schema up to date (requires a driver.Migrator). Open and New never migrate automatically.
func (*Watcher) Watch ¶
Watch subscribes to change hints matching f. The channel is closed when ctx ends or the store closes; the first delivery is always an EntityReset. Hints are best-effort and at-most-once: when the subscription's buffer fills, dropped hints are replaced by one in-band reset, so a slow consumer sees "refetch" instead of a silent gap. A poll-only driver cannot push changes; Watch reports that as ErrPollOnly.
Source Files