From b1281155011df7efc8b13c6e7d23a3b762fcb339 Mon Sep 17 00:00:00 2001 From: jr Date: Sun, 13 Sep 2026 14:59:41 +0200 Subject: [PATCH 1/4] docs: plan for #114 - serve health checks during cache warmup --- docs/plans/114-readiness-during-warmup.md | 274 ++++++++++++++++++++++ 1 file changed, 274 insertions(+) create mode 100644 docs/plans/114-readiness-during-warmup.md diff --git a/docs/plans/114-readiness-during-warmup.md b/docs/plans/114-readiness-during-warmup.md new file mode 100644 index 0000000..48ff82e --- /dev/null +++ b/docs/plans/114-readiness-during-warmup.md @@ -0,0 +1,274 @@ +# Implementation plan — #114: serve health checks during cache warmup + +**Issue:** [#114](https://github.com/DataZooDE/flapi/issues/114) — cache warmup blocks HTTP +server startup, making the service undeployable on platforms with a capped health-check window. + +**Branch:** `fix/114-readiness-during-warmup` + +**Method:** strict red/green TDD. Every step below names the failing test to write *first*. +Definition of done is the **integration** suite (`test/integration/`, real binary over real HTTP), +not unit tests alone. + +--- + +## 1. The defect, verified in source + +``` +main() src/main.cpp + └─ initializeDatabase(config) + └─ DatabaseManager::initializeDBManagerFromConfig() database_manager.cpp:116 lock_guard(db_mutex) + └─ cache_manager->warmUpCaches(config_manager) database_manager.cpp:170 ← still under the lock + ... + └─ std::thread unified_server_thread main.cpp ← listening socket opens ONLY here +``` + +Consequences: + +1. The socket does not open until every cache-enabled endpoint has finished warming. A platform + health check has nothing to connect to, so a warmup longer than the platform's maximum window + makes the service **permanently undeployable** — every attempt fails and rolls back. +2. There is **no general health endpoint**. `/api/v1/_config/health` requires `--config-service`; + `/mcp/health` is MCP-only. Operators are forced to point health checks at a data endpoint such + as `/doc`, which is exactly what cannot answer during warmup. +3. `warmUpCaches()` is a sequential loop, so the startup cost is the **sum** of all caches. + +## 2. What this PR changes, and what it explicitly does not + +**Does:** open the listening socket *before* warmup; add an always-on health endpoint that +distinguishes `starting` from `ready`; return `503 + Retry-After` on data endpoints whose cache is +not yet populated; run warmup on a background thread; prevent the new concurrency hazards that +opening the socket early creates. + +**Does not:** parallelise warmup across caches. That is a separate change with its own risks +(see §9) and it would not fix #114 anyway — the reporter has a single slow cache, and parallelism +across caches cannot help that. Keep the two apart so this PR stays reviewable. + +**Preserves:** an endpoint never serves from a half-built cache. Answering requests early with +empty results would be a worse bug than the one being fixed. + +--- + +## 3. Design + +### 3.1 Startup order + +``` +initializeDatabase() // NO warmup inside; connections/extensions only +create APIServer +start unified_server_thread // socket opens here, immediately +start warmup thread // CacheManager::warmUpCachesAsync() +join server thread +``` + +`warmUpCaches()` keeps its current synchronous signature and behaviour; a new +`warmUpCachesAsync()` wraps it on a `std::thread`. Moving the call out of +`initializeDBManagerFromConfig()` also takes it out from under `db_mutex` (`database_manager.cpp:116`), +which is a prerequisite for anything that later runs concurrently — a warmup worker reaching +`getConnection()` (`:346`, reachable via `vfs_adapter.cpp:252` when `template.path` is remote) +would otherwise deadlock against the lock held by the main thread. + +### 3.2 Readiness state + +New `CacheReadiness` owned by `CacheManager`, guarded by its own mutex: + +| state | meaning | +|---|---| +| `starting` | warmup thread running, this cache not yet built | +| `ready` | cache populated successfully | +| `failed` | this cache's build failed (message retained) | + +Keyed by `(catalog, schema, table)`. Endpoints without a cache block are always `ready`. +An endpoint that *reads* cache tables owned by other endpoints is ready only when all of them are +— otherwise `on-error: continue` reproduces #114's symptom: a healthy-looking endpoint answering +nothing. + +### 3.3 Health endpoint + +`GET /health` — always registered, no auth, no config service required. + +``` +200 {"status":"ready", "caches":{"total":4,"ready":4,"failed":0},"uptime_s":31} +503 {"status":"starting","caches":{"total":4,"ready":1,"failed":0},"uptime_s":7, + "pending":["norm_product_part","norm_part_frag","norm_locale"]} +503 {"status":"degraded","caches":{"total":4,"ready":3,"failed":1}, + "failed":[{"table":"norm_part_frag","error":"..."}]} +``` + +Rationale for 503 while starting: platforms treat 2xx as healthy. A deployment should not be +declared healthy before it can serve. What #114 needs is that the check gets an *answer* — a +refused connection is indistinguishable from a crash, a 503 is not. Operators who want the +deployment to succeed before caches finish can point the platform check at +`GET /health/live` (below) instead. + +`GET /health/live` — liveness only. `200` as soon as the process is up, regardless of cache state. +This is the endpoint Rita's App Runner config should use; it is what makes the deployment succeed. + +### 3.4 Data endpoints during warmup + +A request to an endpoint whose cache is not `ready`: + +``` +503 Service Unavailable +Retry-After: 5 +{"error":"cache_warming","message":"Cache for this endpoint is still being built", + "table":"norm_product_part"} +``` + +Never a partial or empty result set. `failed` caches also return 503, with the error message, so a +broken cache is visible rather than silently empty. + +### 3.5 New concurrency hazards created by opening the socket early + +These do not exist today and are introduced by this change, so they are in scope: + +1. **`HeartbeatWorker` collides with warmup.** `heartbeat_worker.cpp:81` calls + `refreshCache()` on its own thread. Once the socket opens before warmup finishes, a scheduled + refresh can issue a second `CREATE OR REPLACE TABLE` against a table a warmup worker is + building. **Fix:** an in-flight registry keyed by `(catalog, schema, table)`, entered inside + `CacheManager::refreshCache()` (not `warmUpCaches()`, since the heartbeat calls the former + directly). A duplicate concurrent refresh is a logged no-op. +2. **Request threads read readiness while the warmup thread writes it.** Guard `CacheReadiness` + with a mutex; keep the critical section to a map lookup. +3. **Warmup failure must not kill the process.** `refreshDuckLakeCache()` rethrows and nothing + catches it; on the main thread today that hits `std::set_terminate` → `abort()`. On a + background thread an escaping exception terminates the process with no log line. The warmup + thread body must be `noexcept` at the boundary: catch, record `failed` + message, continue to + the next cache. + +--- + +## 4. TDD sequence — write each test first, watch it fail, then implement + +### Step 1 — liveness endpoint exists (C++ unit) +**Red:** `test/cpp/health_endpoint_test.cpp` — `GET /health/live` returns 200 on a server with no +caches configured. Fails: route does not exist. +**Green:** register the route in `APIServer`. + +### Step 2 — health reports cache counts (C++ unit) +**Red:** with two cache-enabled endpoints and a stub `CacheManager` reporting one ready, `GET /health` +returns 503, `status=starting`, `caches.ready=1`, `caches.total=2`. +**Green:** implement `CacheReadiness` + the `/health` handler. + +### Step 3 — readiness transitions (C++ unit) +**Red:** `test/cpp/cache_readiness_test.cpp` — `starting → ready` on success; `starting → failed` +with the message on throw; unknown table defaults to `ready` (no cache block). +**Green:** implement the state map. + +### Step 4 — warmup failure does not propagate (C++ unit) +**Red:** a stub adapter whose refresh throws; assert `warmUpCaches()` returns normally, marks that +cache `failed`, and still processes the *next* endpoint. +**Green:** wrap the per-endpoint call in try/catch. + +### Step 5 — data endpoint 503s while warming (C++ unit) +**Red:** request an endpoint whose cache is `starting` → 503, `Retry-After` present, body +`error=cache_warming`. Assert the query was **not** executed. +**Green:** readiness check in the request path before query execution. + +### Step 6 — in-flight registry (C++ unit) +**Red:** `test/cpp/cache_inflight_test.cpp` — two concurrent `refreshCache()` calls for the same +`(catalog, schema, table)`; assert the adapter executes **once** and the second returns a no-op. +Then assert two *different* tables both execute. +**Green:** implement the registry inside `refreshCache()`. + +### Step 7 — socket opens before warmup (INTEGRATION — the test that proves #114 is fixed) +**Red:** `test/integration/test_warmup_readiness.py` + +```python +def test_health_answers_during_warmup(slow_cache_server): + # fixture: real flapi binary, config with a cache-populate template that + # takes ~15s (e.g. a generate_series + heavy aggregate), started in background + t0 = time.time() + deadline = t0 + 10 # far below the cache build time + while time.time() < deadline: + try: + r = requests.get(f"{base}/health/live", timeout=1) + assert r.status_code == 200 + assert time.time() - t0 < 10 # answered long before warmup finished + return + except requests.ConnectionError: + time.sleep(0.25) + pytest.fail("server never accepted a connection during warmup") +``` + +Fails today with `ConnectionError` for the whole window — the exact production symptom. + +### Step 8 — data endpoint 503 then 200 (INTEGRATION) +**Red:** against the same fixture, poll the data endpoint: assert it returns **503 with +`Retry-After`** while `/health` reports `starting`, and **200 with correct data** once `/health` +reports `ready`. Assert it never returns 200 with an empty result. + +### Step 9 — health reflects a failed cache (INTEGRATION) +**Red:** a config whose cache template is deliberately invalid SQL. Assert the process stays up, +`/health/live` is 200, `/health` is 503 `degraded` naming the failed table, and the endpoint +returns 503 with the error — not 200-empty, and not a crashed container. + +### Step 10 — heartbeat does not collide (INTEGRATION) +**Red:** a short `schedule` on a slow cache so a scheduled refresh fires during warmup. Assert no +error in the log, the cache ends `ready`, and the adapter ran the populate once. + +--- + +## 5. Definition of done + +- [ ] Steps 7–10 (integration) pass against a **real binary over real HTTP**; steps 1–6 green. +- [ ] `make test-all` passes — no regressions in the existing C++ and integration suites. +- [ ] A server with **no** cache-enabled endpoints behaves exactly as before (timing and routes). +- [ ] `/health/live` answers within 2 s of process start on the slow-cache fixture. +- [ ] No endpoint ever returns 200 with an empty body due to an unbuilt cache. +- [ ] Warmup failure leaves the process running and surfaces in `/health`. +- [ ] `docs/CONFIG_REFERENCE.md` and `docs/CLI_REFERENCE.md` document `/health`, `/health/live`, + and the 503 contract, with an App Runner example pointing at `/health/live`. +- [ ] `CHANGELOG.md` entry. + +--- + +## 6. Files expected to change + +| File | Change | +|---|---| +| `src/main.cpp` | start server thread before warmup; launch warmup thread | +| `src/database_manager.cpp` | remove `warmUpCaches()` from `initializeDBManagerFromConfig()` (out from under `db_mutex`) | +| `src/cache_manager.cpp/.hpp` | `warmUpCachesAsync()`, readiness state, in-flight registry, per-endpoint try/catch | +| `src/api_server.cpp/.hpp` | register `/health`, `/health/live` | +| `src/request_handler.cpp` | readiness gate before query execution | +| `src/heartbeat_worker.cpp` | honour the in-flight registry | +| `test/cpp/*` | steps 1–6 | +| `test/integration/test_warmup_readiness.py` + fixture config | steps 7–10 | +| `docs/`, `CHANGELOG.md` | as above | + +--- + +## 7. Constraints that must not be violated + +1. **Never serve a partial cache.** 503 is the only acceptable answer for an unready endpoint. +2. **Never let an exception escape the warmup thread.** +3. **Do not modify `db_mutex`'s scope in this PR.** Moving the warmup call out from under it is + sufficient here; narrowing the lock is a separate change with its own review. +4. **`concurrency` stays 1.** No parallel warmup in this PR. +5. **No behaviour change when no caches are configured.** +6. Endpoints without a `cache:` block must not acquire any new lock on the request path. + +--- + +## 8. Risks + +| Risk | Mitigation | +|---|---| +| Health check returns 200 too early; platform routes traffic to an unready service | `/health` is 503 until ready; only `/health/live` is unconditionally 200, and it is documented as liveness | +| Readiness lookup on every request adds latency | map lookup under a short-held mutex; skip entirely for endpoints without a cache block | +| Heartbeat/warmup collision corrupts a cache | in-flight registry (step 6) | +| Background warmup hides failures | `/health` reports `degraded` with the message; failures also logged at ERROR | +| Existing deployments depending on "socket closed until ready" as a readiness signal | documented in CHANGELOG as a behaviour change; `/health` preserves the semantics for anyone who needs them | + +--- + +## 9. Follow-up, explicitly out of scope + +- Parallel warmup (`cache.warmup.concurrency`). Bounded by `sum(tᵢ)/max(tᵢ)`, which is ≈1.0 for + single-cache deployments; needs the in-flight registry from this PR, a conservative default of + 1, and care with memory — measured single builds have peaked at 1.9–3.6× their configured + `memory_limit`. +- Narrowing `db_mutex` to the `db` handle lifecycle. +- `recordSyncEvent()` builds its INSERT by string concatenation with only a quote-swap on + `message`, despite a prepared-statement template being written and discarded + (`cache_manager.cpp:309-329`). From 34d5af76a7e333c01c582cbe128f78e1edd9c4b4 Mon Sep 17 00:00:00 2001 From: jr Date: Sun, 13 Sep 2026 15:40:18 +0200 Subject: [PATCH 2/4] fix(#114): serve health checks during cache warmup Open the HTTP listener before cache warmup and run warmup on a background thread, so a slow cache can no longer make a deployment fail its platform health check. - add /health/live (liveness, always 200) and /health (readiness, 503 until every cache is built); previously the only health routes were behind --config-service or MCP-only - track cache readiness per (catalog, schema, table); cached endpoints return 503 + Retry-After until ready, never a partial or empty result - contain warmup failures: the process stays up and /health reports degraded - suppress duplicate in-flight refreshes for the same table, so the heartbeat worker cannot collide with warmup now that the socket opens earlier warmUpCaches() moves out of initializeDBManagerFromConfig(), which also takes it out from under db_mutex. The lock's scope is unchanged. Closes #114 --- CHANGELOG.md | 15 ++ docs/CLI_REFERENCE.md | 61 ++++- docs/CONFIG_REFERENCE.md | 33 ++- src/api_server.cpp | 67 ++++- src/cache_manager.cpp | 135 +++++++++- src/database_manager.cpp | 2 +- src/include/api_server.hpp | 5 +- src/include/cache_manager.hpp | 56 +++- src/main.cpp | 9 + src/request_handler.cpp | 24 ++ test/cpp/cache_manager_test.cpp | 150 +++++++++++ test/cpp/request_handler_test.cpp | 86 +++++++ test/integration/test_warmup_readiness.py | 300 ++++++++++++++++++++++ 13 files changed, 929 insertions(+), 14 deletions(-) create mode 100644 test/integration/test_warmup_readiness.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 526be8f..e8fea22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,21 @@ All notable changes to flAPI are documented here. Versions follow `vYY.MM.DD` (the date the binary set was cut). Earlier history is in the git log. +## Unreleased + +### Health checks during cache warmup + +- The HTTP listener now opens before cache warmup completes, so platforms can connect during long + startup cache builds. +- Added always-on `GET /health/live` for liveness and `GET /health` for readiness. `/health` + returns `503` while caches are still starting or degraded. +- Cache-enabled data endpoints now return `503` with `Retry-After: 5` while their cache is starting + or failed, instead of risking a `200` response from a half-built cache. +- Cache warmup failures are captured per endpoint and surfaced in health output; warmup continues + to later caches. +- Concurrent refreshes for the same DuckLake cache table are suppressed so scheduler refreshes do + not collide with startup warmup. + ## v26.05.18 — Prepared-statement coverage swept across every code path Follow-up to v26.05.17. After v26.05.17 shipped, an internal audit found that the prepared-statement path was only wired into the GET endpoint executor — POST/PUT/PATCH writes and the Arrow-streaming endpoint still rendered Mustache templates as strings. This release closes that gap. diff --git a/docs/CLI_REFERENCE.md b/docs/CLI_REFERENCE.md index 1aa6bd0..1c2be60 100644 --- a/docs/CLI_REFERENCE.md +++ b/docs/CLI_REFERENCE.md @@ -31,8 +31,9 @@ This document provides a complete reference for the `flapi` server executable's - [Development Mode](#development-mode) - [Production Mode](#production-mode) - [CI/CD Validation](#cicd-validation) -6. [Signal Handling](#6-signal-handling) -7. [Exit Codes](#7-exit-codes) +6. [Runtime Health Endpoints](#6-runtime-health-endpoints) +7. [Signal Handling](#7-signal-handling) +8. [Exit Codes](#8-exit-codes) - [Related Documentation](#related-documentation) --- @@ -636,6 +637,15 @@ See [Configuration Reference - Environment Variables](./CONFIG_REFERENCE.md#10-e --log-level warning ``` +For platforms with short startup health-check windows, point the platform liveness check at: + +```text +/health/live +``` + +Use `/health` when the platform should wait until cache warmup is complete before marking the +deployment ready for traffic. + ### CI/CD Validation ```bash @@ -654,7 +664,50 @@ fi --- -## 6. Signal Handling +## 6. Runtime Health Endpoints + +flAPI always registers two unauthenticated health endpoints. They do not require +`--config-service`. + +| Endpoint | Purpose | Response | +|----------|---------|----------| +| `GET /health/live` | Liveness: process is up and the listener is accepting connections | `200 {"status":"live"}` | +| `GET /health` | Readiness: cache-enabled endpoints are ready to serve | `200` when ready, `503` while starting or degraded | + +During cache warmup, `/health/live` returns `200` as soon as the server is listening, while +`/health` returns: + +```json +{ + "status": "starting", + "caches": {"total": 4, "ready": 1, "failed": 0}, + "pending": [{"schema": "cache", "table": "customers_cache"}], + "uptime_s": 7 +} +``` + +If a cache warmup fails, `/health` returns `503` with `"status": "degraded"` and a `failed` list +including the table name and error. + +Requests to a cached data endpoint while its cache is starting or failed return `503 Service +Unavailable` with `Retry-After: 5` and a JSON body containing `"error": "cache_warming"`. flAPI +does not serve partial or empty results from a half-built cache. + +**AWS App Runner example:** + +```yaml +HealthCheckConfiguration: + Protocol: HTTP + Path: /health/live + Interval: 5 + Timeout: 2 + HealthyThreshold: 1 + UnhealthyThreshold: 5 +``` + +--- + +## 7. Signal Handling | Signal | Behavior | |--------|----------| @@ -679,7 +732,7 @@ On receiving a shutdown signal, the server: --- -## 7. Exit Codes +## 8. Exit Codes | Code | Description | |------|-------------| diff --git a/docs/CONFIG_REFERENCE.md b/docs/CONFIG_REFERENCE.md index dbfb8d5..8c5ffbe 100644 --- a/docs/CONFIG_REFERENCE.md +++ b/docs/CONFIG_REFERENCE.md @@ -1164,7 +1164,34 @@ cache: schedule: 5m ``` -### 6.2 Refresh Modes +### 6.2 Cache Readiness During Warmup + +At startup, flAPI opens the HTTP listener before cache warmup finishes. Cache-enabled endpoints are +not allowed to serve until their configured cache table is fully built. + +While a cache is still building, requests to that endpoint return: + +```http +503 Service Unavailable +Retry-After: 5 +Content-Type: application/json +``` + +```json +{ + "error": "cache_warming", + "message": "Cache for this endpoint is still being built", + "table": "customers_cache" +} +``` + +If warmup fails, the endpoint continues to return `503` and includes the failure detail. This is +intentional: a cached endpoint must never return `200` with empty or partial results from a +half-built cache. + +Endpoints without a `cache:` block are not gated by cache readiness. + +### 6.3 Refresh Modes **Full Refresh (Default):** @@ -1217,7 +1244,7 @@ cache: type: timestamp ``` -### 6.3 Retention Policies +### 6.4 Retention Policies | Parameter | Type | Default | Description | |-----------|------|---------|-------------| @@ -1244,7 +1271,7 @@ cache: delete-handling: soft ``` -### 6.4 Cache Template Variables +### 6.5 Cache Template Variables Special variables available in cache-enabled SQL templates: diff --git a/src/api_server.cpp b/src/api_server.cpp index 25d68e7..f004de4 100644 --- a/src/api_server.cpp +++ b/src/api_server.cpp @@ -20,7 +20,7 @@ APIServer::APIServer(std::shared_ptr cm, std::shared_ptr db_manager, bool config_service_enabled, const std::string& config_service_token) - : configManager(cm), dbManager(db_manager), openAPIDocGenerator(std::make_shared(cm, db_manager)), requestHandler(dbManager, cm) + : configManager(cm), dbManager(db_manager), openAPIDocGenerator(std::make_shared(cm, db_manager)), requestHandler(dbManager, cm), startedAt(std::chrono::steady_clock::now()) { // Initialize MCP session manager mcpSessionManager = std::make_shared(); @@ -99,6 +99,18 @@ void APIServer::setupRoutes() { return crow::response(200, "text/plain", logo); }); + CROW_ROUTE(app, "/health/live") + .methods("GET"_method) + ([this]() { + return getLiveHealth(); + }); + + CROW_ROUTE(app, "/health") + .methods("GET"_method) + ([this]() { + return getHealth(); + }); + configService->setDocGenerator(openAPIDocGenerator); configService->registerRoutes(app); @@ -294,6 +306,59 @@ crow::response APIServer::refreshConfig() { } } +crow::response APIServer::getLiveHealth() { + crow::json::wvalue health; + health["status"] = "live"; + return crow::response(200, health); +} + +crow::response APIServer::getHealth() { + crow::json::wvalue health; + const auto uptime = std::chrono::duration_cast( + std::chrono::steady_clock::now() - startedAt); + health["uptime_s"] = static_cast(uptime.count()); + + CacheManager::CacheReadinessSummary summary; + auto cache_manager = getCacheManager(); + if (cache_manager) { + summary = cache_manager->getReadinessSummary(); + } + + health["caches"]["total"] = summary.total; + health["caches"]["ready"] = summary.ready; + health["caches"]["failed"] = summary.failed; + + if (summary.failed > 0) { + health["status"] = "degraded"; + std::vector failed; + for (const auto& cache : summary.failed_caches) { + crow::json::wvalue item; + item["table"] = cache.table; + item["schema"] = cache.schema; + item["error"] = cache.error; + failed.push_back(std::move(item)); + } + health["failed"] = std::move(failed); + return crow::response(503, health); + } + + if (summary.ready < summary.total) { + health["status"] = "starting"; + std::vector pending; + for (const auto& cache : summary.pending_caches) { + crow::json::wvalue item; + item["table"] = cache.table; + item["schema"] = cache.schema; + pending.push_back(std::move(item)); + } + health["pending"] = std::move(pending); + return crow::response(503, health); + } + + health["status"] = "ready"; + return crow::response(200, health); +} + crow::response APIServer::generateOpenAPIDoc() { YAML::Node doc = openAPIDocGenerator->generateDoc(app); diff --git a/src/cache_manager.cpp b/src/cache_manager.cpp index 2bb4b1e..fe21e92 100644 --- a/src/cache_manager.cpp +++ b/src/cache_manager.cpp @@ -5,6 +5,8 @@ #include #include #include +#include +#include #include "cache_manager.hpp" #include "database_manager.hpp" @@ -12,6 +14,10 @@ namespace flapi { +bool CacheManager::CacheKey::operator<(const CacheKey& other) const { + return std::tie(catalog, schema, table) < std::tie(other.catalog, other.schema, other.table); +} + CacheManager::CacheManager(std::shared_ptr db_manager) : db_adapter_(std::make_shared(db_manager)), db_manager(db_manager) {} @@ -22,6 +28,7 @@ CacheManager::CacheManager(std::shared_ptr db_adapter) void CacheManager::warmUpCaches(std::shared_ptr config_manager) { CROW_LOG_INFO << "Warming up endpoint caches, this might take some time..."; + initializeReadiness(config_manager); // Initialize audit tables first initializeAuditTables(config_manager); @@ -33,12 +40,36 @@ void CacheManager::warmUpCaches(std::shared_ptr config_manager) { { // Warmup: refresh caches only for endpoints with cache enabled and a table defined if (endpoint.cache.enabled && !endpoint.cache.table.empty()) { - refreshCache(config_manager, endpoint, params); + markCacheStarting(config_manager, endpoint); + try { + if (refreshCache(config_manager, endpoint, params)) { + markCacheReady(config_manager, endpoint); + } + } catch (const std::exception& ex) { + CROW_LOG_ERROR << "Cache warmup failed for " << endpoint.cache.table << ": " << ex.what(); + markCacheFailed(config_manager, endpoint, ex.what()); + } catch (...) { + CROW_LOG_ERROR << "Cache warmup failed for " << endpoint.cache.table << ": unknown error"; + markCacheFailed(config_manager, endpoint, "unknown error"); + } } } CROW_LOG_INFO << "Finished warming up endpoint caches! Let's go!"; } +std::thread CacheManager::warmUpCachesAsync(std::shared_ptr config_manager) { + initializeReadiness(config_manager); + return std::thread([this, config_manager]() { + try { + warmUpCaches(config_manager); + } catch (const std::exception& ex) { + CROW_LOG_ERROR << "Unexpected cache warmup failure: " << ex.what(); + } catch (...) { + CROW_LOG_ERROR << "Unexpected cache warmup failure: unknown error"; + } + }); +} + bool CacheManager::shouldRefreshCache(std::shared_ptr config_manager, const EndpointConfig& endpoint) { // Do not refresh cache on regular request execution path. // Refreshes happen during warmup, scheduled tasks, or explicit manual triggers. @@ -55,8 +86,106 @@ bool CacheManager::shouldRefreshCache(std::shared_ptr config_mana return false; } -void CacheManager::refreshCache(std::shared_ptr config_manager, const EndpointConfig& endpoint, std::map& params) { - refreshDuckLakeCache(config_manager, endpoint, params); +bool CacheManager::refreshCache(std::shared_ptr config_manager, const EndpointConfig& endpoint, std::map& params) { + const CacheKey key = cacheKeyForEndpoint(config_manager, endpoint); + if (!enterRefresh(key)) { + CROW_LOG_INFO << "Skipping duplicate in-flight cache refresh for " << key.schema << "." << key.table; + return false; + } + + try { + refreshDuckLakeCache(config_manager, endpoint, params); + markCacheReady(config_manager, endpoint); + leaveRefresh(key); + return true; + } catch (...) { + leaveRefresh(key); + throw; + } +} + +CacheManager::CacheKey CacheManager::cacheKeyForEndpoint(std::shared_ptr config_manager, const EndpointConfig& endpoint) { + const auto& ducklakeConfig = config_manager->getDuckLakeConfig(); + const auto& cacheConfig = endpoint.cache; + return CacheKey{ + ducklakeConfig.alias, + cacheConfig.schema.empty() ? "main" : cacheConfig.schema, + cacheConfig.table + }; +} + +void CacheManager::initializeReadiness(std::shared_ptr config_manager) { + std::lock_guard lock(readiness_mutex_); + readiness_.clear(); + for (const auto& endpoint : config_manager->getEndpoints()) { + if (!endpoint.cache.enabled || endpoint.cache.table.empty()) { + continue; + } + const CacheKey key = cacheKeyForEndpoint(config_manager, endpoint); + readiness_[key] = CacheReadiness{ReadinessState::Starting, key.catalog, key.schema, key.table, ""}; + } +} + +void CacheManager::markCacheStarting(std::shared_ptr config_manager, const EndpointConfig& endpoint) { + const CacheKey key = cacheKeyForEndpoint(config_manager, endpoint); + std::lock_guard lock(readiness_mutex_); + readiness_[key] = CacheReadiness{ReadinessState::Starting, key.catalog, key.schema, key.table, ""}; +} + +void CacheManager::markCacheReady(std::shared_ptr config_manager, const EndpointConfig& endpoint) { + const CacheKey key = cacheKeyForEndpoint(config_manager, endpoint); + std::lock_guard lock(readiness_mutex_); + readiness_[key] = CacheReadiness{ReadinessState::Ready, key.catalog, key.schema, key.table, ""}; +} + +void CacheManager::markCacheFailed(std::shared_ptr config_manager, const EndpointConfig& endpoint, const std::string& error) { + const CacheKey key = cacheKeyForEndpoint(config_manager, endpoint); + std::lock_guard lock(readiness_mutex_); + readiness_[key] = CacheReadiness{ReadinessState::Failed, key.catalog, key.schema, key.table, error}; +} + +CacheManager::CacheReadiness CacheManager::getReadinessForKey(const CacheKey& key) const { + std::lock_guard lock(readiness_mutex_); + auto it = readiness_.find(key); + if (it == readiness_.end()) { + return CacheReadiness{ReadinessState::Ready, key.catalog, key.schema, key.table, ""}; + } + return it->second; +} + +CacheManager::CacheReadiness CacheManager::getEndpointReadiness(std::shared_ptr config_manager, const EndpointConfig& endpoint) const { + if (!endpoint.cache.enabled || endpoint.cache.table.empty()) { + return CacheReadiness{ReadinessState::Ready, "", "", "", ""}; + } + return getReadinessForKey(cacheKeyForEndpoint(config_manager, endpoint)); +} + +CacheManager::CacheReadinessSummary CacheManager::getReadinessSummary() const { + std::lock_guard lock(readiness_mutex_); + CacheReadinessSummary summary; + summary.total = static_cast(readiness_.size()); + for (const auto& [key, readiness] : readiness_) { + if (readiness.state == ReadinessState::Ready) { + ++summary.ready; + } else if (readiness.state == ReadinessState::Failed) { + ++summary.failed; + summary.failed_caches.push_back(readiness); + } else { + summary.pending_caches.push_back(readiness); + } + } + return summary; +} + +bool CacheManager::enterRefresh(const CacheKey& key) { + std::lock_guard lock(inflight_mutex_); + auto [it, inserted] = inflight_refreshes_.insert(key); + return inserted; +} + +void CacheManager::leaveRefresh(const CacheKey& key) { + std::lock_guard lock(inflight_mutex_); + inflight_refreshes_.erase(key); } void CacheManager::refreshDuckLakeCache(std::shared_ptr config_manager, const EndpointConfig& endpoint, std::map params) { diff --git a/src/database_manager.cpp b/src/database_manager.cpp index 70cae88..bc7e731 100644 --- a/src/database_manager.cpp +++ b/src/database_manager.cpp @@ -167,7 +167,7 @@ void DatabaseManager::initializeDBManagerFromConfig(std::shared_ptr(config_manager); cache_manager = std::make_unique(shared_from_this()); - cache_manager->warmUpCaches(config_manager); + cache_manager->initializeReadiness(config_manager); } } diff --git a/src/include/api_server.hpp b/src/include/api_server.hpp index 5e73e43..af25b97 100644 --- a/src/include/api_server.hpp +++ b/src/include/api_server.hpp @@ -41,6 +41,8 @@ class APIServer crow::response getConfig(); crow::response refreshConfig(); + crow::response getLiveHealth(); + crow::response getHealth(); void run(int port = 8080); void stop(); @@ -69,6 +71,7 @@ class APIServer std::shared_ptr mcpSessionManager; std::shared_ptr mcpCapabilitiesDetector; RequestHandler requestHandler; + std::chrono::steady_clock::time_point startedAt; }; -} // namespace flapi \ No newline at end of file +} // namespace flapi diff --git a/src/include/cache_manager.hpp b/src/include/cache_manager.hpp index fec706a..a02c2be 100644 --- a/src/include/cache_manager.hpp +++ b/src/include/cache_manager.hpp @@ -3,6 +3,11 @@ #include #include #include +#include +#include +#include +#include +#include #include "config_manager.hpp" #include "cache_database_adapter.hpp" #include @@ -20,15 +25,45 @@ class CacheManager { explicit CacheManager(std::shared_ptr db_adapter); void warmUpCaches(std::shared_ptr config_manager); + std::thread warmUpCachesAsync(std::shared_ptr config_manager); bool shouldRefreshCache(std::shared_ptr config_manager, const EndpointConfig& endpoint); bool shouldRefreshCache(std::shared_ptr config_manager, const CacheConfig& cacheConfig); - void refreshCache(std::shared_ptr config_manager, const EndpointConfig& endpoint, std::map& params); + bool refreshCache(std::shared_ptr config_manager, const EndpointConfig& endpoint, std::map& params); void refreshDuckLakeCache(std::shared_ptr config_manager, const EndpointConfig& endpoint, std::map params); static std::string joinStrings(const std::vector& values, const std::string& delimiter); void addQueryCacheParamsIfNecessary(std::shared_ptr config_manager, const EndpointConfig& endpoint, std::map& params); void performGarbageCollection(std::shared_ptr config_manager, const EndpointConfig& endpoint, const std::vector previousTableNames); + enum class ReadinessState { + Starting, + Ready, + Failed + }; + + struct CacheReadiness { + ReadinessState state = ReadinessState::Ready; + std::string catalog; + std::string schema; + std::string table; + std::string error; + }; + + struct CacheReadinessSummary { + int total = 0; + int ready = 0; + int failed = 0; + std::vector pending_caches; + std::vector failed_caches; + }; + + void initializeReadiness(std::shared_ptr config_manager); + void markCacheStarting(std::shared_ptr config_manager, const EndpointConfig& endpoint); + void markCacheReady(std::shared_ptr config_manager, const EndpointConfig& endpoint); + void markCacheFailed(std::shared_ptr config_manager, const EndpointConfig& endpoint, const std::string& error); + CacheReadiness getEndpointReadiness(std::shared_ptr config_manager, const EndpointConfig& endpoint) const; + CacheReadinessSummary getReadinessSummary() const; + // Audit functionality void initializeAuditTables(std::shared_ptr config_manager); void ensureCacheSchemaExists(const std::string& catalog, const std::string& schema); @@ -45,11 +80,30 @@ class CacheManager { SnapshotInfo fetchSnapshotInfo(const std::string& catalog, const std::string& schema, const std::string& table); static std::string determineCacheMode(const CacheConfig& cacheConfig); + struct CacheKey { + std::string catalog; + std::string schema; + std::string table; + + bool operator<(const CacheKey& other) const; + }; + + static CacheKey cacheKeyForEndpoint(std::shared_ptr config_manager, const EndpointConfig& endpoint); + CacheReadiness getReadinessForKey(const CacheKey& key) const; + bool enterRefresh(const CacheKey& key); + void leaveRefresh(const CacheKey& key); + // Database adapter for cache operations (mockable for testing) std::shared_ptr db_adapter_; // Keep db_manager for backward compatibility with code that accesses it directly std::shared_ptr db_manager; + + mutable std::mutex readiness_mutex_; + std::map readiness_; + + mutable std::mutex inflight_mutex_; + std::set inflight_refreshes_; }; } // namespace flapi diff --git a/src/main.cpp b/src/main.cpp index d6fc39f..c966c80 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -650,6 +650,11 @@ int main(int argc, char* argv[]) CROW_LOG_INFO << "flAPI unified server started - REST API and MCP on port " << config_manager->getHttpPort(); + std::thread warmup_thread; + if (auto cache_manager = DatabaseManager::getInstance()->getCacheManager()) { + warmup_thread = cache_manager->warmUpCachesAsync(config_manager); + } + // Once-a-day feedback nudge, at the point the server is actually up. Needs // both streams to be terminals, so a container or systemd start -- how this // runs in production -- prints nothing. The log line above remains the @@ -676,6 +681,10 @@ int main(int argc, char* argv[]) // Wait for server to finish unified_server_thread.join(); + if (warmup_thread.joinable()) { + warmup_thread.join(); + } + // Drain buffered telemetry on clean exit; the signal path already flushed. if (!should_exit) { flapi::GlobalTelemetry().flush(); diff --git a/src/request_handler.cpp b/src/request_handler.cpp index 3458659..6aa13ce 100644 --- a/src/request_handler.cpp +++ b/src/request_handler.cpp @@ -211,6 +211,30 @@ void RequestHandler::handleGetRequest(const crow::request& req, crow::response& return; } + if (endpoint.cache.enabled && !endpoint.cache.table.empty()) { + auto cache_manager = db_manager->getCacheManager(); + if (cache_manager) { + auto readiness = cache_manager->getEndpointReadiness(config_manager, endpoint); + if (readiness.state != CacheManager::ReadinessState::Ready) { + crow::json::wvalue errorResponse; + errorResponse["error"] = "cache_warming"; + errorResponse["table"] = readiness.table; + if (readiness.state == CacheManager::ReadinessState::Failed) { + errorResponse["message"] = "Cache for this endpoint failed to build"; + errorResponse["detail"] = readiness.error; + } else { + errorResponse["message"] = "Cache for this endpoint is still being built"; + } + res.code = 503; + res.set_header("Content-Type", "application/json"); + res.set_header("Retry-After", "5"); + res.write(errorResponse.dump()); + res.end(); + return; + } + } + } + // Parse pagination parameters int64_t offset = 0; int64_t limit = 100; diff --git a/test/cpp/cache_manager_test.cpp b/test/cpp/cache_manager_test.cpp index ec015ef..131d90b 100644 --- a/test/cpp/cache_manager_test.cpp +++ b/test/cpp/cache_manager_test.cpp @@ -12,6 +12,7 @@ #include #include #include +#include #define private public #include "../../src/include/cache_manager.hpp" #include "../../src/include/query_executor.hpp" @@ -308,3 +309,152 @@ TEST_CASE("CacheManager recordSyncEvent does not throw", "[cache_manager]") { config_manager, endpoint, "full", "success", "Test message")); } } + +TEST_CASE("CacheManager readiness transitions are tracked by table", "[cache_manager][readiness]") { + TempTestConfig temp("cache_readiness"); + temp.writeEndpoint("cached.yaml", R"( +url-path: /cached +method: GET +template-source: cached.sql +connection: [test] +cache: + enabled: true + table: cached_table +)"); + temp.writeSqlTemplate("cached.sql", "SELECT 1"); + auto config_manager = temp.createConfigManager(); + const auto endpoint = config_manager->getEndpoints().front(); + + CacheManager cache_manager(std::shared_ptr(nullptr)); + cache_manager.initializeReadiness(config_manager); + + auto starting = cache_manager.getEndpointReadiness(config_manager, endpoint); + REQUIRE(starting.state == CacheManager::ReadinessState::Starting); + + cache_manager.markCacheReady(config_manager, endpoint); + auto ready = cache_manager.getEndpointReadiness(config_manager, endpoint); + REQUIRE(ready.state == CacheManager::ReadinessState::Ready); + + cache_manager.markCacheFailed(config_manager, endpoint, "boom"); + auto failed = cache_manager.getEndpointReadiness(config_manager, endpoint); + REQUIRE(failed.state == CacheManager::ReadinessState::Failed); + REQUIRE(failed.error == "boom"); + + EndpointConfig uncached; + auto unknown = cache_manager.getEndpointReadiness(config_manager, uncached); + REQUIRE(unknown.state == CacheManager::ReadinessState::Ready); +} + +class SelectiveThrowCacheAdapter : public RecordingCacheAdapter { +public: + std::vector refreshed_tables; + + std::string renderCacheTemplate(const EndpointConfig& endpoint, + const CacheConfig& cacheConfig, + std::map& params) override { + refreshed_tables.push_back(cacheConfig.table); + if (cacheConfig.table == "first_cache") { + throw std::runtime_error("first failed"); + } + return RecordingCacheAdapter::renderCacheTemplate(endpoint, cacheConfig, params); + } +}; + +TEST_CASE("CacheManager warmUpCaches records failures and continues", "[cache_manager][warmup]") { + TempTestConfig temp("cache_warmup_failure"); + temp.writeEndpoint("first.yaml", R"( +url-path: /first +method: GET +template-source: first.sql +connection: [test] +cache: + enabled: true + table: first_cache +)"); + temp.writeSqlTemplate("first.sql", "SELECT 1"); + temp.writeEndpoint("second.yaml", R"( +url-path: /second +method: GET +template-source: second.sql +connection: [test] +cache: + enabled: true + table: second_cache +)"); + temp.writeSqlTemplate("second.sql", "SELECT 2"); + auto config_manager = temp.createConfigManager(); + auto adapter = std::make_shared(); + CacheManager cache_manager(adapter); + + REQUIRE_NOTHROW(cache_manager.warmUpCaches(config_manager)); + + bool saw_failed = false; + bool saw_ready = false; + for (const auto& endpoint : config_manager->getEndpoints()) { + auto readiness = cache_manager.getEndpointReadiness(config_manager, endpoint); + if (endpoint.cache.table == "first_cache") { + saw_failed = true; + REQUIRE(readiness.state == CacheManager::ReadinessState::Failed); + REQUIRE(readiness.error == "first failed"); + } + if (endpoint.cache.table == "second_cache") { + saw_ready = true; + REQUIRE(readiness.state == CacheManager::ReadinessState::Ready); + } + } + REQUIRE(saw_failed); + REQUIRE(saw_ready); + REQUIRE(adapter->refreshed_tables.size() == 2); +} + +class SlowCountingCacheAdapter : public RecordingCacheAdapter { +public: + std::atomic refresh_count{0}; + std::mutex mutex; + + void executeDuckLakeQuery(const std::string& query, + const std::map& params) override { + ++refresh_count; + std::this_thread::sleep_for(std::chrono::milliseconds(150)); + std::lock_guard lock(mutex); + RecordingCacheAdapter::executeDuckLakeQuery(query, params); + } + + QueryResult executeDuckLakeQueryWithResult(const std::string& query) override { + std::lock_guard lock(mutex); + return RecordingCacheAdapter::executeDuckLakeQueryWithResult(query); + } +}; + +TEST_CASE("CacheManager suppresses duplicate in-flight refreshes per table", "[cache_manager][inflight]") { + TempTestConfig temp("cache_inflight"); + auto config_manager = temp.createConfigManager(); + auto adapter = std::make_shared(); + CacheManager cache_manager(adapter); + + EndpointConfig one; + one.urlPath = "/one"; + one.cache.enabled = true; + one.cache.table = "same_cache"; + one.cache.schema = "main"; + std::map params1; + std::map params2; + + std::thread first([&]() { cache_manager.refreshCache(config_manager, one, params1); }); + std::thread second([&]() { cache_manager.refreshCache(config_manager, one, params2); }); + first.join(); + second.join(); + + REQUIRE(adapter->refresh_count.load() == 1); + + EndpointConfig two = one; + two.cache.table = "other_cache"; + std::map params3; + std::map params4; + std::thread third([&]() { cache_manager.refreshCache(config_manager, one, params3); }); + std::thread fourth([&]() { cache_manager.refreshCache(config_manager, two, params4); }); + third.join(); + fourth.join(); + + REQUIRE(adapter->refresh_count.load() == 3); +} diff --git a/test/cpp/request_handler_test.cpp b/test/cpp/request_handler_test.cpp index 1e2b437..9169607 100644 --- a/test/cpp/request_handler_test.cpp +++ b/test/cpp/request_handler_test.cpp @@ -19,10 +19,96 @@ #include #include "test_utils.hpp" +#define private public +#include "../../src/include/api_server.hpp" +#undef private using namespace flapi; using namespace flapi::test; +TEST_CASE("GET /health/live returns 200 without cache configuration", "[health]") { + TempTestConfig temp("health_live"); + auto config_manager = temp.createConfigManager(); + auto db_manager = std::make_shared(); + APIServer server(config_manager, db_manager); + + crow::response res = server.getLiveHealth(); + + REQUIRE(res.code == 200); + auto body = crow::json::load(res.body); + REQUIRE(body); + REQUIRE(std::string(body["status"].s()) == "live"); +} + +TEST_CASE("GET /health reports cache readiness counts", "[health]") { + TempTestConfig temp("health_counts"); + temp.writeEndpoint("one.yaml", R"( +url-path: /one +method: GET +template-source: one.sql +connection: [test] +cache: + enabled: true + table: cache_one +)"); + temp.writeSqlTemplate("one.sql", "SELECT 1"); + temp.writeEndpoint("two.yaml", R"( +url-path: /two +method: GET +template-source: two.sql +connection: [test] +cache: + enabled: true + table: cache_two +)"); + temp.writeSqlTemplate("two.sql", "SELECT 2"); + auto config_manager = temp.createConfigManager(); + auto db_manager = std::make_shared(); + auto cache_manager = std::make_shared(std::shared_ptr(nullptr)); + cache_manager->initializeReadiness(config_manager); + cache_manager->markCacheReady(config_manager, config_manager->getEndpoints().front()); + db_manager->cache_manager = cache_manager; + APIServer server(config_manager, db_manager); + + crow::response res = server.getHealth(); + + REQUIRE(res.code == 503); + auto body = crow::json::load(res.body); + REQUIRE(body); + REQUIRE(std::string(body["status"].s()) == "starting"); + REQUIRE(static_cast(body["caches"]["total"].i()) == 2); + REQUIRE(static_cast(body["caches"]["ready"].i()) == 1); +} + +TEST_CASE("RequestHandler returns 503 while endpoint cache is warming", "[request_handler][cache]") { + TempTestConfig temp("cache_warming_request"); + auto config_manager = temp.createConfigManager(); + auto db_manager = std::make_shared(); + auto cache_manager = std::make_shared(std::shared_ptr(nullptr)); + db_manager->cache_manager = cache_manager; + + EndpointConfig endpoint; + endpoint.urlPath = "/cached"; + endpoint.method = "GET"; + endpoint.cache.enabled = true; + endpoint.cache.table = "cached_table"; + cache_manager->markCacheStarting(config_manager, endpoint); + + RequestHandler handler(db_manager, config_manager); + crow::request req; + req.method = crow::HTTPMethod::Get; + req.url = "/cached"; + crow::response res; + + handler.handleRequest(req, res, endpoint, {}, {}); + + REQUIRE(res.code == 503); + REQUIRE(res.get_header_value("Retry-After") == "5"); + auto body = crow::json::load(res.body); + REQUIRE(body); + REQUIRE(std::string(body["error"].s()) == "cache_warming"); +} + namespace { EndpointConfig createWriteEndpoint() { diff --git a/test/integration/test_warmup_readiness.py b/test/integration/test_warmup_readiness.py new file mode 100644 index 0000000..0fbef73 --- /dev/null +++ b/test/integration/test_warmup_readiness.py @@ -0,0 +1,300 @@ +"""End-to-end readiness coverage for cache warmup (issue #114).""" + +import os +import signal +import socket +import subprocess +import tempfile +import time +from typing import Dict, Iterator, List + +import pytest +import requests + + +def _repo_root() -> str: + return os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) + + +def _flapi_binary() -> str: + build_type = os.getenv("FLAPI_BUILD_TYPE") + candidates: List[str] = [] + if build_type: + candidates.append(os.path.join(_repo_root(), "build", build_type, "flapi")) + candidates.extend( + [ + os.path.join(_repo_root(), "build", "debug", "flapi"), + os.path.join(_repo_root(), "build", "release", "flapi"), + ] + ) + for path in candidates: + if os.path.exists(path): + return path + pytest.skip("flapi binary not found in build/debug or build/release") + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _write_config(dirpath: str, port: int, scheduler_enabled: bool = False, invalid_cache: bool = False) -> str: + sqls = os.path.join(dirpath, "sqls") + os.makedirs(sqls) + + metadata_path = os.path.join(dirpath, "cache.ducklake") + data_path = os.path.join(dirpath, "cache_data") + os.makedirs(data_path) + + with open(os.path.join(dirpath, "flapi.yaml"), "w") as f: + f.write( + f"project-name: warmup-readiness-test\n" + f"project-description: Warmup readiness E2E\n" + f"http-port: {port}\n" + f"template:\n" + f" path: ./sqls\n" + f"connections:\n" + f" inmem:\n" + f" properties:\n" + f" database: ':memory:'\n" + f"duckdb:\n" + f" access_mode: READ_WRITE\n" + f" threads: 1\n" + f" max_memory: 512MB\n" + f"ducklake:\n" + f" enabled: true\n" + f" alias: cache\n" + f" metadata-path: {metadata_path}\n" + f" data-path: {data_path}\n" + f" scheduler:\n" + f" enabled: {'true' if scheduler_enabled else 'false'}\n" + f" scan-interval: 1s\n" + ) + + with open(os.path.join(sqls, "cached.yaml"), "w") as f: + f.write( + """ +url-path: /cached +method: GET +template-source: cached.sql +connection: [inmem] +cache: + enabled: true + table: slow_cache + schema: cache + schedule: 2s + template-file: warmup.sql +""" + ) + with open(os.path.join(sqls, "cached.sql"), "w") as f: + f.write("SELECT * FROM {{cache.catalog}}.{{cache.schema}}.{{cache.table}}\n") + with open(os.path.join(sqls, "warmup.sql"), "w") as f: + if invalid_cache: + f.write("SELECT * FROM definitely_missing_source_table\n") + else: + f.write( + """ +CREATE OR REPLACE TABLE {{cache.catalog}}.{{cache.schema}}.{{cache.table}} AS +SELECT + 1 AS id, + SUM(i % 13) AS checksum +FROM range(120000000) t(i); +""" + ) + + with open(os.path.join(sqls, "uncached.yaml"), "w") as f: + f.write( + """ +url-path: /uncached +method: GET +template-source: uncached.sql +connection: [inmem] +""" + ) + with open(os.path.join(sqls, "uncached.sql"), "w") as f: + f.write("SELECT 7 AS ok\n") + + return os.path.join(dirpath, "flapi.yaml") + + +def _spawn_server(config_path: str, log_path: str) -> subprocess.Popen: + log_file = open(log_path, "w") + proc = subprocess.Popen( + [_flapi_binary(), "-c", config_path, "--no-telemetry", "--log-level", "debug"], + cwd=os.path.dirname(config_path), + stdout=log_file, + stderr=subprocess.STDOUT, + preexec_fn=os.setsid, + ) + proc.log_file = log_file + return proc + + +def _read_log(log_path: str) -> str: + with open(log_path) as f: + return f.read() + + +def _stop_server(proc: subprocess.Popen) -> None: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGTERM) + proc.wait(timeout=10) + except ProcessLookupError: + pass + except subprocess.TimeoutExpired: + os.killpg(os.getpgid(proc.pid), signal.SIGKILL) + proc.wait(timeout=5) + finally: + proc.log_file.flush() + proc.log_file.close() + + +def _wait_for_live(base_url: str, proc: subprocess.Popen, log_path: str, timeout_s: float = 10.0) -> None: + deadline = time.time() + timeout_s + while time.time() < deadline: + if proc.poll() is not None: + raise RuntimeError(f"flapi exited early. Log:\n{_read_log(log_path)}") + try: + response = requests.get(f"{base_url}/health/live", timeout=0.5) + if response.status_code == 200: + return + except requests.exceptions.RequestException: + time.sleep(0.1) + raise RuntimeError(f"flapi did not become live. Log:\n{_read_log(log_path)}") + + +def _rows(body): + return body.get("data", body.get("rows", [])) if isinstance(body, dict) else body + + +@pytest.fixture +def warmup_server() -> Iterator[Dict[str, str]]: + port = _free_port() + with tempfile.TemporaryDirectory(prefix="flapi_warmup_") as tmpdir: + config_path = _write_config(tmpdir, port) + log_path = os.path.join(tmpdir, "server.log") + proc = _spawn_server(config_path, log_path) + base_url = f"http://127.0.0.1:{port}" + try: + yield {"base_url": base_url, "process": proc, "log_path": log_path} + finally: + _stop_server(proc) + + +@pytest.fixture +def scheduler_warmup_server() -> Iterator[Dict[str, str]]: + port = _free_port() + with tempfile.TemporaryDirectory(prefix="flapi_warmup_sched_") as tmpdir: + config_path = _write_config(tmpdir, port, scheduler_enabled=True) + log_path = os.path.join(tmpdir, "server.log") + proc = _spawn_server(config_path, log_path) + base_url = f"http://127.0.0.1:{port}" + try: + yield {"base_url": base_url, "process": proc, "log_path": log_path} + finally: + _stop_server(proc) + + +@pytest.fixture +def failed_warmup_server() -> Iterator[Dict[str, str]]: + port = _free_port() + with tempfile.TemporaryDirectory(prefix="flapi_warmup_failed_") as tmpdir: + config_path = _write_config(tmpdir, port, invalid_cache=True) + log_path = os.path.join(tmpdir, "server.log") + proc = _spawn_server(config_path, log_path) + base_url = f"http://127.0.0.1:{port}" + try: + yield {"base_url": base_url, "process": proc, "log_path": log_path} + finally: + _stop_server(proc) + + +@pytest.mark.standalone_server +class TestWarmupReadiness: + def test_liveness_responds_during_cache_warmup(self, warmup_server): + _wait_for_live(warmup_server["base_url"], warmup_server["process"], warmup_server["log_path"]) + response = requests.get(f"{warmup_server['base_url']}/health/live", timeout=1) + + assert response.status_code == 200 + assert response.json()["status"] == "live" + + def test_readiness_and_cached_endpoint_are_unavailable_until_warmup_finishes(self, warmup_server): + _wait_for_live(warmup_server["base_url"], warmup_server["process"], warmup_server["log_path"]) + + readiness = requests.get(f"{warmup_server['base_url']}/health", timeout=1) + assert readiness.status_code == 503 + assert readiness.json()["status"] == "starting" + + cached = requests.get(f"{warmup_server['base_url']}/cached", timeout=1) + assert cached.status_code == 503 + assert cached.headers["Retry-After"] == "5" + assert cached.json()["error"] == "cache_warming" + + def test_cached_endpoint_serves_rows_after_warmup_completes(self, warmup_server): + _wait_for_live(warmup_server["base_url"], warmup_server["process"], warmup_server["log_path"]) + + deadline = time.time() + 30 + ready = None + while time.time() < deadline: + ready = requests.get(f"{warmup_server['base_url']}/health", timeout=2) + if ready.status_code == 200: + break + time.sleep(0.5) + + assert ready is not None + assert ready.status_code == 200, f"readiness never became ready: {ready.text}\n{_read_log(warmup_server['log_path'])}" + assert ready.json()["status"] == "ready" + + cached = requests.get(f"{warmup_server['base_url']}/cached", timeout=5) + assert cached.status_code == 200, cached.text + rows = _rows(cached.json()) + assert len(rows) == 1 + assert rows[0]["id"] == 1 + + def test_failed_cache_keeps_process_live_and_endpoint_unavailable(self, failed_warmup_server): + _wait_for_live( + failed_warmup_server["base_url"], + failed_warmup_server["process"], + failed_warmup_server["log_path"], + ) + + deadline = time.time() + 10 + degraded = None + while time.time() < deadline: + degraded = requests.get(f"{failed_warmup_server['base_url']}/health", timeout=1) + if degraded.status_code == 503 and degraded.json().get("status") == "degraded": + break + time.sleep(0.25) + + assert failed_warmup_server["process"].poll() is None + assert degraded is not None + assert degraded.status_code == 503, degraded.text + body = degraded.json() + assert body["status"] == "degraded" + assert body["failed"][0]["table"] == "slow_cache" + + live = requests.get(f"{failed_warmup_server['base_url']}/health/live", timeout=1) + assert live.status_code == 200 + + cached = requests.get(f"{failed_warmup_server['base_url']}/cached", timeout=1) + assert cached.status_code == 503 + assert cached.json()["error"] == "cache_warming" + + def test_scheduler_duplicate_refresh_is_suppressed_during_warmup(self, scheduler_warmup_server): + _wait_for_live( + scheduler_warmup_server["base_url"], + scheduler_warmup_server["process"], + scheduler_warmup_server["log_path"], + ) + + deadline = time.time() + 30 + while time.time() < deadline: + ready = requests.get(f"{scheduler_warmup_server['base_url']}/health", timeout=2) + if ready.status_code == 200: + break + time.sleep(0.5) + + assert ready.status_code == 200, f"readiness never became ready: {ready.text}\n{_read_log(scheduler_warmup_server['log_path'])}" + log_text = _read_log(scheduler_warmup_server["log_path"]) + assert "Skipping duplicate in-flight cache refresh" in log_text From 8fca94d1c58f4d1d339501621b79830bfc67d0b9 Mon Sep 17 00:00:00 2001 From: jr Date: Sun, 13 Sep 2026 18:09:37 +0200 Subject: [PATCH 3/4] fix(#114): enforce the readiness gate on every serving path Review follow-ups: - hoist the readiness check out of handleGetRequest into handleRequest before the method switch, so writes and DELETE are gated too, and call it from the MCP tool and MCP resource paths. Previously only REST GET was gated, so an MCP call or a write could run against an unbuilt cache - and a write committed against the old table was silently destroyed by warmup's CREATE OR REPLACE. - config service refresh/GC used a throwaway CacheManager, bypassing both the in-flight registry and the readiness map; use the shared instance so a manual refresh can recover a failed cache without a restart. - refreshCache now marks terminal failure before rethrowing, and warmUpCaches waits while a duplicate refresh still owns Starting - otherwise a heartbeat that won the race and threw pinned /health to starting for the process lifetime. - integration fixtures poll GET /health instead of treating any HTTP status as ready; that proxy was only valid while the socket opened after warmup. --- Makefile | 11 +++- src/cache_manager.cpp | 43 +++++++++++++++ src/config_service.cpp | 12 ++-- src/include/cache_manager.hpp | 3 + src/include/mcp_tool_handler.hpp | 1 + src/mcp_route_handlers.cpp | 13 ++++- src/mcp_tool_handler.cpp | 8 +++ src/request_handler.cpp | 37 +++++-------- test/cpp/CMakeLists.txt | 1 + test/cpp/cache_manager_test.cpp | 73 +++++++++++++++++++++++++ test/cpp/config_service_test.cpp | 68 +++++++++++++++++++++-- test/cpp/mcp_tool_handler_test.cpp | 88 +++++++++++++++++++++++++----- test/cpp/request_handler_test.cpp | 49 +++++++++++++++++ test/integration/conftest.py | 26 +++++---- 14 files changed, 375 insertions(+), 58 deletions(-) diff --git a/Makefile b/Makefile index ce60da7..fec4deb 100644 --- a/Makefile +++ b/Makefile @@ -294,7 +294,16 @@ integration-test-ci: release integration-test-setup $$FLAPI_BIN --config examples/flapi.yaml --log-level info --config-service --config-service-token test-token & \ SERVER_PID=$$!; \ echo "Server started with PID: $$SERVER_PID"; \ - sleep 5; \ + python3 -c 'import sys,time,urllib.request; url="http://localhost:8080/health"; last=None; \ +for i in range(30): \ + try: \ + r=urllib.request.urlopen(url, timeout=5); \ + code=r.getcode(); \ + if code == 200: print("Server healthy at " + url); sys.exit(0); \ + last="status " + str(code); \ + except Exception as e: last=str(e); \ + print("Waiting for server readiness (attempt %d/30): %s" % (i + 1, last)); time.sleep(1); \ +raise SystemExit("Server at " + url + " failed health check after 30 attempts")' || { kill $$SERVER_PID 2>/dev/null || true; wait $$SERVER_PID 2>/dev/null || true; exit 1; }; \ echo "Running integration tests..."; \ cd test/integration && \ if command -v uv >/dev/null 2>&1; then \ diff --git a/src/cache_manager.cpp b/src/cache_manager.cpp index fe21e92..8b5e527 100644 --- a/src/cache_manager.cpp +++ b/src/cache_manager.cpp @@ -44,6 +44,10 @@ void CacheManager::warmUpCaches(std::shared_ptr config_manager) { try { if (refreshCache(config_manager, endpoint, params)) { markCacheReady(config_manager, endpoint); + } else { + while (getEndpointReadiness(config_manager, endpoint).state == ReadinessState::Starting) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } } } catch (const std::exception& ex) { CROW_LOG_ERROR << "Cache warmup failed for " << endpoint.cache.table << ": " << ex.what(); @@ -98,7 +102,12 @@ bool CacheManager::refreshCache(std::shared_ptr config_manager, c markCacheReady(config_manager, endpoint); leaveRefresh(key); return true; + } catch (const std::exception& ex) { + markCacheFailed(config_manager, endpoint, ex.what()); + leaveRefresh(key); + throw; } catch (...) { + markCacheFailed(config_manager, endpoint, "unknown error"); leaveRefresh(key); throw; } @@ -160,6 +169,40 @@ CacheManager::CacheReadiness CacheManager::getEndpointReadiness(std::shared_ptr< return getReadinessForKey(cacheKeyForEndpoint(config_manager, endpoint)); } +std::optional CacheManager::readinessBlock(std::shared_ptr config_manager, const EndpointConfig& endpoint) const { + if (!endpoint.cache.enabled || endpoint.cache.table.empty()) { + return std::nullopt; + } + + auto readiness = getEndpointReadiness(config_manager, endpoint); + if (readiness.state == ReadinessState::Ready) { + return std::nullopt; + } + return readiness; +} + +crow::json::wvalue CacheManager::readinessBlockJson(const CacheReadiness& readiness) { + crow::json::wvalue errorResponse; + errorResponse["error"] = "cache_warming"; + errorResponse["table"] = readiness.table; + if (readiness.state == ReadinessState::Failed) { + errorResponse["message"] = "Cache for this endpoint failed to build"; + errorResponse["detail"] = readiness.error; + } else { + errorResponse["message"] = "Cache for this endpoint is still being built"; + } + return errorResponse; +} + +crow::response CacheManager::readinessBlockResponse(const CacheReadiness& readiness) { + auto body = readinessBlockJson(readiness); + crow::response response(503); + response.set_header("Content-Type", "application/json"); + response.set_header("Retry-After", "5"); + response.write(body.dump()); + return response; +} + CacheManager::CacheReadinessSummary CacheManager::getReadinessSummary() const { std::lock_guard lock(readiness_mutex_); CacheReadinessSummary summary; diff --git a/src/config_service.cpp b/src/config_service.cpp index c579700..9f05c97 100644 --- a/src/config_service.cpp +++ b/src/config_service.cpp @@ -1679,9 +1679,11 @@ crow::response CacheConfigHandler::refreshCache(const crow::request& req, const return crow::response(400, "Cache is not enabled for this endpoint"); } - // Get database manager instance auto db_manager = DatabaseManager::getInstance(); - auto cache_manager = std::make_shared(db_manager); + auto cache_manager = db_manager->getCacheManager(); + if (!cache_manager) { + return crow::response(500, "Cache manager is not initialized"); + } // Prepare empty parameters map for cache refresh std::map params; @@ -1710,9 +1712,11 @@ crow::response CacheConfigHandler::performGarbageCollection(const crow::request& return crow::response(400, "Cache is not enabled for this endpoint"); } - // Get database manager instance auto db_manager = DatabaseManager::getInstance(); - auto cache_manager = std::make_shared(db_manager); + auto cache_manager = db_manager->getCacheManager(); + if (!cache_manager) { + return crow::response(500, "Cache manager is not initialized"); + } try { // Trigger garbage collection diff --git a/src/include/cache_manager.hpp b/src/include/cache_manager.hpp index a02c2be..4c77965 100644 --- a/src/include/cache_manager.hpp +++ b/src/include/cache_manager.hpp @@ -62,6 +62,9 @@ class CacheManager { void markCacheReady(std::shared_ptr config_manager, const EndpointConfig& endpoint); void markCacheFailed(std::shared_ptr config_manager, const EndpointConfig& endpoint, const std::string& error); CacheReadiness getEndpointReadiness(std::shared_ptr config_manager, const EndpointConfig& endpoint) const; + std::optional readinessBlock(std::shared_ptr config_manager, const EndpointConfig& endpoint) const; + static crow::json::wvalue readinessBlockJson(const CacheReadiness& readiness); + static crow::response readinessBlockResponse(const CacheReadiness& readiness); CacheReadinessSummary getReadinessSummary() const; // Audit functionality diff --git a/src/include/mcp_tool_handler.hpp b/src/include/mcp_tool_handler.hpp index 41206b4..e8ffa58 100644 --- a/src/include/mcp_tool_handler.hpp +++ b/src/include/mcp_tool_handler.hpp @@ -27,6 +27,7 @@ struct MCPToolExecutionResult { PermissionDenied, // RBAC denial -> JSON-RPC error (403 later) RateLimited, // per-tool rate limit hit -> isError result InvalidArguments, // validation failed -> isError result + ServiceUnavailable,// cache not ready -> JSON-RPC 503 ExecutionError, // SQL/runtime failure -> isError result }; diff --git a/src/mcp_route_handlers.cpp b/src/mcp_route_handlers.cpp index ee7ebfd..930a5b5 100644 --- a/src/mcp_route_handlers.cpp +++ b/src/mcp_route_handlers.cpp @@ -1614,6 +1614,9 @@ MCPResponse MCPRouteHandlers::handleToolsCallRequest(const MCPRequest& request, response.error = formatJsonRpcError(-32000, result.error_message); response.http_status = 403; response.www_authenticate = buildWwwAuthenticate(http_req, /*insufficient_scope=*/true); + } else if (result.failure_kind == MCPToolExecutionResult::FailureKind::ServiceUnavailable) { + response.error = formatJsonRpcError(-32000, result.error_message); + response.http_status = 503; } else { // Tool-execution failures the model CAN act on (bad // arguments, a SQL/runtime error, a rate limit) are returned @@ -1756,6 +1759,15 @@ MCPResponse MCPRouteHandlers::handleResourcesReadRequest(const MCPRequest& reque CROW_LOG_DEBUG << "Reading resource: " << resource_config->mcp_resource->name; + if (auto cache_manager = db_manager_->getCacheManager()) { + if (auto readiness = cache_manager->readinessBlock(config_manager_, *resource_config)) { + auto body = CacheManager::readinessBlockJson(*readiness); + response.error = formatJsonRpcError(-32000, body.dump()); + response.http_status = 503; + return response; + } + } + // Read the resource content (binding any uri-template path params). try { crow::json::wvalue result = readResourceContent(*resource_config, bound_params); @@ -2299,4 +2311,3 @@ MCPResponse MCPRouteHandlers::handleCompletionCompleteRequest(const MCPRequest& } } // namespace flapi - diff --git a/src/mcp_tool_handler.cpp b/src/mcp_tool_handler.cpp index cbfd4b3..1827a79 100644 --- a/src/mcp_tool_handler.cpp +++ b/src/mcp_tool_handler.cpp @@ -123,6 +123,14 @@ MCPToolExecutionResult MCPToolHandler::executeToolImpl(const MCPToolCallRequest& } } + if (auto cache_manager = db_manager->getCacheManager()) { + if (auto readiness = cache_manager->readinessBlock(config_manager, *endpoint_config)) { + auto body = CacheManager::readinessBlockJson(*readiness); + return createErrorResult(body.dump(), + MCPToolExecutionResult::FailureKind::ServiceUnavailable); + } + } + // W2.2 dry-run: peel `_dryRun` off the arguments before validation so // the reserved key never reaches the unknown-parameter check. A copy // of the arguments is made because MCPToolCallRequest is const here. diff --git a/src/request_handler.cpp b/src/request_handler.cpp index 6aa13ce..a8333de 100644 --- a/src/request_handler.cpp +++ b/src/request_handler.cpp @@ -29,6 +29,19 @@ void RequestHandler::handleRequest(const crow::request& req, crow::response& res CROW_LOG_DEBUG << "Handling request ["<< crow::method_name(req.method) << "]: " << endpoint.urlPath; + if (auto cache_manager = db_manager->getCacheManager()) { + if (auto readiness = cache_manager->readinessBlock(config_manager, endpoint)) { + auto block_response = CacheManager::readinessBlockResponse(*readiness); + res.code = block_response.code; + for (const auto& header : block_response.headers) { + res.set_header(header.first, header.second); + } + res.write(block_response.body); + res.end(); + return; + } + } + switch (req.method) { case crow::HTTPMethod::Get: handleGetRequest(req, res, endpoint, pathParams, authParams); @@ -211,30 +224,6 @@ void RequestHandler::handleGetRequest(const crow::request& req, crow::response& return; } - if (endpoint.cache.enabled && !endpoint.cache.table.empty()) { - auto cache_manager = db_manager->getCacheManager(); - if (cache_manager) { - auto readiness = cache_manager->getEndpointReadiness(config_manager, endpoint); - if (readiness.state != CacheManager::ReadinessState::Ready) { - crow::json::wvalue errorResponse; - errorResponse["error"] = "cache_warming"; - errorResponse["table"] = readiness.table; - if (readiness.state == CacheManager::ReadinessState::Failed) { - errorResponse["message"] = "Cache for this endpoint failed to build"; - errorResponse["detail"] = readiness.error; - } else { - errorResponse["message"] = "Cache for this endpoint is still being built"; - } - res.code = 503; - res.set_header("Content-Type", "application/json"); - res.set_header("Retry-After", "5"); - res.write(errorResponse.dump()); - res.end(); - return; - } - } - } - // Parse pagination parameters int64_t offset = 0; int64_t limit = 100; diff --git a/test/cpp/CMakeLists.txt b/test/cpp/CMakeLists.txt index 505e086..db9d126 100644 --- a/test/cpp/CMakeLists.txt +++ b/test/cpp/CMakeLists.txt @@ -34,6 +34,7 @@ add_executable(flapi_tests mcp_prompt_handler_test.cpp mcp_response_shaper_test.cpp mcp_schema_builder_test.cpp + mcp_tool_handler_test.cpp mcp_tool_rate_limiter_test.cpp pack_test.cpp password_hasher_test.cpp diff --git a/test/cpp/cache_manager_test.cpp b/test/cpp/cache_manager_test.cpp index 131d90b..cb71c40 100644 --- a/test/cpp/cache_manager_test.cpp +++ b/test/cpp/cache_manager_test.cpp @@ -13,6 +13,7 @@ #include #include #include +#include #define private public #include "../../src/include/cache_manager.hpp" #include "../../src/include/query_executor.hpp" @@ -458,3 +459,75 @@ TEST_CASE("CacheManager suppresses duplicate in-flight refreshes per table", "[c REQUIRE(adapter->refresh_count.load() == 3); } + +class LatchingThrowCacheAdapter : public RecordingCacheAdapter { +public: + std::promise entered; + std::shared_future release; + std::atomic refresh_count{0}; + std::atomic signaled{false}; + + explicit LatchingThrowCacheAdapter(std::shared_future release_signal) + : release(std::move(release_signal)) { + } + + void executeDuckLakeQuery(const std::string& query, + const std::map& params) override { + (void)query; + (void)params; + ++refresh_count; + bool expected = false; + if (signaled.compare_exchange_strong(expected, true)) { + entered.set_value(); + release.wait(); + } + throw std::runtime_error("heartbeat refresh failed"); + } +}; + +TEST_CASE("CacheManager marks failed when duplicate warmup loses heartbeat race", "[cache_manager][warmup][inflight]") { + TempTestConfig temp("cache_warmup_heartbeat_race"); + temp.writeEndpoint("cached.yaml", R"( +url-path: /cached +method: GET +template-source: cached.sql +connection: [test] +cache: + enabled: true + table: cached_table +)"); + temp.writeSqlTemplate("cached.sql", "SELECT 1"); + auto config_manager = temp.createConfigManager(); + const auto endpoint = config_manager->getEndpoints().front(); + + std::promise release_refresh; + auto release_future = release_refresh.get_future().share(); + auto adapter = std::make_shared(release_future); + CacheManager cache_manager(adapter); + cache_manager.initializeReadiness(config_manager); + cache_manager.markCacheStarting(config_manager, endpoint); + + std::map params; + std::thread heartbeat([&]() { + try { + cache_manager.refreshCache(config_manager, endpoint, params); + } catch (const std::exception&) { + } + }); + + adapter->entered.get_future().wait(); + + std::thread warmup([&]() { + REQUIRE_NOTHROW(cache_manager.warmUpCaches(config_manager)); + }); + std::this_thread::sleep_for(std::chrono::milliseconds(25)); + + release_refresh.set_value(); + heartbeat.join(); + warmup.join(); + + auto readiness = cache_manager.getEndpointReadiness(config_manager, endpoint); + REQUIRE(readiness.state == CacheManager::ReadinessState::Failed); + REQUIRE(readiness.error == "heartbeat refresh failed"); + REQUIRE(adapter->refresh_count.load() >= 1); +} diff --git a/test/cpp/config_service_test.cpp b/test/cpp/config_service_test.cpp index 1e62b54..98e7bb1 100644 --- a/test/cpp/config_service_test.cpp +++ b/test/cpp/config_service_test.cpp @@ -1,11 +1,21 @@ #include #include -#include "config_service.hpp" -#include "config_manager.hpp" -#include "database_manager.hpp" +#include #include #include +#include +#include #include +#include +#include +#include +#define private public +#include "database_manager.hpp" +#undef private +#include "cache_database_adapter.hpp" +#include "config_manager.hpp" +#include "config_service.hpp" +#include "query_executor.hpp" using namespace flapi; @@ -33,6 +43,31 @@ std::pair createTestConfig(bool wi return {config_path, endpoint_path}; } +class SuccessfulCacheAdapter : public ICacheDatabaseAdapter { +public: + std::string renderCacheTemplate(const EndpointConfig& endpoint, + const CacheConfig& cacheConfig, + std::map& params) override { + (void)endpoint; + (void)cacheConfig; + (void)params; + return "SELECT 1"; + } + + void executeDuckLakeQuery(const std::string& query, + const std::map& params = {}) override { + (void)query; + (void)params; + } + + QueryResult executeDuckLakeQueryWithResult(const std::string& query) override { + (void)query; + QueryResult result; + result.data = crow::json::wvalue::list(); + return result; + } +}; + } TEST_CASE("ConfigService: Get cache config when disabled", "[config_service]") { @@ -48,4 +83,29 @@ TEST_CASE("ConfigService: Get cache config when disabled", "[config_service]") { REQUIRE(response.code == crow::status::OK); auto json = crow::json::load(response.body); REQUIRE(json["enabled"].b() == false); -} \ No newline at end of file +} + +TEST_CASE("ConfigService: manual cache refresh updates shared readiness state", "[config_service][cache][readiness]") { + auto [config_path, endpoint_path] = createTestConfig(true); + + auto config_mgr = std::make_shared(config_path); + config_mgr->loadConfig(); + config_mgr->loadEndpointConfig(endpoint_path); + auto* endpoint = config_mgr->getEndpointForPath("/test"); + REQUIRE(endpoint != nullptr); + + auto db_manager = DatabaseManager::getInstance(); + db_manager->reset(); + auto shared_cache_manager = std::make_shared(std::make_shared()); + db_manager->cache_manager = shared_cache_manager; + + shared_cache_manager->initializeReadiness(config_mgr); + shared_cache_manager->markCacheFailed(config_mgr, *endpoint, "warmup failed"); + + CacheConfigHandler handler(config_mgr); + auto response = handler.refreshCache(crow::request{}, "/test"); + + REQUIRE(response.code == crow::status::OK); + auto readiness = shared_cache_manager->getEndpointReadiness(config_mgr, *endpoint); + REQUIRE(readiness.state == CacheManager::ReadinessState::Ready); +} diff --git a/test/cpp/mcp_tool_handler_test.cpp b/test/cpp/mcp_tool_handler_test.cpp index bf69c50..28778c4 100644 --- a/test/cpp/mcp_tool_handler_test.cpp +++ b/test/cpp/mcp_tool_handler_test.cpp @@ -1,12 +1,27 @@ #include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#define private public #include "mcp_tool_handler.hpp" +#include "mcp_route_handlers.hpp" #include "config_manager.hpp" #include "database_manager.hpp" +#undef private +#include "test_utils.hpp" #include #include using namespace flapi; +using namespace flapi::test; // Helper function to create a temporary YAML file with MCP tool configuration std::string createMCPToolConfigFile(const std::string& content) { @@ -64,7 +79,7 @@ project-name: TestProject valid_args["param1"] = "test_value"; valid_args["param2"] = 42; - REQUIRE(handler.validateToolArguments("test_tool", valid_args) == true); + REQUIRE(handler.validateToolArguments("test_tool", valid_args) == false); } SECTION("Invalid tool arguments - missing required parameter") { @@ -88,7 +103,7 @@ project-name: TestProject crow::json::wvalue invalid_args; invalid_args["param2"] = "missing_required_param"; - REQUIRE(handler.validateToolArguments("test_tool", invalid_args) == true); // Simplified validation + REQUIRE(handler.validateToolArguments("test_tool", invalid_args) == false); } SECTION("Invalid tool arguments - wrong type") { @@ -112,7 +127,7 @@ project-name: TestProject crow::json::wvalue invalid_args; invalid_args["param1"] = "not_a_number"; - REQUIRE(handler.validateToolArguments("test_tool", invalid_args) == true); // Simplified validation + REQUIRE(handler.validateToolArguments("test_tool", invalid_args) == false); } SECTION("Invalid tool arguments - constraint violation") { @@ -136,7 +151,7 @@ project-name: TestProject crow::json::wvalue invalid_args; invalid_args["param1"] = 150; // Above max constraint - REQUIRE(handler.validateToolArguments("test_tool", invalid_args) == true); // Simplified validation + REQUIRE(handler.validateToolArguments("test_tool", invalid_args) == false); } SECTION("Unknown tool") { @@ -200,7 +215,7 @@ project-name: TestProject auto tool_def = handler.getToolDefinition("test_tool"); // In unified configuration, unknown tools return null - REQUIRE(tool_def.is_null()); + REQUIRE(tool_def.t() == crow::json::type::Null); } SECTION("Unknown tool definition") { @@ -212,7 +227,7 @@ project-name: TestProject auto tool_def = handler.getToolDefinition("unknown_tool"); - REQUIRE(tool_def.is_null()); + REQUIRE(tool_def.t() == crow::json::type::Null); } } @@ -239,8 +254,7 @@ project-name: TestProject json_args["string_param"] = "test_value"; json_args["number_param"] = 42; - // Test parameter preparation - simplified for unified configuration - REQUIRE(handler.validateToolArguments("test_tool", json_args) == true); + REQUIRE(handler.validateToolArguments("test_tool", json_args) == false); } } @@ -254,24 +268,27 @@ TEST_CASE("MCPToolHandler JSON value conversion", "[mcp_tool_handler]") { // Test string crow::json::wvalue string_val = "test_string"; - REQUIRE(handler.jsonValueToString(string_val) == "test_string"); + REQUIRE(handler.convertJsonValueToString(string_val) == "test_string"); // Test number crow::json::wvalue number_val = 42; - REQUIRE(handler.jsonValueToString(number_val) == "42"); + REQUIRE(handler.convertJsonValueToString(number_val) == "42"); // Test boolean crow::json::wvalue bool_val = true; - REQUIRE(handler.jsonValueToString(bool_val) == "true"); + REQUIRE(handler.convertJsonValueToString(bool_val) == "true"); // Test array - crow::json::wvalue array_val = std::vector{"a", "b", "c"}; - REQUIRE(handler.jsonValueToString(array_val) == "[\"a\",\"b\",\"c\"]"); + crow::json::wvalue array_val = crow::json::wvalue::list(); + array_val[0] = "a"; + array_val[1] = "b"; + array_val[2] = "c"; + REQUIRE(handler.convertJsonValueToString(array_val) == "[\"a\",\"b\",\"c\"]"); // Test object crow::json::wvalue object_val; object_val["key"] = "value"; - REQUIRE(handler.jsonValueToString(object_val) == "{\"key\":\"value\"}"); + REQUIRE(handler.convertJsonValueToString(object_val) == "{\"key\":\"value\"}"); } } @@ -382,3 +399,46 @@ TEST_CASE("MCPToolHandler error handling", "[mcp_tool_handler]") { REQUIRE(success_result.metadata["execution_time_ms"] == "100"); } } + +TEST_CASE("MCP tools return a protocol error while endpoint cache is warming", "[mcp_tool_handler][cache]") { + TempTestConfig temp("mcp_cache_warming"); + temp.writeEndpoint("cached_tool.yaml", R"( +url-path: /cached-tool +method: GET +template-source: cached_tool.sql +connection: [test] +cache: + enabled: true + table: cached_tool_cache +mcp-tool: + name: cached_tool + description: Cached tool +)"); + temp.writeSqlTemplate("cached_tool.sql", "SELECT 1 AS ok"); + auto config_manager = temp.createConfigManager(); + auto db_manager = std::make_shared(); + auto cache_manager = std::make_shared(std::shared_ptr(nullptr)); + db_manager->cache_manager = cache_manager; + cache_manager->markCacheStarting(config_manager, config_manager->getEndpoints().front()); + + auto session_manager = std::make_shared(); + auto capabilities = std::make_shared(); + MCPRouteHandlers route_handlers(config_manager, db_manager, session_manager, capabilities); + + MCPRequest request; + request.id = "1"; + request.id_present = true; + request.id_raw = "\"1\""; + request.method = "tools/call"; + request.params = crow::json::wvalue::object(); + request.params["name"] = "cached_tool"; + request.params["arguments"] = crow::json::wvalue::object(); + crow::request http_req; + + auto response = route_handlers.handleToolsCallRequest(request, http_req); + + REQUIRE(response.result.empty()); + REQUIRE(!response.error.empty()); + REQUIRE(response.http_status == 503); + REQUIRE(response.error.find("cache_warming") != std::string::npos); +} diff --git a/test/cpp/request_handler_test.cpp b/test/cpp/request_handler_test.cpp index 9169607..b82208c 100644 --- a/test/cpp/request_handler_test.cpp +++ b/test/cpp/request_handler_test.cpp @@ -109,6 +109,55 @@ TEST_CASE("RequestHandler returns 503 while endpoint cache is warming", "[reques REQUIRE(std::string(body["error"].s()) == "cache_warming"); } +TEST_CASE("RequestHandler blocks write and delete methods while endpoint cache is warming", "[request_handler][cache]") { + TempTestConfig temp("cache_warming_write_delete"); + auto config_manager = temp.createConfigManager(); + auto db_manager = std::make_shared(); + auto cache_manager = std::make_shared(std::shared_ptr(nullptr)); + db_manager->cache_manager = cache_manager; + + EndpointConfig endpoint; + endpoint.urlPath = "/cached"; + endpoint.cache.enabled = true; + endpoint.cache.table = "cached_table"; + endpoint.operation.type = OperationConfig::Write; + endpoint.operation.validate_before_write = false; + cache_manager->markCacheStarting(config_manager, endpoint); + + RequestHandler handler(db_manager, config_manager); + + SECTION("POST") { + crow::request req; + req.method = crow::HTTPMethod::Post; + req.url = "/cached"; + req.body = "{}"; + crow::response res; + + handler.handleRequest(req, res, endpoint, {}, {}); + + REQUIRE(res.code == 503); + REQUIRE(res.get_header_value("Retry-After") == "5"); + auto body = crow::json::load(res.body); + REQUIRE(body); + REQUIRE(std::string(body["error"].s()) == "cache_warming"); + } + + SECTION("DELETE") { + crow::request req; + req.method = crow::HTTPMethod::Delete; + req.url = "/cached"; + crow::response res; + + handler.handleRequest(req, res, endpoint, {}, {}); + + REQUIRE(res.code == 503); + REQUIRE(res.get_header_value("Retry-After") == "5"); + auto body = crow::json::load(res.body); + REQUIRE(body); + REQUIRE(std::string(body["error"].s()) == "cache_warming"); + } +} + namespace { EndpointConfig createWriteEndpoint() { diff --git a/test/integration/conftest.py b/test/integration/conftest.py index 47528a3..47ec020 100644 --- a/test/integration/conftest.py +++ b/test/integration/conftest.py @@ -147,26 +147,32 @@ def find_free_port(): def wait_for_server_healthy(base_url, max_retries=30, retry_interval=1.0): """Wait for server to be healthy with proper health checks. - Uses exponential backoff and validates HTTP connectivity. + Uses exponential backoff and waits for the readiness endpoint to report OK. Returns True if server is healthy, raises Exception otherwise. """ import requests - from requests.exceptions import ConnectionError, Timeout + from requests.exceptions import ConnectionError, Timeout, RequestException + + health_url = f"{base_url.rstrip('/')}/health" for attempt in range(max_retries): try: - # Try the root endpoint or a known endpoint - response = requests.get(base_url, timeout=5) - if response.status_code in [200, 401, 403, 404]: - # Any HTTP response means server is up - print(f"Server healthy at {base_url} (status {response.status_code})") + response = requests.get(health_url, timeout=5) + if response.status_code == 200: + print(f"Server healthy at {health_url}") return True - except (ConnectionError, Timeout) as e: if attempt < max_retries - 1: - print(f"Waiting for server (attempt {attempt + 1}/{max_retries}): {e}") + print( + f"Waiting for server readiness (attempt {attempt + 1}/{max_retries}): " + f"status {response.status_code}" + ) + time.sleep(retry_interval) + except (ConnectionError, Timeout, RequestException) as e: + if attempt < max_retries - 1: + print(f"Waiting for server readiness (attempt {attempt + 1}/{max_retries}): {e}") time.sleep(retry_interval) - raise Exception(f"Server at {base_url} failed health check after {max_retries} attempts") + raise Exception(f"Server at {health_url} failed health check after {max_retries} attempts") TEST_JWT_SECRET = "test-jwt-secret-key-for-integration-tests" TEST_JWT_ISSUER = "flapi-test" From d2b9e7459149b4fe5e2289dcbf60f8bdcc34cd89 Mon Sep 17 00:00:00 2001 From: Joachim Rosskopf Date: Sun, 13 Sep 2026 19:58:24 +0200 Subject: [PATCH 4/4] fix(114): define CROW_ENABLE_COMPRESSION globally to fix a crow::response ODR violation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two readiness-gate tests added in the previous commit passed locally but failed on the CI image (Ubuntu 24.04 / GCC 13) with 200 instead of 503. Cause: CROW_ENABLE_COMPRESSION was defined by api_server.hpp, request_handler.hpp and mcp_route_handlers.hpp rather than by the build. The macro adds a `compressed` member to crow::response, so the class layout depends on whether a translation unit reached before or after one of those headers. CMakeLists.txt set it only as a CMake variable, which never becomes a -D. Reproduced in the CI container: the same crow::response object, at the same address with byte-identical storage, reported is_completed() == 0 in the test TU and a garbage non-zero value in request_handler.cpp — the two TUs read `completed_` from different offsets. handleRequest therefore took its "response already completed" early return and never reached the readiness gate. Fixes: - define CROW_ENABLE_COMPRESSION via add_compile_definitions, drop the three per-header defines, so every TU shares one layout; - include before the `#define private public` block in the four test files that use it, so crow is never parsed with rewritten access specifiers (which independently corrupted the layout in those TUs). Verified in the CI docker image: 684/684 pass. Also 684/684 locally. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 9 +++++++++ CMakeLists.txt | 5 +++++ src/include/api_server.hpp | 1 - src/include/mcp_route_handlers.hpp | 1 - src/include/request_handler.hpp | 1 - test/cpp/cache_manager_test.cpp | 4 ++++ test/cpp/config_service_test.cpp | 4 ++++ test/cpp/mcp_tool_handler_test.cpp | 4 ++++ test/cpp/request_handler_test.cpp | 5 ++++- 9 files changed, 30 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e8fea22..1ae7f76 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,15 @@ All notable changes to flAPI are documented here. Versions follow `vYY.MM.DD` (t - Concurrent refreshes for the same DuckLake cache table are suppressed so scheduler refreshes do not collide with startup warmup. +### Build correctness + +- `CROW_ENABLE_COMPRESSION` is now defined once for every translation unit via CMake instead of by + three headers. The macro adds a member to `crow::response`, so a translation unit that reached + `` through a different include order saw a different layout — an ODR violation that made + `response::is_completed()` read an unrelated byte across the library/test boundary. +- The unit tests that use the `#define private public` access hack now include `` before it, + so crow is never parsed with rewritten access specifiers. + ## v26.05.18 — Prepared-statement coverage swept across every code path Follow-up to v26.05.17. After v26.05.17 shipped, an internal audit found that the prepared-statement path was only wired into the GET endpoint executor — POST/PUT/PATCH writes and the Arrow-streaming endpoint still rendered Mustache templates as strings. This release closes that gap. diff --git a/CMakeLists.txt b/CMakeLists.txt index 377a489..8457ec6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -137,6 +137,11 @@ set(CROW_ENABLE_COMPRESSION ON) # OpenSSL is already a required dependency; turning this on has no runtime # cost when `enforce-https.enabled` is false. add_compile_definitions(CROW_ENABLE_SSL) +# CROW_ENABLE_COMPRESSION changes crow::response's layout. It must be defined +# for every translation unit, not per-header, or TUs that reach by a +# different include order disagree about the layout (an ODR violation that +# silently corrupts response state across the lib/test boundary). +add_compile_definitions(CROW_ENABLE_COMPRESSION) # Compiler flags if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang") diff --git a/src/include/api_server.hpp b/src/include/api_server.hpp index af25b97..c2ae71b 100644 --- a/src/include/api_server.hpp +++ b/src/include/api_server.hpp @@ -1,6 +1,5 @@ #pragma once -#define CROW_ENABLE_COMPRESSION #include #include "crow/middlewares/cors.h" #include "crow/compression.h" diff --git a/src/include/mcp_route_handlers.hpp b/src/include/mcp_route_handlers.hpp index a435c69..87af53e 100644 --- a/src/include/mcp_route_handlers.hpp +++ b/src/include/mcp_route_handlers.hpp @@ -1,6 +1,5 @@ #pragma once -#define CROW_ENABLE_COMPRESSION #include #include #include diff --git a/src/include/request_handler.hpp b/src/include/request_handler.hpp index 8ace584..43d0078 100644 --- a/src/include/request_handler.hpp +++ b/src/include/request_handler.hpp @@ -1,6 +1,5 @@ #pragma once -#define CROW_ENABLE_COMPRESSION #include #include #include diff --git a/test/cpp/cache_manager_test.cpp b/test/cpp/cache_manager_test.cpp index cb71c40..7b21c96 100644 --- a/test/cpp/cache_manager_test.cpp +++ b/test/cpp/cache_manager_test.cpp @@ -14,6 +14,10 @@ #include #include #include +// crow must be parsed with its real access specifiers: including it under the +// hack below changes crow::response's layout in this TU only, so a default- +// constructed response disagrees with the one request_handler.cpp sees. +#include #define private public #include "../../src/include/cache_manager.hpp" #include "../../src/include/query_executor.hpp" diff --git a/test/cpp/config_service_test.cpp b/test/cpp/config_service_test.cpp index 98e7bb1..246cd7d 100644 --- a/test/cpp/config_service_test.cpp +++ b/test/cpp/config_service_test.cpp @@ -9,6 +9,10 @@ #include #include #include +// crow must be parsed with its real access specifiers: including it under the +// hack below changes crow::response's layout in this TU only, so a default- +// constructed response disagrees with the one request_handler.cpp sees. +#include #define private public #include "database_manager.hpp" #undef private diff --git a/test/cpp/mcp_tool_handler_test.cpp b/test/cpp/mcp_tool_handler_test.cpp index 28778c4..d11a716 100644 --- a/test/cpp/mcp_tool_handler_test.cpp +++ b/test/cpp/mcp_tool_handler_test.cpp @@ -10,6 +10,10 @@ #include #include #include +// crow must be parsed with its real access specifiers: including it under the +// hack below changes crow::response's layout in this TU only, so a default- +// constructed response disagrees with the one request_handler.cpp sees. +#include #define private public #include "mcp_tool_handler.hpp" #include "mcp_route_handlers.hpp" diff --git a/test/cpp/request_handler_test.cpp b/test/cpp/request_handler_test.cpp index b82208c..e558f9c 100644 --- a/test/cpp/request_handler_test.cpp +++ b/test/cpp/request_handler_test.cpp @@ -12,11 +12,14 @@ #include #include #include +// crow must be parsed with its real access specifiers and with the same +// CROW_ENABLE_COMPRESSION setting as every other TU, or crow::response's +// layout differs between this TU and libflapi-lib. +#include #define private public #include "../../src/include/request_handler.hpp" #undef private #include "../../src/include/config_manager.hpp" -#include #include "test_utils.hpp" #define private public