Documentation
¶
Overview ¶
Package asyncqueue is packtrail's durable asynchronous Invoker: it makes any ordinary synchronous Invoker run off the engine's critical path with at-least-once delivery. The Invoker contract's StatusPending + Server.CompleteActivity is the seam — asyncqueue supplies the machinery the engine deliberately leaves out, so embedding apps with slow nodes (an agent call, an HTTP request, a shell-exec) do not block an engine slot.
It has two halves:
- Dispatcher is the parking side. It implements invoker.Invoker: instead of running the node inline it publishes a durable job to a JetStream work-queue and returns StatusPending, so the engine parks the execution (waiting) and frees its slot.
- Worker is the execution side. It consumes those jobs, runs the embedder's own synchronous Invoker, and settles the result via Completer (CompleteActivity). A job is acked only after the completion is handed back, so a crashed worker's job is redelivered (at-least-once).
It is ecosystem-agnostic: the executor is a plain invoker.Invoker, so asyncqueue knows nothing about what the slow work actually is. Like invoker/natstask it depends on nothing beyond NATS.
Index ¶
- func EnsureStream(ctx context.Context, js jetstream.JetStream, prefix, kind string, ...) error
- func StreamName(prefix, kind string) string
- func Subject(prefix, kind string) string
- type Completer
- type DeadLetterSink
- type Dispatcher
- type Option
- func WithAckWait(d time.Duration) Option
- func WithActivityTimeout(d time.Duration) Option
- func WithConcurrency(n int) Option
- func WithDeadLetterSink(sink DeadLetterSink) Option
- func WithDedupWindow(d time.Duration) Option
- func WithDrainTimeout(d time.Duration) Option
- func WithMaxDeliver(n int) Option
- func WithMaxQueuedBytes(n int64) Option
- func WithMaxQueuedJobs(n int64) Option
- type Worker
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func EnsureStream ¶
func EnsureStream(ctx context.Context, js jetstream.JetStream, prefix, kind string, opts ...Option) error
EnsureStream creates (idempotently) the JetStream work-queue stream that carries jobs for kind, with a dedup window so a redelivered dispatch does not double-enqueue. Call it once before publishing or consuming.
func StreamName ¶
StreamName derives the work-queue stream name for an async invoker kind.
Types ¶
type Completer ¶
type Completer interface {
CompleteActivity(ctx context.Context, execID, node string, attempt int, res invoker.Result) error
}
Completer settles an asynchronous activity back on the packtrail engine. *packtrail.Server satisfies it; workers use its generation-aware completion method when the concrete completer exposes one.
type DeadLetterSink ¶
DeadLetterSink records a job a Worker gave up on (Term'd). key is "<exec>/<node>", reason is the completion error, deliveries is the message's delivery count.
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher implements invoker.Invoker by publishing a durable job and reporting it as pending, so the engine parks the execution and frees its slot for the node's whole runtime. The job's dedup id is exec.node.generation.attempt, so a redelivered dispatch of the same attempt is collapsed; a genuine retry, Resume, or legal cycle revisit is not.
func NewDispatcher ¶
func NewDispatcher(js jetstream.JetStream, prefix, kind string) *Dispatcher
NewDispatcher returns a Dispatcher publishing to the job subject for kind.
type Option ¶
type Option func(*config)
Option configures a Worker and/or the work-queue stream. The same options are accepted by EnsureStream, NewWorker and packtrail's WithAsyncInvoker.
func WithAckWait ¶
WithAckWait sets the job ack window, extended by heartbeats while a job runs (default 30s). A worker that dies mid-job has its job redelivered after this.
func WithActivityTimeout ¶
WithActivityTimeout sets the ceiling/backstop for each invocation the Worker runs (default 5m). A node's own per-call timeout tightens this when shorter; the effective bound is min(node timeout, activityTimeout). A node timeout longer than this ceiling is capped at it, so raise this when nodes legitimately need longer calls. The ack window is extended by heartbeats for the whole duration.
func WithConcurrency ¶
WithConcurrency caps how many jobs a Worker runs at once (default 64).
func WithDeadLetterSink ¶
func WithDeadLetterSink(sink DeadLetterSink) Option
WithDeadLetterSink registers a callback invoked just before a Worker Terms a job it gave up on (a terminal completion error or an exhausted delivery cap), so the caller can record a durable trace of the dropped job. packtrail wires this to the dead-letter stream automatically.
func WithDedupWindow ¶
WithDedupWindow sets the JetStream dedup window on the work-queue stream (default 2m). It only affects EnsureStream.
func WithDrainTimeout ¶
WithDrainTimeout sets how long a graceful Worker shutdown waits for in-flight jobs to settle before aborting the stragglers (default 30s). Within the window a clean shutdown lets running invocations finish and settle via CompleteActivity instead of abandoning them to redelivery.
func WithMaxDeliver ¶
WithMaxDeliver caps how many times a job is delivered before the Worker dead-letters it (Term, no redelivery) rather than Nak-looping on a persistent completion failure (default 10). A terminal completion error (e.g. the engine no longer knows the flow) is dead-lettered immediately, regardless of this cap. A non-positive value disables the cap (only terminal errors dead-letter).
func WithMaxQueuedBytes ¶
WithMaxQueuedBytes caps the total bytes retained in the durable work queue (default 1GiB). When full, new distinct jobs are rejected at publish time so the engine can retry dispatch instead of silently growing infrastructure storage without bound. Non-positive values keep the bounded default.
func WithMaxQueuedJobs ¶
WithMaxQueuedJobs caps the number of jobs retained in the durable work queue (default 10,000). When full, new distinct jobs are rejected at publish time so the engine can retry dispatch instead of silently growing infrastructure storage without bound. Non-positive values keep the bounded default.
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker consumes jobs for one async invoker kind and runs the embedder's synchronous Invoker off the engine's critical path, reporting completion via Completer. It is the durability boundary: a job is acked only after the completion is handed back, so a crashed worker's job is redelivered (at-least-once). Many Workers (in or out of process) can share a kind's work-queue stream to scale horizontally.
func NewWorker ¶
func NewWorker( js jetstream.JetStream, prefix, kind string, exec invoker.Invoker, completer Completer, opts ...Option, ) *Worker
NewWorker builds a Worker that runs exec for jobs of kind and settles them via completer. prefix and kind must match the Dispatcher's. EnsureStream must have been called for the same kind (packtrail's WithAsyncInvoker does both).
Source Files