async def subscribe(self, job_id: UUID) -> redis.client.PubSub:
        """Create a new pubsub instance and subscribe to a sync job's updates.

        Args:
            job_id: The sync job ID to subscribe to.

        Returns:
            redis.client.PubSub: A new Redis pubsub instance subscribed to this job's channel.
        """
        channel = self._channel_name(job_id)

        # Get socket keepalive options based on OS
        if platform.system() == "Darwin":
            socket_keepalive_options = {}
        else:
            # Use the correct Linux TCP keepalive constants
            import socket

            if hasattr(socket, "TCP_KEEPIDLE"):
                socket_keepalive_options = {
                    socket.TCP_KEEPIDLE: 60,  # Start keepalive after 60s idle
                    socket.TCP_KEEPINTVL: 10,  # Interval between keepalive probes
                    socket.TCP_KEEPCNT: 6,  # Number of keepalive probes
                }
            else:
                socket_keepalive_options = {}

        # Build Redis URL with authentication
        if settings.REDIS_PASSWORD:
            redis_url = f"redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}"
        else:
            redis_url = f"redis://{settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}"

        # Create a new Redis client directly for pubsub to avoid connection pool issues
        # This is a workaround for async pubsub issues in Docker environments
        pubsub_redis = await redis.from_url(
            redis_url,
            decode_responses=True,
            socket_keepalive=True,
            socket_connect_timeout=5,
            # Don't set socket_timeout for pubsub connections - they need to stay open
            socket_keepalive_options=socket_keepalive_options,
        )

        # Create pubsub instance
        pubsub = pubsub_redis.pubsub()

        # Subscribe to the specific channel
        await pubsub.subscribe(channel)

        logger.info(f"Created new pubsub subscription for sync job {job_id}")
        return pubsub
Create a new pubsub instance and subscribe to a sync job's updates.

        Args:
            job_id: The sync job ID to subscribe to.

        Returns:
            redis.client.PubSub: A new Redis pubsub instance subscribed to this job's channel.
        
backend/airweave/platform/sync/pubsub.py