coordinator: multi-channel priority queue + stream affinity + multi-subscription consume - #2
Open
simonhachey wants to merge 5 commits into
Open
simonhachey wants to merge 5 commits into
simonhachey wants to merge 5 commits into
Conversation
…ware recv_for Purely additive priority-queue plumbing for long-running queues Task 1: - Channel enum (Sync/Async/NotifSteady/NotifBatch) replaces Priority - ServedQueue enum + from_str (unknown/"" -> Fast, legacy default) - PriorityQueueSender::send(action, Channel), len(Channel), send_notif_steady/send_notif_batch - PriorityQueueReceiver::recv_for(ServedQueue, metrics) with affinity-aware dispatch; recv() now delegates to recv_for(Fast, ..), preserving the existing biased sync>async behavior and async-queue-time metric recording No subscription routing into the new channels yet (later task). Existing send_sync/send_async/recv/len_sync/len_async callers are unaffected.
simonhachey
marked this pull request as ready for review
July 22, 2026 01:11
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds workload-isolation plumbing to the coordinator so long-running, IO-bound UDFs (e.g. LLM classification, on the order of seconds per call) can be processed on dedicated worker pools without head-of-line-blocking the fast path. Dormant and backwards-compatible by default — the existing Kafka default path and the single-
PUBSUB_SUBSCRIPTION_IDpubsub path are unchanged (therecv→recv_for(ServedQueue::Fast)refactor is the same biased[sync, async]select as before).What's in it
priority_queue.rs): addsnotif-steadyandnotif-batchchannels alongsidesync/async, aChannelenum, and aServedQueueenum. A new affinity-awarerecv_for(ServedQueue)lets a connection pull only from the channel(s) its pool serves;ServedQueue::Fast(the default) preserves the existing[sync, async]biased select exactly (sync first, async→sync fallback, unchanged queue-time metric).main.rs,consumer/pubsub.rs): the coordinator can now consume several Pub/Sub subscriptions — one pull manager each, its ownFlowControlbudget — routing each subscription's actions into a specific channel. Configured viaOSPREY_COORDINATOR_SUBSCRIPTIONS(asub_id:channellist). Unset ⇒ falls back to the singlePUBSUB_SUBSCRIPTION_ID→asyncchannel (today's behavior).sync/async.Dormant by design
Nothing is activated in this PR: with
OSPREY_COORDINATOR_SUBSCRIPTIONSunset the coordinator runs its single-subscription async path exactly as before, and no worker advertises aserved_queueyet — so the new channels are never fed or drained. The dispatch-side affinity (workers advertising which channel they serve) and the dedicated worker pools land in follow-up PRs.One intentional behavior change: a
pubsubcoordinator started with neitherOSPREY_COORDINATOR_SUBSCRIPTIONSnorPUBSUB_SUBSCRIPTION_IDset now fails fast at startup instead of silently subscribing to a hardcoded default name. In practice both real configs are unaffected (Kafka default is untouched; the pubsub path requiresPUBSUB_SUBSCRIPTION_ID), but it's a deliberate fail-on-misconfig, not byte-for-byte.Testing
cargo test— 60/60 (new: channel/affinityrecv_for,OSPREY_COORDINATOR_SUBSCRIPTIONSparsing incl. error paths, notif-channel shutdown nacks).cargo checkclean.Follow-ups (separate PRs)
ClientDetails.served_queue+ affinity-aware dispatch + worker-side advertising.🤖 Generated with Claude Code