# How the Health Checker Verifies Memgraph and Qdrant Connections in code-graph-rag

> Learn how the Health Checker verifies Memgraph and Qdrant connections in code-graph-rag using TCP and HTTP checks. Ensure your RAG pipeline starts smoothly.

- Repository: [Vitali Avagyan/code-graph-rag](https://github.com/vitali87/code-graph-rag)
- Tags: how-to-guide
- Published: 2026-08-16

---

**The `HealthChecker` class in [`codebase_rag/tools/health_checker.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/tools/health_checker.py) orchestrates TCP and HTTP health checks to confirm Memgraph and Qdrant are reachable before the RAG pipeline starts.**

This article explains how the [vitali87/code-graph-rag](https://github.com/vitali87/code-graph-rag) repository validates external service dependencies at runtime. The health checker implementation ensures that both the **Memgraph** graph database and **Qdrant** vector store are operational before queries are processed, preventing failures downstream in the retrieval-augmented generation (RAG) workflow.

## Architecture of the Health Checking System

The health verification system consists of three coordinated components: a reusable checker utility, service-specific validation functions, and a stack manager that orchestrates startup.

### Core Components

| Component | Purpose | Implementation Location |
|-----------|---------|------------------------|
| `HealthChecker` class | Registry and executor for health checks | [`codebase_rag/tools/health_checker.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/tools/health_checker.py) |
| `add_check(name, fn)` | Registers a named check function | [`codebase_rag/tools/health_checker.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/tools/health_checker.py) |
| `run_all_checks()` | Executes all registered checks sequentially | [`codebase_rag/tools/health_checker.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/tools/health_checker.py) |
| `wait_for_memgraph()` | TCP/Bolt handshake validation for Memgraph | [`codebase_rag/stack/health.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/health.py) |
| `wait_for_qdrant()` | HTTP GET validation for Qdrant REST API | [`codebase_rag/stack/health.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/health.py) |
| `wait_healthy()` | Orchestrates blocking startup verification | [`codebase_rag/stack/manager.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/manager.py) |

The flow proceeds as follows:

1. The **stack manager** instantiates `HealthChecker`
2. It registers `wait_for_memgraph` and `wait_for_qdrant` as named checks
3. `run_all_checks()` executes with retry logic until both succeed or timeout
4. Startup aborts if any check fails, ensuring no operations against unavailable services

## How Memgraph Connection Verification Works

The `wait_for_memgraph(host, port, timeout)` function in [[`codebase_rag/stack/health.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/health.py)](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/health.py#L12-L30) validates the graph database through **TCP socket connection** followed by the **Bolt protocol handshake**.

```python
from codebase_rag.stack.health import wait_for_memgraph

# Verify Memgraph is accepting connections on default Bolt port 7687

is_healthy = wait_for_memgraph(
    host="localhost",
    port=7687,
    timeout=30.0  # seconds

)

```

The implementation attempts to:

- Open a TCP socket to the specified `host:port`
- Perform the Memgraph Bolt protocol handshake using the Python driver
- Return `True` upon successful handshake, `False` if timeout expires

This confirms Memgraph is not merely listening on the port but actually ready to process Cypher queries.

## How Qdrant Connection Verification Works

The `wait_for_qdrant(host, port, timeout)` function in [[`codebase_rag/stack/health.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/health.py)](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/health.py#L31-L48) validates the vector store through **HTTP health endpoint polling**.

```python
from codebase_rag.stack.health import wait_for_qdrant

# Verify Qdrant REST API is responsive on default port 6333

is_healthy = wait_for_qdrant(
    host="localhost",
    port=6333,
    timeout=30.0  # seconds

)

```

The implementation:

- Sends HTTP `GET` request to `http://{host}:{port}/`
- Expects HTTP 200 response indicating the REST API is operational
- Retries with exponential backoff until success or timeout

This lightweight check confirms Qdrant's vector search capabilities are available without requiring actual query execution.

## Using HealthChecker Directly

For custom integrations, instantiate `HealthChecker` and register checks explicitly as shown in [[`codebase_rag/tools/health_checker.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/tools/health_checker.py)](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/tools/health_checker.py):

```python
from codebase_rag.tools.health_checker import HealthChecker
from codebase_rag.stack.health import wait_for_memgraph, wait_for_qdrant

checker = HealthChecker()

# Register checks with descriptive names

checker.add_check(
    "memgraph",
    lambda: wait_for_memgraph("memgraph", 7687, timeout=30)
)
checker.add_check(
    "qdrant",
    lambda: wait_for_qdrant("qdrant", 6333, timeout=30)
)

# Execute all checks

healthy, details = checker.run_all_checks()

print(f"All services healthy: {healthy}")
print(f"Results: {details}")  # {'memgraph': True, 'qdrant': True}

```

The `run_all_checks()` method returns:
- `healthy`: Boolean indicating if all checks passed
- `details`: Dictionary mapping check names to their boolean results

## Integrated Startup Sequence

The default behavior uses `StackManager` in [[`codebase_rag/stack/manager.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/manager.py)](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/manager.py#L13-L27), which internally calls `wait_healthy()`:

```python
from codebase_rag.stack.manager import StackManager

# Blocks until both Memgraph and Qdrant are verified healthy

stack = StackManager()
stack.start()  # Raises exception if health checks fail within timeout

```

The manager handles:

- **Retry logic**: Exponential backoff between check attempts
- **Timeout enforcement**: Hard limit on total startup wait time
- **Diagnostic logging**: Records duration and failure reasons per service
- **Graceful failure**: Raises descriptive exception identifying which service failed

## Configuration and Timeout Behavior

Both check functions accept configurable parameters:

| Parameter | Default | Description |
|-----------|---------|-------------|
| `host` | `"localhost"` | Service hostname or IP address |
| `port` | `7687` (Memgraph), `6333` (Qdrant) | Service port number |
| `timeout` | `30.0` | Maximum seconds to wait for service readiness |

Adjust these when deploying with container orchestration or non-standard ports:

```python

# Docker Compose service names as hosts

checker.add_check("memgraph", 
    lambda: wait_for_memgraph("memgraph-db", 7687, timeout=60))
checker.add_check("qdrant",
    lambda: wait_for_qdrant("qdrant-server", 6333, timeout=60))

```

## Summary

- **`HealthChecker`** provides a reusable registry pattern for service health validation in [`codebase_rag/tools/health_checker.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/tools/health_checker.py)
- **`wait_for_memgraph`** verifies Bolt protocol connectivity to confirm Memgraph query readiness
- **`wait_for_qdrant`** polls the HTTP REST endpoint to confirm Qdrant vector search availability
- **`StackManager`** orchestrates blocking startup verification, ensuring the RAG pipeline only proceeds when dependencies are confirmed healthy
- All checks are **idempotent**, **retry-safe**, and include **diagnostic timing** for observability

## Frequently Asked Questions

### How does the health checker handle transient network failures?

The `wait_for_memgraph` and `wait_for_qdrant` functions implement retry loops with exponential backoff. Each attempt is timed and logged, allowing the caller to specify a total timeout budget. As implemented in [`codebase_rag/stack/health.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/health.py), checks continue retrying until success or timeout expiration, making them resilient to container startup delays and network blips.

### Can custom health checks be added beyond Memgraph and Qdrant?

Yes. The `HealthChecker.add_check(name, fn)` method accepts any callable returning a boolean. You can register additional checks for services like Redis, Elasticsearch, or custom APIs. The `run_all_checks()` method executes all registered checks sequentially and aggregates results, as defined in [`codebase_rag/tools/health_checker.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/tools/health_checker.py).

### What happens if a health check fails during `StackManager.start()`?

The `wait_healthy()` method in [`codebase_rag/stack/manager.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/stack/manager.py) raises a runtime exception identifying the failing service and the accumulated error details. This prevents the application from starting in a degraded state. The exception message includes which specific check failed (Memgraph or Qdrant) and the final error encountered during the timeout period.

### Are there unit tests for the health checker implementation?

Yes. The test suite in [`codebase_rag/tests/test_health_checker.py`](https://github.com/vitali87/code-graph-rag/blob/main/codebase_rag/tests/test_health_checker.py) validates the `HealthChecker` class behavior, including successful check registration, proper result aggregation, and timeout handling. These tests ensure the health verification logic remains reliable across code changes.