Subscriber resilience: recreate consumer after connection failure

Move consumer creation from Subscriber.start() into the run() loop,
matching the pattern used by Consumer. If the connection drops and the
consumer is closed in the finally block, the loop now recreates it on
the next iteration instead of spinning forever on a None consumer.

Previously, start() created the consumer once and run() assumed it
existed for the lifetime of the subscriber. A connection failure would
set self.consumer = None in the finally cleanup, and the outer retry
loop would re-enter with no consumer, causing an infinite
'NoneType has no attribute receive' error loop.
This commit is contained in:
Cyber MacGeddon 2026-04-07 12:33:52 +01:00
parent ddd4bd7790
commit 34d87727fc
3 changed files with 22 additions and 30 deletions

View file

@ -45,15 +45,6 @@ class Subscriber:
async def start(self):
# Create consumer via backend
self.consumer = await asyncio.to_thread(
self.backend.create_consumer,
topic=self.topic,
subscription=self.subscription,
schema=self.schema,
consumer_type='exclusive',
)
self.task = asyncio.create_task(self.run())
async def stop(self):
@ -80,6 +71,16 @@ class Subscriber:
try:
# Create consumer if needed (first run or after failure)
if self.consumer is None:
self.consumer = await asyncio.to_thread(
self.backend.create_consumer,
topic=self.topic,
subscription=self.subscription,
schema=self.schema,
consumer_type='exclusive',
)
if self.metrics:
self.metrics.state("running")
@ -179,8 +180,8 @@ class Subscriber:
if not self.running and not self.draining:
return
# If handler drops out, sleep a retry
# Sleep before retry
await asyncio.sleep(1)
async def subscribe(self, id):