From bc30c81d5381c88418fb14be68082358a5983783 Mon Sep 17 00:00:00 2001 From: cybermaggedon Date: Thu, 23 Jul 2026 12:13:42 +0100 Subject: [PATCH] fix: producer send loop reconnects after failure instead of using None (#1057) The send method had two sequential while loops: one for connecting and one for sending. After a send failure the producer was closed and set to None, but control stayed in the send loop rather than returning to the connect loop. The next iteration called .send() on None, then .close() on None in the except block, propagating an AttributeError that masked the original error (e.g. MessageTooBig). Nest the connect loop inside the send loop so failures trigger reconnection, and guard the .close() call against None. --- trustgraph-base/trustgraph/base/producer.py | 32 +++++++++++---------- 1 file changed, 17 insertions(+), 15 deletions(-) diff --git a/trustgraph-base/trustgraph/base/producer.py b/trustgraph-base/trustgraph/base/producer.py index 9af9d22e..e21f317f 100644 --- a/trustgraph-base/trustgraph/base/producer.py +++ b/trustgraph-base/trustgraph/base/producer.py @@ -42,24 +42,25 @@ class Producer: if not self.running: return - while self.running and self.producer is None: + while self.running: - try: - logger.info(f"Connecting publisher to {self.topic}...") - self.producer = self.backend.create_producer( - topic = self.topic, - schema = self.schema, - chunking_enabled = self.chunking_enabled, - ) - logger.info(f"Connected publisher to {self.topic}") - except Exception as e: - logger.error(f"Exception connecting publisher: {e}", exc_info=True) - await asyncio.sleep(2) + # Reconnect if needed + while self.running and self.producer is None: + + try: + logger.info(f"Connecting publisher to {self.topic}...") + self.producer = self.backend.create_producer( + topic = self.topic, + schema = self.schema, + chunking_enabled = self.chunking_enabled, + ) + logger.info(f"Connected publisher to {self.topic}") + except Exception as e: + logger.error(f"Exception connecting publisher: {e}", exc_info=True) + await asyncio.sleep(2) if not self.running: break - while self.running: - try: await asyncio.to_thread( @@ -75,6 +76,7 @@ class Producer: except Exception as e: logger.error(f"Exception sending message: {e}", exc_info=True) - self.producer.close() + if self.producer: + self.producer.close() self.producer = None