async def increment(self, stat_name: str, amount: int = 1) -> None:
        """Increment a counter and trigger update if threshold reached.

        Uses async lock to prevent race conditions from concurrent workers.
        """
        async with self._lock:  # CRITICAL FIX: Synchronize access
            current_value = getattr(self.stats, stat_name, 0)
            setattr(self.stats, stat_name, current_value + amount)

            # Include ALL operations in threshold calculation (including skipped)
            total_ops = sum(
                [
                    self.stats.inserted,
                    self.stats.updated,
                    self.stats.deleted,
                    self.stats.kept,
                    self.stats.skipped,
                ]
            )

            # Check if we should publish
            if total_ops - self._last_published >= self._publish_threshold:
                logger.info(f"Progress threshold reached: {total_ops} total ops, publishing update")
                await self._publish()
                self._last_published = total_ops
            else:
                logger.debug(
                    f"Progress: {stat_name}={current_value + amount}, total={total_ops}, "
                    f"threshold={self._publish_threshold}"
                )
Increment a counter and trigger update if threshold reached.

        Uses async lock to prevent race conditions from concurrent workers.
        
backend/airweave/platform/sync/pubsub.py