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. Throw or return falsey on failure so
the op is re-queued for the next pass. 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).
Returns {:result :drained|:circuit-reopened|:all-failed|:already-running :attempted N :succeeded N :failed 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. Throw or return falsey on failure so
the op is re-queued for the next pass. 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).
Returns `{:result :drained|:circuit-reopened|:all-failed|:already-running
:attempted N :succeeded N :failed 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.
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 |