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

The HealthChecker class in 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 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
add_check(name, fn)	Registers a named check function	codebase_rag/tools/health_checker.py
run_all_checks()	Executes all registered checks sequentially	codebase_rag/tools/health_checker.py
wait_for_memgraph()	TCP/Bolt handshake validation for Memgraph	codebase_rag/stack/health.py
wait_for_qdrant()	HTTP GET validation for Qdrant REST API	codebase_rag/stack/health.py
wait_healthy()	Orchestrates blocking startup verification	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#L12-L30) validates the graph database through TCP socket connection followed by the Bolt protocol handshake.

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#L31-L48) validates the vector store through HTTP health endpoint polling.

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):

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#L13-L27), which internally calls wait_healthy():

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:

# 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
• 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, 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.

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

The wait_healthy() method in 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}