async def _producer(self):
        """Producer task that fills the queue from the source generator."""
        try:
            items_produced = 0
            async for item in self.source_generator:
                if not self.is_running:
                    self.logger.info("Producer stopping early")
                    break

                # Put item in queue, waiting if queue is full.
                # This is a blocking call, so consumer will wait until the queue has space
                # Effectively, this is a backpressure mechanism.
                await self.queue.put(item)
                items_produced += 1

                # Log progress periodically
                if items_produced % 50 == 0:
                    self.logger.info(
                        f"AsyncSourceStream producer progress: {items_produced} items queued, "
                        f"queue size: {self.queue.qsize()}/{self.queue.maxsize}"
                    )

            self.logger.info(f"Source generator exhausted after producing {items_produced} items")
        except Exception as e:
            self.logger.error(f"Error in producer: {e}")
            self.producer_exception = e
            # Re-raise to ensure proper error handling -> THIS DOES NOT WORK
            raise
        finally:
            # Signal we're done by putting None in the queue and setting the done event
            await self.queue.put(None)
            self.producer_done.set()
Producer task that fills the queue from the source generator.
backend/airweave/platform/sync/stream.py