azync

package module

v0.0.8 Latest Latest Go to latest Published: Jul 31, 2026 License: MIT Imports: 9 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/kausys/azync

Links

Open Source Insights

README ¶

azync

Durable background work for Go over one job table and a pluggable driver.Store. First-party PostgreSQL driver: azyncpgx.

Runtimes

Package	`source`	Model	Guide	Package notes
`queue`	`queue`	Typed jobs, cron, transactional outbox	queue.md	README
`event`	`event`	CQRS ledger + fan-out deliveries + Replay	event.md	README
`dag`	`dag`	Static declared graph (no replay)	dag.md	README
`workflow`	`workflow`	Workflow-as-code (Go + history replay)	workflow.md	README
`watch`	—	Change-hint observer (runs no jobs)	watch.md	README

All four runtimes compose over one azync.Core and never import each other; watch observes the same Core under the same isolation rule.

Guides (queue.md, …) — how to use the runtime in an application.
Package notes (queue/README.md, …) — technical layout, driver surface, boundaries for maintainers.

Why azync

One table for runnable work. Queue jobs, event deliveries, DAG tasks, and workflow tasks share azync_jobs, partitioned by source.
Durable event bus. Insert-only ledger, atomic fan-out, Replay from history.
Transactional outbox. Enlist enqueue / publish / DAG run in your own backend transaction.
Driver contract. Implement driver.Store (+ optional capabilities); validate with the conformance suite.
Lease fencing + reaper. At-least-once delivery with real fencing tokens.
Two orchestration styles. Declared graphs (dag.md) or ordinary Go with deterministic replay (workflow.md).
Change hints for ops UIs. Row-change notifications over LISTEN/NOTIFY (watch.md) — no polling, no reload.

Install

go get github.com/kausys/azync@latest
go get github.com/kausys/azync/driver/azyncpgx@latest

Go 1.26+, PostgreSQL 13+.

import (
    "github.com/kausys/azync"
    "github.com/kausys/azync/queue"
    "github.com/kausys/azync/event"
    "github.com/kausys/azync/dag"
    "github.com/kausys/azync/workflow"

    _ "github.com/kausys/azync/driver/azyncpgx"
)

core, err := azync.Open(dsn)
if err != nil { /* ... */ }
if err := core.Migrate(ctx); err != nil { /* ... */ } // always explicit

q, _ := queue.New(core)
ev, _ := event.New(core)
d, _ := dag.New(core)       // needs driver.DAGStore
wf, _ := workflow.New(core) // needs driver.WorkflowStore

Open / New never migrate. Prefer one shared Core per process (examples/shared-core); use queue.Open / event.Open / … when a runtime should own a private Core.

Guides and examples

Guide	Example
queue.md	`examples/queue-basic`
event.md	`examples/event-basic`
dag.md	`examples/dag-basic`
workflow.md	`examples/workflow-kyc`
watch.md	`examples/watch-sse`

Roadmap

Concurrent Operations in workflow-as-code. Today ExecuteOperation always parks — Operations are strictly serial, which is what rules the workflow runtime out for fan-out shapes (N verifications in parallel; use dag for those). The design: an ExecuteOperationAsync that records OperationScheduled without parking and returns an unready Future; Select extended to park on Operation futures; and replay matching keyed by execution_key instead of strict sequence, so out-of-order completions replay deterministically. With that, workflow-as-code supports fan-out and the dag-vs-workflow choice becomes expressiveness, not capability.
Workflow runtime on the shared engine. The workflow worker bypasses internal/engine today, so it lacks the consumer spans, metrics and trace propagation the other three runtimes get for free. Migrating it is a large refactor with an observable payoff.
Compensation in workflow-as-code stays a documented pattern (the workflow function can run its own compensating Operations after a failed step — it is plain code), not a new primitive; dag keeps first-class Compensate for declarative chains.
Workflow-as-code hardening (continue-as-new, child workflows, richer archives)
Scheduled DAGs; first-class parent/child links
Admin UI over the Manager surfaces
database/sql driver and more backends

Releasing

Conventional commits on main; release-please opens the release PR and merging it cuts the tag, mirrored onto driver/azyncpgx/vX.Y.Z. Never edit .release-please-manifest.json or CHANGELOG.md by hand — that PR owns them.

While the line is 0.0.x, a commit carrying a BREAKING CHANGE: footer must also carry a Release-As: footer naming the intended version:

BREAKING CHANGE: driver.Store gains Foo (custom drivers must add it).

Release-As: 0.0.8

bump-minor-pre-major is not set in release-please-config.json and its default is false, so release-please reads any breaking change on a 0.x line as 1.0.0. Release-As: is what pins it. The API is still moving and nearly every release so far has been breaking, so the footer is the norm here, not an exception — and omitting it does not fail anything, it silently proposes a major.

If a release PR appears with the wrong version, land another commit on main carrying the right Release-As:; release-please rebuilds the PR from it. The tag is only cut when that PR merges, so nothing is published in the meantime.

License

MIT

Documentation ¶

Overview ¶

Package azync provides durable background jobs and a CQRS event bus for Go, unified over a single job table with pluggable storage drivers.

A Core is the shared root: it owns the storage driver, the resolved layered defaults and the logger. Open builds one from a DSN, resolving the driver from its scheme through a registry populated by a blank import (in the style of database/sql, see RegisterDriver); New wraps an already-constructed driver.Store directly. Neither migrates automatically — call Core.Migrate once the driver supports it.

The queue and event runtimes each compose over a Core: their New shares one (so jobs and event deliveries live behind a single connection pool, schema and migrations table), or their own Open builds a private one. Every runtime setting resolves in layers — a runtime-specific option overrides a Core option, which overrides the built-in Defaults — so a queue- or event-only override never has to touch the shared Core.

Index ¶

func RegisterDriver(scheme string, opener driver.Opener)
type Core
- func New(store driver.Store, opts ...Option) (*Core, error)
- func Open(dsn string, opts ...Option) (*Core, error)
type Defaults
type Option

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func RegisterDriver ¶

func RegisterDriver(scheme string, opener driver.Opener)

RegisterDriver registers a driver.Opener under a DSN scheme, in the style of database/sql. Drivers call it from an init function so a blank import wires them in. It panics if scheme is empty, opener is nil, or the scheme is already registered.

Types ¶

type Core ¶

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

Core is the shared root: it owns the storage driver, the resolved defaults and the logger. The queue and event runtimes compose over one Core, either sharing it (queue.New(core)) or owning a private one (queue.Open(dsn)).

func New ¶

func New(store driver.Store, opts ...Option) (*Core, error)

New builds a Core over an already-constructed Store. Infrastructure options (WithSchema, WithNotifyChannel, WithMigrationsTable, PollOnly) are rejected here because the store is already built; only the logger and defaults options apply.

func Open ¶

func Open(dsn string, opts ...Option) (*Core, error)

Open resolves the driver for the DSN's scheme from the registry, builds the driver.Config from the options, and opens a Store. The DSN scheme selects the driver; register one with a blank import. Open never migrates and never includes the DSN (which may carry credentials) in an error.

func (*Core) Close ¶

func (c *Core) Close(ctx context.Context) error

Close releases the driver's resources.

func (*Core) Defaults ¶

func (c *Core) Defaults() Defaults

Defaults returns the resolved shared defaults. Runtimes read these as their baseline and may override individual values per runtime.

func (*Core) Logger ¶

func (c *Core) Logger() *slog.Logger

Logger returns the Core's structured logger (never nil).

func (*Core) Migrate ¶

func (c *Core) Migrate(ctx context.Context) error

Migrate brings the backend schema up to date. It requires a driver.Migrator; otherwise it returns an error wrapping driver.ErrNotSupported.

func (*Core) Store ¶

func (c *Core) Store() driver.Store

Store returns the underlying storage driver the runtimes operate on.

type Defaults ¶

type Defaults struct {
	// LeaseTTL is how long a worker holds a job before its lease is reclaimable.
	LeaseTTL time.Duration
	// DefaultMaxAttempts is the retry budget applied to jobs enqueued without an
	// explicit budget.
	DefaultMaxAttempts int
	// ShutdownDrain is how long Close waits for in-flight jobs to settle.
	ShutdownDrain time.Duration
	// MaxConcurrency caps the total concurrent handlers across a runtime.
	MaxConcurrency int
	// DefaultConcurrency is the per-kind handler concurrency when unset.
	DefaultConcurrency int
	// FetchBatchSize is how many jobs one dequeue leases at a time.
	FetchBatchSize int
	// FetchPollInterval is the polling period when no wakeups arrive.
	FetchPollInterval time.Duration
	// FetchCooldown is the pause after a full batch before fetching again.
	FetchCooldown time.Duration
	// IdleBackoffMax caps the backoff a fetch loop reaches while idle.
	IdleBackoffMax time.Duration
	// MaxReaps is how many lease expirations a job survives before it is killed.
	MaxReaps int
	// StatsRetention is how long daily stat counters are kept; 0 keeps them
	// forever.
	StatsRetention time.Duration
	// CompletedRetention is how long succeeded jobs are kept; 0 keeps them
	// forever.
	CompletedRetention time.Duration
	// DeadRetention is how long dead (exhausted-retry) jobs are kept; 0 keeps
	// them forever. Unlike CompletedRetention, dead jobs are diagnostic
	// history an operator may want to inspect indefinitely, so the default is
	// conservative: opt in explicitly to automatic removal.
	DeadRetention time.Duration
}

Defaults are the shared baseline settings a Core resolves from options. Each value is a starting point the queue and event runtimes may override per runtime (package option > core option > default).

type Option ¶

type Option func(*coreConfig) error

Option configures a Core. Options compose; later options win.

func PollOnly ¶

func PollOnly() Option

PollOnly disables push wakeups, forcing the always-correct polling path. Infrastructure option: valid only with Open.

func WithCompletedRetention ¶

func WithCompletedRetention(d time.Duration) Option

WithCompletedRetention sets Defaults.CompletedRetention. A negative value is rejected; zero means retain succeeded jobs forever.

func WithDeadRetention ¶ added in v0.0.4

func WithDeadRetention(d time.Duration) Option

WithDeadRetention sets Defaults.DeadRetention. A negative value is rejected; zero (the default) means retain dead jobs forever.

func WithDefaultConcurrency ¶

func WithDefaultConcurrency(n int) Option

WithDefaultConcurrency sets Defaults.DefaultConcurrency. Must be positive.

func WithDefaultMaxAttempts ¶

func WithDefaultMaxAttempts(n int) Option

WithDefaultMaxAttempts sets Defaults.DefaultMaxAttempts. Must be positive.

func WithFetchBatchSize ¶

func WithFetchBatchSize(n int) Option

WithFetchBatchSize sets Defaults.FetchBatchSize. Must be positive.

func WithFetchCooldown ¶

func WithFetchCooldown(d time.Duration) Option

WithFetchCooldown sets Defaults.FetchCooldown. Must be positive.

func WithFetchPollInterval ¶

func WithFetchPollInterval(d time.Duration) Option

WithFetchPollInterval sets Defaults.FetchPollInterval. Must be positive.

func WithIdleBackoffMax ¶

func WithIdleBackoffMax(d time.Duration) Option

WithIdleBackoffMax sets Defaults.IdleBackoffMax. Must be positive.

func WithLeaseTTL ¶

func WithLeaseTTL(d time.Duration) Option

WithLeaseTTL sets Defaults.LeaseTTL. Must be positive.

func WithLogger ¶

func WithLogger(logger *slog.Logger) Option

WithLogger sets the Core's structured logger. A nil logger is rejected.

func WithMaxConcurrency ¶

func WithMaxConcurrency(n int) Option

WithMaxConcurrency sets Defaults.MaxConcurrency. Must be positive.

func WithMaxReaps ¶

func WithMaxReaps(n int) Option

WithMaxReaps sets Defaults.MaxReaps. Must be positive.

func WithMigrationsTable ¶

func WithMigrationsTable(table string) Option

WithMigrationsTable overrides the migration version-tracking table name (default azync_migrations in the pg driver). Infrastructure option: valid only with Open.

func WithNotifyChannel ¶

func WithNotifyChannel(channel string) Option

WithNotifyChannel sets the driver's wakeup channel name. Infrastructure option: valid only with Open.

func WithSchema ¶

func WithSchema(schema string) Option

WithSchema isolates azync's tables in the named backend schema (empty uses the backend default). The name is validated as an identifier. Infrastructure option: valid only with Open.

func WithShutdownDrain ¶

func WithShutdownDrain(d time.Duration) Option

WithShutdownDrain sets Defaults.ShutdownDrain. Must be positive.

func WithStatsRetention ¶

func WithStatsRetention(d time.Duration) Option

WithStatsRetention sets Defaults.StatsRetention. A negative value is rejected; zero means retain stat counters forever.

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
dag Package dag provides durable static DAGs over an azync Core: a task graph declared up front, executed by ordinary job machinery, with durable timers, signals, task results, compensation and a per-DAG failure policy.	Package dag provides durable static DAGs over an azync Core: a task graph declared up front, executed by ordinary job machinery, with durable timers, signals, task results, compensation and a per-DAG failure policy.
driver Package driver defines the backend-agnostic contract that azync storage drivers implement.	Package driver defines the backend-agnostic contract that azync storage drivers implement.
drivertest Package drivertest provides a public conformance suite that any azync storage driver can run against its own driver.Store to prove it honors the backend-agnostic contract.	Package drivertest provides a public conformance suite that any azync storage driver can run against its own driver.Store to prove it honors the backend-agnostic contract.
azyncpgx module
event Package event is a durable CQRS event bus over an azync Core.	Package event is a durable CQRS event bus over an azync Core.
eventtest Package eventtest provides an in-memory publisher seam for tests: a Recorder that satisfies the same Publish signature as the real event.Publisher, so application code under test can publish without a running runtime and the test can assert on what was published.	Package eventtest provides an in-memory publisher seam for tests: a Recorder that satisfies the same Publish signature as the real event.Publisher, so application code under test can publish without a running runtime and the test can assert on what was published.
internal
clock Package clock provides a minimal injectable time source so runtimes and the in-memory test store can be driven by a controllable clock in tests while using the real wall clock in production.	Package clock provides a minimal injectable time source so runtimes and the in-memory test store can be driven by a controllable clock in tests while using the real wall clock in production.
drivertest Package drivertest provides an in-memory driver.Store used by the queue and event runtimes' unit tests.	Package drivertest provides an in-memory driver.Store used by the queue and event runtimes' unit tests.
engine Package engine is the shared fetch/execute/settle/maintenance machinery the queue and event runtimes are built on, neutral over driver.Source.	Package engine is the shared fetch/execute/settle/maintenance machinery the queue and event runtimes are built on, neutral over driver.Source.
queue Package queue provides durable background jobs over an azync Core.	Package queue provides durable background jobs over an azync Core.
watch 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.	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.
workflow Package workflow is the workflow-as-code (WAC) runtime: workflows and Operations defined as ordinary Go functions, executed by deterministic replay over an append-only history.	Package workflow is the workflow-as-code (WAC) runtime: workflows and Operations defined as ordinary Go functions, executed by deterministic replay over an append-only history.
kernel Package kernel is the pure in-memory history/command/replay engine for workflow-as-code.	Package kernel is the pure in-memory history/command/replay engine for workflow-as-code.

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL