redis/redis-py · critical · InitialHealthCheckFailedError

Initial health check failed. Initial health check policy

Error message

Initial health check failed. Initial health check policy: {initial_health_check_policy}

What it means

Raised as InitialHealthCheckFailedError by MultiDBClient._perform_initial_health_check() (client.py:414) when the proportion of healthy databases after the initial check does not satisfy the configured initial_health_check_policy (ALL_AVAILABLE, MAJORITY_AVAILABLE, or ONE_AVAILABLE). The message interpolates the active policy so the caller can see which bar was missed.

Solutions

  1. Inspect which databases failed (catch the error, then check each endpoint with redis-cli PING) and fix them.
  2. If partial availability is acceptable, switch initial_health_check_policy to MAJORITY_AVAILABLE or ONE_AVAILABLE.
  3. Defer client construction until enough databases are reachable (readiness gate).
  4. Review health-check definitions and circuit-breaker thresholds to ensure they aren't too strict.

Example fix

# before
config.initial_health_check_policy = InitialHealthCheck.ALL_AVAILABLE
client = MultiDBClient(config)  # raises if any DB unhealthy

# after
config.initial_health_check_policy = InitialHealthCheck.MAJORITY_AVAILABLE
client = MultiDBClient(config)  # tolerates a minority of unhealthy DBs
Defensive patterns

Strategy: validation

Validate before calling

# Pre-flight: probe each endpoint before constructing the client
from redis.config import MultiDbConfig, InitialHealthCheck  # adjust import

healthy = probe_all_endpoints(config)  # your PING-based check
if config.initial_health_check_policy == InitialHealthCheck.ALL_AVAILABLE and not all(healthy):
    raise RuntimeError('not all endpoints healthy; ALL_AVAILABLE policy will fail')
if config.initial_health_check_policy == InitialHealthCheck.MAJORITY_AVAILABLE and sum(healthy) <= len(healthy) / 2:
    raise RuntimeError('no majority healthy; MAJORITY_AVAILABLE policy will fail')
if config.initial_health_check_policy == InitialHealthCheck.ONE_AVAILABLE and not any(healthy):
    raise RuntimeError('no endpoint healthy; ONE_AVAILABLE policy will fail')

Type guard

from redis.multidb.config import InitialHealthCheck

def policy_satisfied(policy, healthy_flags) -> bool:
    if policy == InitialHealthCheck.ALL_AVAILABLE:
        return all(healthy_flags)
    if policy == InitialHealthCheck.MAJORITY_AVAILABLE:
        return sum(healthy_flags) > len(healthy_flags) / 2
    if policy == InitialHealthCheck.ONE_AVAILABLE:
        return any(healthy_flags)
    return False

Try / catch

from redis.multidb.exception import InitialHealthCheckFailedError
import time

for _ in range(30):
    try:
        client = MultiDBClient(config)
        client.initialize()
        break
    except InitialHealthCheckFailedError:
        time.sleep(3)
else:
    # fall back to a more permissive policy
    config.initial_health_check_policy = InitialHealthCheck.ONE_AVAILABLE
    client = MultiDBClient(config)

Prevention

When it happens

Trigger: Constructing MultiDBClient and calling initialize() (or execute_command, which triggers initialize) when health-check results violate the policy: ALL_AVAILABLE with any unhealthy DB; MAJORITY_AVAILABLE with <= half healthy; ONE_AVAILABLE with none healthy.

Common situations: Partial fleet outage at boot (some regions down); ALL_AVAILABLE policy being too strict for geographically distributed fleets; one misconfigured endpoint dragging the majority check below 50%; network issues affecting a subset of databases.

Related errors


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

Appendix: source

Thrown at redis/multidb/client.py:414

        Runs initial health check and evaluate healthiness based on initial_health_check_policy.
        """
        results = await self._check_databases_health()
        is_healthy = True

        if self._config.initial_health_check_policy == InitialHealthCheck.ALL_AVAILABLE:
            is_healthy = False not in results.values()
        elif (
            self._config.initial_health_check_policy
            == InitialHealthCheck.MAJORITY_AVAILABLE
        ):
            is_healthy = sum(results.values()) > len(results) / 2
        elif (
            self._config.initial_health_check_policy == InitialHealthCheck.ONE_AVAILABLE
        ):
            is_healthy = True in results.values()

        if not is_healthy:
            raise InitialHealthCheckFailedError(
                f"Initial health check failed. Initial health check policy: {self._config.initial_health_check_policy}"
            )

    def _on_circuit_state_change_callback(
        self, circuit: CircuitBreaker, old_state: CBState, new_state: CBState
    ):
        if new_state == CBState.HALF_OPEN:
            self._bg_scheduler.run_coro_fire_and_forget(
                self._check_db_health, circuit.database
            )
            return

        if old_state == CBState.CLOSED and new_state == CBState.OPEN:
            logger.warning(
                f"Database {circuit.database} is unreachable. Failover has been initiated."
            )

            self._bg_scheduler.run_once(

View on GitHub (pinned to 6a6b581b48)