apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-27 06:01:01 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: producer send loop reconnects after failure instead of using None (#1057)

Browse source 
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.
This commit is contained in:
cybermaggedon 2026-07-23 12:13:42 +01:00 • committed by GitHub
parent a0e40950fe
commit bc30c81d53
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 17 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

32
trustgraph-base/trustgraph/base/producer.py
View file

@ -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