dj.concurrency.policy — net.clojars.bmillare/dj.concurrency 0.1.0-alpha3

Functional core for dj.concurrency: the pure reference policy that decides what should happen, with no side effects. The impure execution of those decisions lives in dj.concurrency.shell.

A policy is a pure function (fn [event state] -> {:directives [...] :state s'}). Events and directives are positional pairs [type payload-map]; every other value is a MAP so the code reads by key rather than by position.

NOTE: the qualified keywords :dj.concurrency/attempts, :dj.concurrency/max-attempts, and :dj.concurrency/shutdown are part of the public contract — consumers set :dj.concurrency/max-attempts in a task's context and match on :dj.concurrency/shutdown in ex-data. They are keyed to the dj.concurrency namespace, so they are written out in full here rather than with :: auto-resolution (which would key them to THIS namespace).

abort-requested? clj

(abort-requested? error)

True when error's ex-data carries the explicit abort marker :dj.concurrency/abort — the zero-config escape hatch a worker throws to end a task terminally (build it with dj.concurrency/abort-error). Custom :classify-error fns can call this to honor the marker in one line.

default-backoff clj

(default-backoff attempts)

Simple exponential backoff (1s, 2s, 4s...) based on attempts.

default-classify-error clj

(default-classify-error error)

Reference heuristic mapping a worker's thrown error to a verdict — :abort | :rate-limited | :transient:

• explicit {:dj.concurrency/abort true} marker -> :abort (terminal escape hatch: the task aborts and @f re-throws the error; nothing is retried or parked)
• {:status 429} -> :rate-limited (throttle the task's pool, retry when the window lifts)
• anything else -> :transient (retry with backoff, then park)

It aborts ONLY on the explicit marker: an unrecognized error is assumed recoverable and parks, so fix-forward stays the default. For systematic abort rules (e.g. all 4xx), supply your own via make-reference-policy.

default-policy clj

The reference policy built with all defaults; equals (make-reference-policy {}).

default-reference-opts clj

Defaults for the reference policy. Override any via make-reference-policy (or by passing them to create-supervisor).

Per-pool concurrency caps are deliberately NOT here: they are runtime-updatable and so live in supervisor STATE (:pool-caps {pool cap}, seeded by create-supervisor from its :pool-caps opt and written live by set-pool-cap!), not in this frozen config closure. A cap is either a positive integer or the :dj.concurrency/unbounded sentinel; in-flight is bounded PER POOL at the single admission chokepoint (scan-deadlines), a permit == a :running slot. No :pool-caps ⇒ one unbounded :default pool ⇒ today's exact behavior. See the multi-resource design doc.

make-reference-policy clj

(make-reference-policy opts)

Builds a pure reference policy (fn [event state] -> {:directives :state}), closing over opts. Recognized opts (others ignored): :classify-error (fn [error] -> :abort | :rate-limited | :transient) :backoff-fn (fn [attempts] -> backoff-ms) for transient retries :max-attempts default max attempts when a task's context omits :dj.concurrency/max-attempts (default 3) :default-throttle-ms throttle window for a 429 with no :retry-after (ms) Per-pool concurrency caps are NOT a policy opt — they live in supervisor state (:pool-caps, seeded by create-supervisor, updatable via set-pool-cap!). See default-reference-opts.

This is how supervisor-level configuration threads into the (otherwise pure, 2-arg) policy: config is baked in at construction time rather than passed on every event. create-supervisor calls this with its own opts when you don't supply an explicit :policy.

terminal-statuses clj