async def _process_entities(self) -> None:
        """Process entities with pull-based concurrency control.

        Only pulls an entity from the stream when a worker is available,
        ensuring natural backpressure throughout the pipeline.
        """
        source_node = self.sync_context.dag.get_source_node()

        self.sync_context.logger.info(
            f"Starting pull-based processing from source {self.sync_context.source._name} "
            f"(buffer: {self.stream_buffer_size}, max workers: {self.worker_pool.max_workers})"
        )

        stream_error: Optional[Exception] = None

        try:
            async with AsyncSourceStream(
                self.sync_context.source.generate_entities(),
                queue_size=self.stream_buffer_size,
                logger=self.sync_context.logger,
            ) as stream:
                async for entity in stream.get_entities():
                    # Handle skipped entities without using a worker
                    if getattr(entity, "should_skip", False):
                        self.sync_context.logger.debug(f"Skipping entity: {entity.entity_id}")
                        await self.sync_context.progress.increment("skipped")
                        continue

                    # Wait for a worker slot before processing
                    # This creates natural backpressure - we only pull the next entity
                    # when we have capacity to process it
                    async with self.worker_pool.semaphore:
                        await self.entity_processor.process(
                            entity=entity,
                            source_node=source_node,
                            sync_context=self.sync_context,
                        )

        except Exception as e:
            stream_error = e
            self.sync_context.logger.error(f"Error during entity streaming: {e}")
            raise

        finally:
            # No need to wait for completion - all processing is done inline
            await self.sync_context.progress.finalize(is_complete=(stream_error is None))
Process entities with pull-based concurrency control.

        Only pulls an entity from the stream when a worker is available,
        ensuring natural backpressure throughout the pipeline.
        
backend/airweave/platform/sync/orchestrator.py