What causes "Bad response from PING health check" in redis/redis-py?

Inspect what the server actually returns by running redis-cli PING against the same endpoint.

redis/redis-py · error · ConnectionError

Bad response from PING health check

Error message

Bad response from PING health check

What it means

Raised as ConnectionError from _send_ping() when a PING returns a response that is not the literal 'PONG'. This is the health-check probe invoked by check_health() before commands when health_check_interval is set. A non-PONG reply indicates the server (or a middlebox) returned unexpected bytes on the same socket.

Solutions

Inspect what the server actually returns by running redis-cli PING against the same endpoint.
Remove any non-RESP-aware proxy between the client and Redis, or configure it for transparent L4 passthrough.
Temporarily raise or disable health_check_interval to isolate whether the failure is the probe vs. real traffic.
Confirm protocol= matches the server's capability (RESP3 push types can confuse an older server).

Example fix

// before
r = redis.asyncio.Redis(host=h, health_check_interval=1)
// after
r = redis.asyncio.Redis(host=h, health_check_interval=30)

Defensive patterns

Strategy: retry

Validate before calling

null

Type guard

from redis.exceptions import ConnectionError

def is_bad_ping(exc: BaseException) -> bool:
    return isinstance(exc, ConnectionError) and 'PING' in str(exc)

Try / catch

from redis.exceptions import ConnectionError
from redis.retry import Retry
from redis.backoff import ExponentialBackoff

# retry the health-check-driven failure automatically
r = redis.asyncio.Redis(
    host=h,
    health_check_interval=30,
    retry=Retry(ExponentialBackoff(), 2),
    retry_on_error=[ConnectionError],
)

Prevention

Use a RESP-aware (L4) proxy; never inject bytes into the Redis stream.
Match protocol= to server capability.
Raise health_check_interval to avoid aggressive probing on slow links.

When it happens

Trigger: Setting health_check_interval=N on the client and the elapsed interval triggers a PING; a proxy/SSL-terminator/reset returns something other than PONG (e.g. an error frame, an AUTH-required reply, or a connection that was silently reassigned).

Common situations: A load balancer or RESP-unaware proxy injecting its own response; the server sent a maintenance/notification frame the parser surfaced as the PING reply; shared connection handed off mid-stream in a forked/branching process; RESP2/RESP3 mismatch causing a push reply to be read as the PONG.

Related errors

AI-assisted analysis of redis/redis-py@6a6b581b48 (2026-08-10). Data as JSON: /api/errors/c98fc5775361023e. Report an issue: GitHub.

Appendix: source

Thrown at redis/asyncio/connection.py:1149

            await record_connection_closed(
                close_reason=CloseReason.APPLICATION_CLOSE,
            )

        if self.maintenance_state == MaintenanceState.MAINTENANCE:
            # MOVING state is owned by the pool-level TTL cleanup. Regular
            # maintenance timeout relaxation can be restored when this
            # connection closes, matching the sync lifecycle.
            self.reset_tmp_settings(reset_relaxed_timeout=True)
            self.maintenance_state = MaintenanceState.NONE
            # reset the sets that keep track of received start maint
            # notifications and skipped end maint notifications
            self.reset_received_notifications()

    async def _send_ping(self):
        """Send PING, expect PONG in return"""
        await self.send_command("PING", check_health=False)
        if str_if_bytes(await self.read_response()) != "PONG":
            raise ConnectionError("Bad response from PING health check")

    async def _ping_failed(self, error, failure_count):
        """Function to call when PING fails"""
        await self.disconnect(
            error=error, failure_count=failure_count, health_check_failed=True
        )

    async def check_health(self):
        """Check the health of the connection with a PING/PONG"""
        if (
            self.health_check_interval
            and asyncio.get_running_loop().time() > self.next_health_check
        ):
            await self.retry.call_with_retry(
                self._send_ping, self._ping_failed, with_failure_count=True
            )

    async def _send_packed_command(self, command: Iterable[bytes]) -> None:

View on GitHub (pinned to 6a6b581b48)