From 468d184a185d73849e24849bbd2551e89fbd6f64 Mon Sep 17 00:00:00 2001 From: Remy Tuyeras Date: Tue, 27 Jan 2026 08:24:46 -0500 Subject: [PATCH] introduce route_key in dna --- summoner/client/client.py | 83 ++++++++++++++++++++++++++++-------- summoner/protocol/flow.py | 4 -- summoner/protocol/payload.py | 1 + 3 files changed, 66 insertions(+), 22 deletions(-) diff --git a/summoner/client/client.py b/summoner/client/client.py index 7f63eb0..436102a 100644 --- a/summoner/client/client.py +++ b/summoner/client/client.py @@ -74,7 +74,7 @@ class SummonerClient: DEFAULT_EVENT_BRIDGE_SIZE = 1000 DEFAULT_MAX_CONSECUTIVE_ERRORS = 3 # Failed attempts to send before disconnecting - core_version = "1.1.0" + core_version = "1.1.1" def __init__(self, name: Optional[str] = None): @@ -362,7 +362,6 @@ def decorator(fn: Callable[[Optional[Union[str, dict]]], Optional[Union[str, dic "source": inspect.getsource(fn), }) - # ----[ Registration Code ]---- async def register(): async with self.hooks_lock: @@ -427,20 +426,29 @@ def decorator(fn: Callable[[Union[str, dict]], Awaitable[Optional[Event]]]): # ----[ Registration Code ]---- async def register(): receiver = Receiver(fn=fn, priority=tuple_priority) + + parsed_route = None + normalized_route = route if self._flow.in_use: - parsed_route = self._flow.parse_route(route) - normalized_route = str(parsed_route) + try: + parsed_route = self._flow.parse_route(route) + normalized_route = str(parsed_route) + except Exception as e: + self.logger.warning( + f"@receive: could not parse route {route!r} while flow is enabled; " + f"registering raw route. Error: {type(e).__name__}: {e}" + ) + parsed_route = None + normalized_route = route async with self.routes_lock: if route in self.receiver_index: self.logger.warning(f"Route '{route}' already exists. Overwriting.") - if self._flow.in_use: + if self._flow.in_use and parsed_route is not None: self.receiver_parsed_routes[normalized_route] = parsed_route - self.receiver_index[normalized_route] = receiver - else: - self.receiver_index[route] = receiver + self.receiver_index[normalized_route] = receiver # ----[ Safe Registration ]---- self._schedule_registration(register()) @@ -521,16 +529,28 @@ async def register(): actions_exist = isinstance(on_actions, set) and bool(on_actions) triggers_exist = isinstance(on_triggers, set) and bool(on_triggers) + parsed_route = None + normalized_route = route + if self._flow.in_use: - parsed_route = self._flow.parse_route(route) - normalized_route = str(parsed_route) + try: + parsed_route = self._flow.parse_route(route) + normalized_route = str(parsed_route) + except Exception as e: + self.logger.warning( + f"@send: could not parse route {route!r} while flow is enabled; " + f"registering raw route. Error: {type(e).__name__}: {e}" + ) + parsed_route = None + normalized_route = route async with self.routes_lock: if self._flow.in_use: self.sender_index.setdefault(normalized_route, []) self.sender_index[normalized_route].append(sender) - if route not in self.sender_parsed_routes and actions_exist or triggers_exist: - self.sender_parsed_routes[normalized_route] = parsed_route + + if parsed_route is not None and (actions_exist or triggers_exist): + self.sender_parsed_routes.setdefault(normalized_route, parsed_route) else: self.sender_index.setdefault(route, []) self.sender_index[route].append(sender) @@ -702,9 +722,21 @@ def dna(self, include_context: bool = False) -> str: # All receivers for dna in self._dna_receivers: fn = dna["fn"] + raw_route = dna["route"] + + try: + if self._flow.in_use: + route_key = str(self._flow.parse_route(raw_route)) + else: + route_key = raw_route + except Exception: + route_key = raw_route + route_key = "".join(str(route_key).split()) + entries.append({ "type": "receive", - "route": dna["route"], + "route": raw_route, # original route string + "route_key": route_key, # stable route representative "priority": dna["priority"], "source": get_callable_source(fn, dna.get("source")), "module": fn.__module__, @@ -714,9 +746,21 @@ def dna(self, include_context: bool = False) -> str: # All senders for dna in self._dna_senders: fn = dna["fn"] + raw_route = dna["route"] + + try: + if self._flow.in_use: + route_key = str(self._flow.parse_route(raw_route)) + else: + route_key = raw_route + except Exception: + route_key = raw_route + route_key = "".join(str(route_key).split()) + entries.append({ "type": "send", - "route": dna["route"], + "route": raw_route, # original route string + "route_key": route_key, # stable route representative "multi": dna["multi"], # Serialize triggers/actions by name so they can be re-resolved later. "on_triggers": [t.name for t in (dna["on_triggers"] or [])], @@ -981,7 +1025,8 @@ async def _safe_call(fn: Callable[[Any], Awaitable], payload: Any) -> Any: except Exception as e: self.logger.error( f"Receiving hook {receiving_hook.__name__} (priority={priority}) " - f"failed on payload {payload!r}: {e}" + f"failed on payload {payload!r}: {e}", + exc_info=True ) new_payload = payload payload = new_payload @@ -1123,7 +1168,8 @@ async def _send_worker( except Exception as e: self.logger.error( f"[route={route}] Sending hook {sending_hook.__name__} (priority={priority}) " - f"failed on payload {payload!r}: {e}" + f"failed on payload {payload!r}: {e}", + exc_info=True ) new_payload = payload payload = new_payload @@ -1454,13 +1500,14 @@ async def _retry_loop(self, host, port, limit, stage = "Primary"): f"[{type(e).__name__}: {e}] " f"({stage}) retry {attempts} of " f"{limit if limit is not None else '∞'}; " - f"sleeping {self.retry_delay_seconds}s" + f"sleeping {self.retry_delay_seconds}s", + exc_info=True ) await asyncio.sleep(self.retry_delay_seconds) # Check retry limit if (limit is not None and attempts >= limit): - self.logger.error(f"{stage} retry limit reached ({limit})") + self.logger.error(f"{stage} retry limit reached ({limit})", exc_info=True) return False async def _get_client_intent(self) -> ClientIntent: diff --git a/summoner/protocol/flow.py b/summoner/protocol/flow.py index da9614b..1f3e6b5 100644 --- a/summoner/protocol/flow.py +++ b/summoner/protocol/flow.py @@ -267,7 +267,3 @@ def parse_route(self, route: str) -> ParsedRoute: def parse_routes(self, routes: list[str]) -> list[ParsedRoute]: return [self.parse_route(route=route) for route in routes] - - - - diff --git a/summoner/protocol/payload.py b/summoner/protocol/payload.py index ba98642..09b7b0c 100644 --- a/summoner/protocol/payload.py +++ b/summoner/protocol/payload.py @@ -156,6 +156,7 @@ def cast_v0_0_1(val: Any, expected: Any) -> Any: register_envelope_version("1.0.0", parse_v0_0_1, cast_v0_0_1) register_envelope_version("1.0.1", parse_v0_0_1, cast_v0_0_1) register_envelope_version("1.1.0", parse_v0_0_1, cast_v0_0_1) +register_envelope_version("1.1.1", parse_v0_0_1, cast_v0_0_1) def wrap_with_types(