fixed async tracer bug

attempting to increment partition for wrong system id
synth-laboratories · Nov 30, 2024 · 7156d84 · 7156d84
1 parent dbeb18c
commit 7156d84
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 18 deletions.
diff --git a/synth_sdk/tracing/decorators.py b/synth_sdk/tracing/decorators.py
@@ -324,9 +324,7 @@ async def async_wrapper(*args, **kwargs):
                         event.partition_index = event_store.increment_partition(
                             system_id_var.get()
                         )
-                        # logger.debug(
-                        #     f"Incremented partition to: {event.partition_index}"
-                        # )
+                        logger.debug(f"Incremented partition to: {event.partition_index}")
 
                     set_current_event(event, decorator_type="async")
                     logger.debug(f"Created and set new event: {event_type}")

diff --git a/synth_sdk/tracing/events/manage.py b/synth_sdk/tracing/events/manage.py
@@ -48,6 +48,9 @@ def set_current_event(event: Optional["Event"], decorator_type: Literal["sync",
 
         # If there's an existing event of the same type, end it
         if event.event_type in _local.active_events:
+
+            # TODO: Check that the active event has the same system_id as the one we're settting
+
             logger.debug(f"Found existing event of type {event.event_type}")
             existing_event = _local.active_events[event.event_type]
             existing_event.closed = time.time()
@@ -76,23 +79,26 @@ def set_current_event(event: Optional["Event"], decorator_type: Literal["sync",
 
         # If there's an existing event of the same type, end it
         if event.event_type in active_events:
-            logger.debug(f"Found existing event of type {event.event_type}")
             existing_event = active_events[event.event_type]
-            existing_event.closed = time.time()
-            logger.debug(
-                f"Closed existing event of type {event.event_type} at {existing_event.closed}"
-            )
 
-            # Store the closed event if system_id is present
-            system_id = system_id_var.get()
-            if system_id:
-                logger.debug(f"Storing closed event for system {system_id}")
-                try:
-                    event_store.add_event(system_id, existing_event)
-                    logger.debug("Successfully stored closed event")
-                except Exception as e:
-                    logger.error(f"Failed to store closed event: {str(e)}")
-                    raise
+            # Check that the active event has the same system_id as the one we're settting
+            if existing_event.system_id == event.system_id:
+                logger.debug(f"Found existing event of type {event.event_type}")
+                existing_event.closed = time.time()
+                logger.debug(
+                    f"Closed existing event of type {event.event_type} at {existing_event.closed}"
+                )
+
+                # Store the closed event if system_id is present
+                system_id = system_id_var.get()
+                if system_id:
+                    logger.debug(f"Storing closed event for system {system_id}")
+                    try:
+                        event_store.add_event(system_id, existing_event)
+                        logger.debug("Successfully stored closed event")
+                    except Exception as e:
+                        logger.error(f"Failed to store closed event: {str(e)}")
+                        raise
 
         # Set the new event
         active_events[event.event_type] = event