diff --git a/CHANGELOG.md b/CHANGELOG.md index 526be8f..1ae7f76 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,30 @@ 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. + +### 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/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/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/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`). 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..8b5e527 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,40 @@ 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); + } 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(); + 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 +90,145 @@ 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 (const std::exception& ex) { + markCacheFailed(config_manager, endpoint, ex.what()); + leaveRefresh(key); + throw; + } catch (...) { + markCacheFailed(config_manager, endpoint, "unknown error"); + 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)); +} + +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; + 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/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/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..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" @@ -41,6 +40,8 @@ class APIServer crow::response getConfig(); crow::response refreshConfig(); + crow::response getLiveHealth(); + crow::response getHealth(); void run(int port = 8080); void stop(); @@ -69,6 +70,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..4c77965 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,48 @@ 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; + 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 void initializeAuditTables(std::shared_ptr config_manager); void ensureCacheSchemaExists(const std::string& catalog, const std::string& schema); @@ -45,11 +83,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/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/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/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/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/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 3458659..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); 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 ec015ef..7b21c96 100644 --- a/test/cpp/cache_manager_test.cpp +++ b/test/cpp/cache_manager_test.cpp @@ -12,6 +12,12 @@ #include #include #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" @@ -308,3 +314,224 @@ 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); +} + +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..246cd7d 100644 --- a/test/cpp/config_service_test.cpp +++ b/test/cpp/config_service_test.cpp @@ -1,11 +1,25 @@ #include #include -#include "config_service.hpp" -#include "config_manager.hpp" -#include "database_manager.hpp" +#include #include #include +#include +#include #include +#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 +#include "cache_database_adapter.hpp" +#include "config_manager.hpp" +#include "config_service.hpp" +#include "query_executor.hpp" using namespace flapi; @@ -33,6 +47,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 +87,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..d11a716 100644 --- a/test/cpp/mcp_tool_handler_test.cpp +++ b/test/cpp/mcp_tool_handler_test.cpp @@ -1,12 +1,31 @@ #include #include +#include +#include +#include +#include +#include +#include +#include +#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" #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 +83,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 +107,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 +131,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 +155,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 +219,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 +231,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 +258,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 +272,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 +403,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 1e2b437..e558f9c 100644 --- a/test/cpp/request_handler_test.cpp +++ b/test/cpp/request_handler_test.cpp @@ -12,17 +12,155 @@ #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 +#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"); +} + +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" 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