@@ -161,6 +161,11 @@ def __init__(self) -> None:
161161 self ._send_tasks : set [Any ] = set ()
162162 self ._timeouts = 0
163163 self ._disabled = False
164+ # Requests are counted against the round trip which carried them - one send of
165+ # a batch, or one answer which came back - rather than against each other.
166+ self ._round_trip = 0
167+ self ._batch_of_request : dict [str , int ] = {}
168+ self ._struck : set [int ] = set ()
164169 self ._cache : dict [tuple [str , tuple [str , ...]], Any ] = {}
165170 self ._inflight : dict [
166171 tuple [str , tuple [str , ...]], tuple [str , Future | None ]
@@ -206,6 +211,7 @@ def _abandon_pending(self, reason: str) -> None:
206211 if not future .done ():
207212 future .set_exception (EventProxyUnavailable (reason ))
208213 self ._pending .clear ()
214+ self ._batch_of_request .clear ()
209215 self ._inflight .clear ()
210216 self ._outgoing .clear ()
211217 self ._flush_scheduled = False
@@ -221,6 +227,9 @@ def handle_response(self, message: Mapping[str, Any]) -> bool:
221227 if message .get ("type" ) != EVENT_PROXY_RESPONSE_TYPE :
222228 return False
223229
230+ # Everything which came back in this message travelled as one round trip
231+ self ._begin_round_trip ()
232+
224233 responses = message .get ("responses" )
225234 if isinstance (responses , list ):
226235 for entry in responses :
@@ -234,9 +243,7 @@ def handle_response(self, message: Mapping[str, Any]) -> bool:
234243 def _route (self , message : Mapping [str , Any ]) -> None :
235244 """Deliver one answer to the request which asked for it"""
236245 request_id = message .get ("requestId" )
237- future = (
238- self ._pending .pop (request_id , None ) if isinstance (request_id , str ) else None
239- )
246+ future = self ._take (request_id )
240247 if future is None :
241248 # Either the request timed out or the layout was closed while it was in
242249 # flight. Either way there is nobody left to deliver the answer to.
@@ -344,7 +351,7 @@ async def resolve_many(
344351 failure : BaseException | None = None
345352 for (index , key , _ ), payload in zip (outstanding , payloads , strict = True ):
346353 if isinstance (payload , asyncio .CancelledError ):
347- raise payload
354+ raise payload # nocov - nothing cancels a request which is in flight
348355 if isinstance (payload , BaseException ):
349356 failure = _prefer_failure (failure , payload )
350357 continue
@@ -439,9 +446,14 @@ def _flush(self) -> None:
439446 batch , self ._outgoing = self ._outgoing , []
440447 if not batch :
441448 return
449+ self ._begin_round_trip ()
442450 if self ._loop is None or self ._send is None :
443451 self ._abandon (batch )
444452 return
453+ for entry in batch :
454+ request_id = entry .get ("requestId" )
455+ if isinstance (request_id , str ) and request_id in self ._pending :
456+ self ._batch_of_request [request_id ] = self ._round_trip
445457 message = (
446458 batch [0 ]
447459 if len (batch ) == 1
@@ -453,12 +465,7 @@ def _flush(self) -> None:
453465
454466 def _abandon (self , batch : Sequence [dict [str , Any ]]) -> None :
455467 for message in batch :
456- request_id = message .get ("requestId" )
457- future = (
458- self ._pending .pop (request_id , None )
459- if isinstance (request_id , str )
460- else None
461- )
468+ future = self ._take (message .get ("requestId" ))
462469 if future is not None and not future .done ():
463470 future .set_exception (
464471 EventProxyUnavailable ("no sender is bound to this connection" )
@@ -505,25 +512,56 @@ async def _send_message(self, message: dict[str, Any]) -> None:
505512 except Exception as error : # any send failure kills live access for this client
506513 self .disable (f"failed to send a live event request ({ error } )" )
507514
508- def _discard (self , request_id : str ) -> None :
509- self ._pending .pop (request_id , None )
515+ def _take (self , request_id : Any ) -> Future | None :
516+ """Stop waiting on one request, forgetting which round trip carried it too"""
517+ if not isinstance (request_id , str ):
518+ return None
519+ self ._batch_of_request .pop (request_id , None )
520+ return self ._pending .pop (request_id , None )
521+
522+ def _begin_round_trip (self ) -> None :
523+ """Count one message, in either direction, as a round trip of its own"""
524+ self ._round_trip += 1
525+ self ._struck .clear ()
510526
511527 def _handle_timeout (
512528 self , request_id : str , event_id : str , path : Sequence [str | int ]
513529 ) -> NoReturn :
514- self ._discard (request_id )
515- self ._timeouts += 1
516- max_timeouts = REACTPY_EVENT_PROXY_MAX_TIMEOUTS .current
517- if max_timeouts > 0 and self ._timeouts >= max_timeouts :
518- self .disable (
519- f"{ self ._timeouts } consecutive live event requests went unanswered"
520- )
530+ self ._strike (request_id )
521531 msg = (
522532 f"the client did not answer the request for { list (path )!r} of event "
523533 f"{ event_id !r} within { REACTPY_EVENT_PROXY_TIMEOUT .current } s"
524534 )
525535 raise EventProxyTimeout (msg )
526536
537+ def _strike (self , request_id : str ) -> None :
538+ """Record one round trip which the client did not answer
539+
540+ A batch which went silent used to be counted once per property it was asking
541+ for, so a handler which read four properties of one event exhausted a budget of
542+ three all by itself, and live access was switched off for a client which had
543+ merely been asked several things at once. What is being measured is whether the
544+ client answers at all, which is why one unanswered message costs one strike no
545+ matter how many properties it was after.
546+
547+ Only the messages which are still outstanding can be struck, since starting a
548+ round trip in either direction clears this - so it cannot accumulate.
549+ """
550+ batch = self ._batch_of_request .pop (request_id , None )
551+ self ._take (request_id )
552+ if batch is None :
553+ # A timeout the client reported itself belongs to the answer being handled
554+ batch = self ._round_trip
555+ if batch in self ._struck :
556+ return
557+ self ._struck .add (batch )
558+ self ._timeouts += 1
559+ max_timeouts = REACTPY_EVENT_PROXY_MAX_TIMEOUTS .current
560+ if max_timeouts > 0 and self ._timeouts >= max_timeouts :
561+ self .disable (
562+ f"{ self ._timeouts } consecutive live event round trips went unanswered"
563+ )
564+
527565 def _to_value (
528566 self ,
529567 payload : Mapping [str , Any ],
0 commit comments