Decrement the count of outstanding tasks before returning in the even…

…t of an exception. (#406)
microsoft · Aug 27, 2024 · 62e5e5b · 62e5e5b
1 parent c8084ad
commit 62e5e5b
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/python/src/agnext/application/_single_threaded_agent_runtime.py b/python/src/agnext/application/_single_threaded_agent_runtime.py
@@ -272,6 +272,7 @@ async def _process_send(self, message_envelope: SendMessageEnvelope) -> None:
                 )
         except BaseException as e:
             message_envelope.future.set_exception(e)
+            self._outstanding_tasks.decrement()
             return
 
         self._message_queue.append(
@@ -327,6 +328,7 @@ async def _process_publish(self, message_envelope: PublishMessageEnvelope) -> No
         except BaseException as e:
             # Ignore cancelled errors from logs
             if isinstance(e, CancelledError):
+                self._outstanding_tasks.decrement()
                 return
             logger.error("Error processing publish message", exc_info=True)
         finally: