Global in-memory write queue + read fail-soft shim for Milvus outages.
Single JVM-wide singleton — shared across ALL projects and agents, not
per-project. Purpose: when the circuit breaker opens because Milvus is
unreachable, mutating protocol calls are enqueued here instead of
failing; reads return a degraded response carrying actionable recovery
tips. On the circuit's :closed transition, a background future drains
the queue via hive-weave.parallel/bounded-pmap with concurrency 8
and a 15 s per-item timeout. Pre-drain the batch is coalesced by
(op,id) keeping the latest mutation, so a burst of writes to the same
entry collapses to one RPC. Any operation that still fails is pushed
to the tail for the next pass; drain repeats until the queue is empty,
the circuit reopens, or a pass makes zero progress.
Intentionally has no hard dependency on the circuit ns or store.clj —
those wire it in (see enqueue!, drain!, degraded-response,
make-store-dispatch). Keeping the boundary clean lets circuit,
store, and queue evolve independently and lets tests exercise the
queue without standing up Milvus.
Global in-memory write queue + read fail-soft shim for Milvus outages. Single JVM-wide singleton — shared across ALL projects and agents, not per-project. Purpose: when the circuit breaker opens because Milvus is unreachable, mutating protocol calls are enqueued here instead of failing; reads return a degraded response carrying actionable recovery tips. On the circuit's :closed transition, a background future drains the queue via `hive-weave.parallel/bounded-pmap` with concurrency 8 and a 15 s per-item timeout. Pre-drain the batch is coalesced by (op,id) keeping the latest mutation, so a burst of writes to the same entry collapses to one RPC. Any operation that still fails is pushed to the tail for the next pass; drain repeats until the queue is empty, the circuit reopens, or a pass makes zero progress. Intentionally has no hard dependency on the circuit ns or store.clj — those wire it in (see `enqueue!`, `drain!`, `degraded-response`, `make-store-dispatch`). Keeping the boundary clean lets circuit, store, and queue evolve independently and lets tests exercise the queue without standing up Milvus.
(clear!)Drop every queued op. Test helper + escape hatch — never call in production without understanding what you're throwing away.
Drop every queued op. Test helper + escape hatch — never call in production without understanding what you're throwing away.
(coalesce ops)Reduce a batch of ops to the latest mutation per (op, id), preserving first-seen order so the drain respects arrival sequence. Public so tests can exercise it directly.
Example: [{:op :add-entry! :id "a" :args [{...}]} {:op :update-entry! :id "b" :args ["b" {...}]} {:op :add-entry! :id "a" :args [{...updated}]}] => collapses the two writes to 'a' into the second one, keeps 'b'.
Reduce a batch of ops to the latest mutation per (op, id), preserving
first-seen order so the drain respects arrival sequence. Public so
tests can exercise it directly.
Example:
[{:op :add-entry! :id "a" :args [{...}]}
{:op :update-entry! :id "b" :args ["b" {...}]}
{:op :add-entry! :id "a" :args [{...updated}]}]
=> collapses the two writes to 'a' into the second one, keeps 'b'.(degraded-response)(degraded-response {:keys [retry-after-ms reconnect-eta-ms operation extra-tips]
:or {retry-after-ms 5000}})Build a read fail-soft response map used by store.clj read methods when the circuit is open. Carries actionable tips instead of an opaque error — callers (agents, MCP tools) can surface retry-after hints and the pending-write depth so the human sees what's pending.
Options: :retry-after-ms Hint for when to retry (default 5000). :reconnect-eta-ms Best-guess ETA from circuit state (nil = unknown). :operation Keyword naming the read that degraded, for tips. :extra-tips Extra hints to append after the standard tips.
Build a read fail-soft response map used by store.clj read methods when the circuit is open. Carries actionable tips instead of an opaque error — callers (agents, MCP tools) can surface retry-after hints and the pending-write depth so the human sees what's pending. Options: :retry-after-ms Hint for when to retry (default 5000). :reconnect-eta-ms Best-guess ETA from circuit state (nil = unknown). :operation Keyword naming the read that degraded, for tips. :extra-tips Extra hints to append after the standard tips.
(drain! {:keys [dispatch-fn circuit-open?]
:or {circuit-open? (constantly false)}})Drain the queue until it's empty, the circuit reopens, or a pass makes zero forward progress (safety against a flapping backend). Intended to be called from a future spawned on the circuit's :closed transition — this function runs drain passes inline in the caller's thread and will block until one of its stop conditions triggers.
Options (map):
:dispatch-fn REQUIRED. (fn [op] ...) that applies one queued op
to Milvus. Signal failure by throwing OR by returning
a write-failure? value. write-outcome splits it:
a TRANSIENT refusal (:reconnecting? true - a spent
resilient budget, an open circuit) or a throw is
re-queued for the next pass; a PERMANENT one (a fatal
failure map, update-entry!'s routing/schema
{:error ...}) is logged and dropped, since retrying
cannot change it. Any other return, nil included
(update of an unknown id), counts as done. Typically
built via make-store-dispatch.
:circuit-open? (fn []) predicate. If it returns truthy between
passes, drain stops and leaves the rest queued.
Default: never-open (best-effort).
A pass makes progress when some op landed or was dropped.
Returns {:result :drained|:circuit-reopened|:all-failed|:already-running :attempted N :succeeded N :failed N :dropped N :passes N}.
Drain the queue until it's empty, the circuit reopens, or a pass
makes zero forward progress (safety against a flapping backend).
Intended to be called from a future spawned on the circuit's
:closed transition — this function runs drain passes inline in the
caller's thread and will block until one of its stop conditions
triggers.
Options (map):
:dispatch-fn REQUIRED. (fn [op] ...) that applies one queued op
to Milvus. Signal failure by throwing OR by returning
a `write-failure?` value. `write-outcome` splits it:
a TRANSIENT refusal (`:reconnecting? true` - a spent
`resilient` budget, an open circuit) or a throw is
re-queued for the next pass; a PERMANENT one (a fatal
failure map, `update-entry!`'s routing/schema
`{:error ...}`) is logged and dropped, since retrying
cannot change it. Any other return, nil included
(update of an unknown id), counts as done. Typically
built via `make-store-dispatch`.
:circuit-open? (fn []) predicate. If it returns truthy between
passes, drain stops and leaves the rest queued.
Default: never-open (best-effort).
A pass makes progress when some op landed or was dropped.
Returns `{:result :drained|:circuit-reopened|:all-failed|:already-running
:attempted N :succeeded N :failed N :dropped N :passes N}`.bounded-pmap worker count. Matches a typical Milvus collection's sweet spot for mixed upserts without saturating the gRPC channel.
bounded-pmap worker count. Matches a typical Milvus collection's sweet spot for mixed upserts without saturating the gRPC channel.
(drain-running?)Whether a drain pass is currently in flight.
Whether a drain pass is currently in flight.
Per-item timeout. Generous because upserts do embedding + insert + index maintenance; stingy enough that one stuck RPC doesn't wedge the drain.
Per-item timeout. Generous because upserts do embedding + insert + index maintenance; stingy enough that one stuck RPC doesn't wedge the drain.
(enqueue! op)Append an op descriptor to the queue. An op is a map of shape:
{:op :add-entry! ;; protocol method keyword :id "entry-uuid" ;; nil for singleton ops like :cleanup-expired! :args [entry]} ;; args to splat into the protocol call
Returns {:success? true :queued? true :depth N :tips [...]} on
success, or {:success? false :error :queue-full :depth N :tips [...]}
when the soft cap is hit. Uses swap-vals! so we can atomically
detect whether the swap added an element — no dropped writes under
contention, no false overflows.
Append an op descriptor to the queue. An op is a map of shape:
{:op :add-entry! ;; protocol method keyword
:id "entry-uuid" ;; nil for singleton ops like :cleanup-expired!
:args [entry]} ;; args to splat into the protocol call
Returns `{:success? true :queued? true :depth N :tips [...]}` on
success, or `{:success? false :error :queue-full :depth N :tips [...]}`
when the soft cap is hit. Uses `swap-vals!` so we can atomically
detect whether the swap added an element — no dropped writes under
contention, no false overflows.(full?)Whether the next enqueue will be rejected. Callers should prefer
enqueue! and check its result — this is for status/metrics.
Whether the next enqueue will be rejected. Callers should prefer `enqueue!` and check its result — this is for status/metrics.
(make-store-dispatch store)Build a drain dispatch-fn bound to a concrete MilvusMemoryStore.
Each queued op becomes a direct proto/* call against the store.
The circuit wiring owns the question of whether these calls go
through resilient (risking re-queueing loops) or a bypass path;
this helper just dispatches on :op. On unknown :op, throws —
the caller's drain-pass! catches and logs it as a failure.
Keeping this in the queue ns (rather than store.clj) means the store doesn't need to know the queue exists, and circuit wiring can swap in its own dispatch-fn for testing.
Build a drain `dispatch-fn` bound to a concrete `MilvusMemoryStore`. Each queued op becomes a direct `proto/*` call against the store. The circuit wiring owns the question of whether these calls go through `resilient` (risking re-queueing loops) or a bypass path; this helper just dispatches on `:op`. On unknown `:op`, throws — the caller's `drain-pass!` catches and logs it as a failure. Keeping this in the queue ns (rather than store.clj) means the store doesn't need to know the queue exists, and circuit wiring can swap in its own dispatch-fn for testing.
Soft cap. Overflow rejects with :queue-full — we'd rather tell the caller 'dropping your write' than OOM the JVM. 100k is ~a few hours of steady-state writes at a realistic agent cadence.
Soft cap. Overflow rejects with :queue-full — we'd rather tell the caller 'dropping your write' than OOM the JVM. 100k is ~a few hours of steady-state writes at a realistic agent cadence.
(size)Current queue depth (cheap atom read).
Current queue depth (cheap atom read).
(stats)Snapshot of cumulative metrics + current depth.
Snapshot of cumulative metrics + current depth.
(write-failure? x)True when a write verb's return value is a refusal: a map whose
:success? is false or whose :error is non-nil. That is the shape
resilient answers once its retry budget is spent, and the shape the
circuit breaker answers while open.
Mirrors hive-spi.memory.contract/failure? (hive-spi f9e3025), which no
released hive-spi carries yet; switch to it when the pin can move.
True when a write verb's return value is a refusal: a map whose :success? is false or whose :error is non-nil. That is the shape `resilient` answers once its retry budget is spent, and the shape the circuit breaker answers while open. Mirrors `hive-spi.memory.contract/failure?` (hive-spi f9e3025), which no released hive-spi carries yet; switch to it when the pin can move.
(write-outcome x)Classify a write verb's return value for the drain:
:ok it landed (anything that is not a write-failure?, nil included:
an update of an unknown id is terminal)
:retry a TRANSIENT refusal - :reconnecting? true, which is what a spent
resilient budget and an open circuit answer. Re-queue it.
:drop a PERMANENT refusal - a fatal failure map (:reconnecting? false)
or an {:error ...} that does not claim to be reconnecting, e.g.
update-entry!'s routing / embedding / schema errors. Retrying
cannot change the answer, so re-queueing would only keep a poison
op in the queue forever.
Classify a write verb's return value for the drain:
:ok it landed (anything that is not a `write-failure?`, nil included:
an update of an unknown id is terminal)
:retry a TRANSIENT refusal - `:reconnecting? true`, which is what a spent
`resilient` budget and an open circuit answer. Re-queue it.
:drop a PERMANENT refusal - a fatal failure map (`:reconnecting? false`)
or an `{:error ...}` that does not claim to be reconnecting, e.g.
`update-entry!`'s routing / embedding / schema errors. Retrying
cannot change the answer, so re-queueing would only keep a poison
op in the queue forever.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 |