FIFO serializer — single-writer queue for resources that don't tolerate concurrent writers (konserve filestore, datahike writer, external services with strict request ordering).
gate (hive-weave.gate) is the right choice when the
caller needs the result back synchronously and you can afford
to block on the permit. Concurrent reads, bounded fan-out, etc.pool (hive-weave.pool) is the right choice when work
items are independent and ordering doesn't matter..ksv.new -> .ksv rename race; SQLite WAL on a single file; an HTTP API
with strict request ordering).Two closed ADTs make the message and outcome surface explicit:
SerializerMsg — what flows through the queue :msg/task { :key, :f, :promise } :msg/poison
SubmitOutcome — what submit! returns
:submit/ok { :promise }
:submit/timeout { :queue-size :timeout-ms }
:submit/closed
TaskOutcome — what the submission promise resolves to :task/ok { :value } :task/failed { :class :message }
Domain ADTs make the worker loop a adt-case exhaustive match —
no string-typing, no half-cases.
The serializer holds a single dedicated worker thread + a bounded
LinkedBlockingQueue. submit! enqueues; if the queue is full,
the submitter blocks up to :submit-timeout-ms waiting for
space (true backpressure — the caller can't outrun the worker).
When :coalesce-key-fn is provided (or a :key is passed to
submit!), every submission carries a key. Before enqueuing,
the queue is scanned for a pending task with the same key; if
found, the pending task is replaced by the new one. Coalescing
is opt-in.
serializer returns a record. close! shuts down the worker
gracefully (drains the queue then exits via a :msg/poison
message). Repeated close! calls are no-ops. After close,
submit! returns (submit-outcome :submit/closed).
FIFO serializer — single-writer queue for resources that don't
tolerate concurrent writers (konserve filestore, datahike writer,
external services with strict request ordering).
## When to reach for this
- **A `gate`** (`hive-weave.gate`) is the right choice when the
caller needs the result back synchronously and you can afford
to block on the permit. Concurrent reads, bounded fan-out, etc.
- **A `pool`** (`hive-weave.pool`) is the right choice when work
items are independent and ordering doesn't matter.
- **A serializer is the right choice when**:
1. Concurrent execution corrupts the resource (konserve `.ksv.new
-> .ksv` rename race; SQLite WAL on a single file; an HTTP API
with strict request ordering).
2. Callers don't need the result synchronously — they fire-and-
forget and check completion via a side channel (a promise the
submitter can deref later, or just log on failure).
3. Ordering matters (FIFO).
4. Repeated submissions for the same key can be coalesced (only
the latest matters — e.g. an upsert that's idempotent on a
unique key, where an in-flight submission about to be
overwritten can be dropped).
## Type model
Two closed ADTs make the message and outcome surface explicit:
SerializerMsg — what flows through the queue
:msg/task { :key, :f, :promise }
:msg/poison
SubmitOutcome — what `submit!` returns
:submit/ok { :promise }
:submit/timeout { :queue-size :timeout-ms }
:submit/closed
TaskOutcome — what the submission promise resolves to
:task/ok { :value }
:task/failed { :class :message }
Domain ADTs make the worker loop a `adt-case` exhaustive match —
no string-typing, no half-cases.
## Backpressure model
The serializer holds a single dedicated worker thread + a bounded
`LinkedBlockingQueue`. `submit!` enqueues; if the queue is full,
the submitter **blocks up to `:submit-timeout-ms`** waiting for
space (true backpressure — the caller can't outrun the worker).
## Coalescing
When `:coalesce-key-fn` is provided (or a `:key` is passed to
`submit!`), every submission carries a key. Before enqueuing,
the queue is scanned for a pending task with the same key; if
found, the pending task is **replaced** by the new one. Coalescing
is opt-in.
## Lifecycle
`serializer` returns a record. `close!` shuts down the worker
gracefully (drains the queue then exits via a `:msg/poison`
message). Repeated `close!` calls are no-ops. After close,
`submit!` returns `(submit-outcome :submit/closed)`.(->serializer-msg kw__16660__auto__)Coerce a keyword to a SerializerMsg variant (no data fields). Returns nil if keyword is not a valid variant.
Coerce a keyword to a SerializerMsg variant (no data fields). Returns nil if keyword is not a valid variant.
(->submit-outcome kw__16660__auto__)Coerce a keyword to a SubmitOutcome variant (no data fields). Returns nil if keyword is not a valid variant.
Coerce a keyword to a SubmitOutcome variant (no data fields). Returns nil if keyword is not a valid variant.
(->task-outcome kw__16660__auto__)Coerce a keyword to a TaskOutcome variant (no data fields). Returns nil if keyword is not a valid variant.
Coerce a keyword to a TaskOutcome variant (no data fields). Returns nil if keyword is not a valid variant.
(close! s)Drain the queue, then shut down the worker. Idempotent. After
close, submit! returns (submit-outcome :submit/closed).
Returns Result with the final stats.
Drain the queue, then shut down the worker. Idempotent. After close, `submit!` returns `(submit-outcome :submit/closed)`. Returns Result with the final stats.
(serializer {:keys [name queue-capacity submit-timeout-ms coalesce-key-fn]
:or
{name "serializer" queue-capacity 256 submit-timeout-ms 30000}})Create a FIFO serializer.
Options:
:name — diagnostic name (default "serializer")
:queue-capacity — bounded queue depth (default 256). Submitters
block when full.
:submit-timeout-ms — max time submit! will wait for queue space
before giving up (default 30000).
:coalesce-key-fn — (fn [submit-opts]) returning a coalescing
key (or nil). When supplied, a queued task
with the same key is replaced by an incoming
submission. Set to nil to disable
coalescing (default).
Create a FIFO serializer.
Options:
:name — diagnostic name (default "serializer")
:queue-capacity — bounded queue depth (default 256). Submitters
block when full.
:submit-timeout-ms — max time `submit!` will wait for queue space
before giving up (default 30000).
:coalesce-key-fn — `(fn [submit-opts])` returning a coalescing
key (or nil). When supplied, a queued task
with the same key is replaced by an incoming
submission. Set to `nil` to disable
coalescing (default).(serializer-msg variant-kw__16657__auto__)(serializer-msg variant-kw__16657__auto__ data__16658__auto__)Construct a SerializerMsg variant. (serializer-msg variant-keyword) for enum variants (serializer-msg variant-keyword data-map) for data variants
Throws ex-info for unknown variants.
Construct a SerializerMsg variant. (serializer-msg variant-keyword) for enum variants (serializer-msg variant-keyword data-map) for data variants Throws ex-info for unknown variants.
(serializer-msg? x__16659__auto__)True if x is a SerializerMsg ADT value.
True if x is a SerializerMsg ADT value.
Messages flowing through a serializer's work queue.
:msg/task — work envelope: caller's fn + the promise to fill :msg/poison — shutdown sentinel; worker drains then exits
Messages flowing through a serializer's work queue. :msg/task — work envelope: caller's fn + the promise to fill :msg/poison — shutdown sentinel; worker drains then exits
(stats s)Current serializer state for observability.
Current serializer state for observability.
(submit! s f)(submit! s {:keys [key]} f)Enqueue f for serialized execution. Returns a SubmitOutcome:
:submit/ok — accepted; @(:promise outcome) resolves to a
TaskOutcome
:submit/timeout — queue full for longer than :submit-timeout-ms
:submit/closed — close! has been called
Submission may block up to :submit-timeout-ms if the queue
is full (true backpressure — caller can't outrun the worker).
:key enables coalescing — a pending task with the same key is
replaced before enqueue. Useful for repeated-state-write workloads
where the upsert is idempotent on a unique key.
Usage — fire-and-forget with side-channel logging:
(let [outcome (submit! s {:key project-id} (fn [] (persist! state)))] (when (= :submit/timeout (:adt/variant outcome)) (log/warn "serializer rejected" outcome)))
Enqueue `f` for serialized execution. Returns a `SubmitOutcome`:
:submit/ok — accepted; @(:promise outcome) resolves to a
`TaskOutcome`
:submit/timeout — queue full for longer than :submit-timeout-ms
:submit/closed — close! has been called
Submission may **block** up to `:submit-timeout-ms` if the queue
is full (true backpressure — caller can't outrun the worker).
`:key` enables coalescing — a pending task with the same key is
replaced before enqueue. Useful for repeated-state-write workloads
where the upsert is idempotent on a unique key.
Usage — fire-and-forget with side-channel logging:
(let [outcome (submit! s {:key project-id} (fn [] (persist! state)))]
(when (= :submit/timeout (:adt/variant outcome))
(log/warn "serializer rejected" outcome)))(submit-and-wait! s f)(submit-and-wait! s opts f)(submit-and-wait! s opts f wait-timeout-ms)Submit and block on the result promise. Returns a TaskOutcome
on success, or a SubmitOutcome describing the rejection.
:wait-timeout-ms defaults to twice :submit-timeout-ms.
On wait-timeout: (task-outcome :task/failed {...}) with class
:weave.serializer/wait-timeout.
Submit and block on the result promise. Returns a `TaskOutcome`
on success, or a `SubmitOutcome` describing the rejection.
`:wait-timeout-ms` defaults to twice `:submit-timeout-ms`.
On wait-timeout: `(task-outcome :task/failed {...})` with class
:weave.serializer/wait-timeout.(submit-ok? outcome)True when a SubmitOutcome is :submit/ok.
True when a SubmitOutcome is `:submit/ok`.
(submit-outcome variant-kw__16657__auto__)(submit-outcome variant-kw__16657__auto__ data__16658__auto__)Construct a SubmitOutcome variant. (submit-outcome variant-keyword) for enum variants (submit-outcome variant-keyword data-map) for data variants
Throws ex-info for unknown variants.
Construct a SubmitOutcome variant. (submit-outcome variant-keyword) for enum variants (submit-outcome variant-keyword data-map) for data variants Throws ex-info for unknown variants.
(submit-outcome? x__16659__auto__)True if x is a SubmitOutcome ADT value.
True if x is a SubmitOutcome ADT value.
What submit! returns. Wraps the submission promise on success;
carries diagnostic data on rejection.
What `submit!` returns. Wraps the submission promise on success; carries diagnostic data on rejection.
(task-ok? outcome)True when a TaskOutcome is :task/ok.
True when a TaskOutcome is `:task/ok`.
(task-outcome variant-kw__16657__auto__)(task-outcome variant-kw__16657__auto__ data__16658__auto__)Construct a TaskOutcome variant. (task-outcome variant-keyword) for enum variants (task-outcome variant-keyword data-map) for data variants
Throws ex-info for unknown variants.
Construct a TaskOutcome variant. (task-outcome variant-keyword) for enum variants (task-outcome variant-keyword data-map) for data variants Throws ex-info for unknown variants.
(task-outcome? x__16659__auto__)True if x is a TaskOutcome ADT value.
True if x is a TaskOutcome ADT value.
What the submission promise resolves to once the worker runs the
task. :task/ok carries whatever f returned; :task/failed
describes the throwable.
What the submission promise resolves to once the worker runs the task. `:task/ok` carries whatever `f` returned; `:task/failed` describes the throwable.
cljdoc builds & hosts documentation for Clojure/Script libraries
| Ctrl+k | Jump to recent docs |
| ← | Move to previous article |
| → | Move to next article |
| Ctrl+/ | Jump to the search field |