Faster hops, no duplicate replay - #73
Merged
Merged
Conversation
The native connector polled /receive without ever passing ack_seq, so the hub's per-client last_acked_seq stayed at zero and its 200-entry unacked ring buffer never drained. A reap followed by a revival re-injected the whole buffer as fresh inbound, duplicating the conversation. Track the highest seq of each batch and piggyback it on the next poll. An ACK a failed poll was carrying is kept pending and retried.
The driver consumed one queued batch per turn, so a busy room cost a full round-trip per batch and had the agent reason on stale context between them. Drain whatever is queued behind the first item into the same prompt, marking every drained item done so Queue.join accounting still holds.
The per-sender bucket defaulted to capacity 5, refill 0.5/s, which braked runaway loops but also stalled honest exchanges: answering two messages and joining a channel already left a peer waiting seconds for a token. Raise it to capacity 10, refill 2/s; a looping pair still converges on a visible, interruptible rate. set_status is a heartbeat, not chatter, so it no longer shares that budget. Each client gets a separate, roomier status bucket that set_rate_limit leaves alone, so clamping the room's message rate never costs the operator the liveness signal.
The /receive loop checked the priority queue at the top of each iteration and then blocked up to a second on the chatter queue, so a steer, interrupt or reset aimed at a mid-turn agent waited out that block. Race the two getters with FIRST_COMPLETED instead; the one-second slice now only bounds the checks no queue can wake the loop on (disconnect, stop, pause gates). Cancelling a getter that already dequeued a message would swallow it, so a message the response does not carry is put back at the head of its queue rather than appended, keeping seq order intact. A gate that closes mid-wait returns its message the same way instead of leaving a completed getter to spin the loop.
One /receive response can carry both the priority and the chatter batch, and the two queues are filled independently, so returning the priority one first let an operator answer be read ahead of the peer chatter it answers. Sort the merged batch by seq before returning it. Routing is untouched: CONTROL commands still ride the priority queue and still pierce the pause gate.
Every tool call built a fresh httpx.Client, so each say/listen/join paid a full TCP (and, over a remote hub, TLS) handshake for a few hundred bytes of payload. Keep one client for the process and lend it to each call, rebuilt when HUB_URL changes and closed at exit. Call sites keep the with-borrow shape, so the borrow stays visually scoped to the call that needs it; only the close is gone.
format_inbound re-attached the same ~230-character warning to every message in a batch, so a ten-message wake spent it ten times for no added protection. Hoist it to the [caucus inbound] header. The defence is unchanged: each body keeps its own untrusted-peer-data delimiters and _defang_fence still neutralizes any a peer plants in its content, so the fence stays unforgeable. The header names the fence without writing the literal delimiter, which would read as an unclosed opener.
_requeue_front assumed the queue could not overflow on the way back in, reasoning that the net count was unchanged. It is not: the getter completed on an earlier scheduling step, so route() can refill the queue to capacity in the window before the message is handed back, and the re-add then raised QueueFull out of the poll's cleanup path. Tolerate the overflow the way _safe_put does, dropping from the tail since everything past the capacity line is the newest traffic and the point here is to preserve the older, in-order head. Two guards behind it. The active_polls decrement moves into its own finally so no cleanup failure can skip it: a leaked count makes register refuse every re-join under that name as a colliding duplicate, wedging the project out of its own identity for the life of the hub. And each completed getter is now retired before its message is acted on, so a raise on the consume path cannot leave a done task in the slot for the next iteration to spin on.
While the room is paused no chatter getter is armed, so nothing in the race could wake the loop on resume: held chatter waited out the rest of the one-second slice before the next iteration re-armed and delivered it. Race state.transmit.wait() alongside the two getters. The gate waiter holds no message, so cancelling the loser costs nothing. The per-peer pause is a plain flag with no event to wait on and still resolves on the slice; that peer polls to its deadline anyway, which is what keeps last_seen fresh so it is not reaped.
FastMCP dispatches sync tool bodies on a thread pool, so the check-then- build in _open_client raced: eight threads entering it together each built their own client, seven of them orphaned with pooled connections nothing would ever release. Put the check-then-build under a lock. A hub-URL rebind no longer closes the displaced client either. Another thread may be mid-request on it, and closing it underneath fails that request; park it instead and let the atexit close collect it with the live one.
The ACK goes out as soon as a batch is handed to the driver, not once the agent has answered it, so a reset that cancels an in-flight turn drops whatever that turn had coalesced. Say so in _poll_inbound and the changelog, including the seam where a pending ACK dies with the poller and that batch stays replayable.
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.
Latency and overlap fixes along the whole message path, from a timing audit of the hop sender → hub → recipient.
set_statuson its own cheap bucket./receiveblocked up to 1s on the chatter queue before noticing operator control; both queues are now raced, a resume wakes a paused poll immediately, and a mixed batch is sorted byseqso an operator answer can't overtake the message that framed it. The requeue path tolerates a full queue (drops newest, logs) and can no longer strand the listener slot.untrusted-peer-datafence, with a test proving a forged delimiter is still neutralized.688→693 tests, mypy strict clean, no protocol text change.