emme coverage

Directory: src/
File: src/backend_pool.c
Date: 2026-05-14 14:35:13
Exec Total Coverage
Lines: 211 393 53.7%
Functions: 19 34 55.9%
Branches: 78 200 39.0%
Line Branch Exec Source
1 /* backend_pool.c - Backend connection pooling for HTTP/2 reverse proxy
2 *
3 * This module manages a pool of reusable HTTP/2 connections to backend servers.
4 * Features:
5 * - Fixed-size connection pool with mutex protection
6 * - Connection reuse across multiple requests
7 * - Health tracking per connection
8 * - Idle timeout for connection cleanup
9 * - Thread-safe acquisition/release
10 */
11
12 #include <stdio.h>
13 #include <stdlib.h>
14 #include <string.h>
15 #include <time.h>
16 #include <unistd.h>
17 #include "backend_pool.h"
18 #include "log.h"
19 #include "http2_client.h"
20 #include "metrics.h"
21 #include "http_status.h"
22
23 #ifndef DEBUG_H2C
24 #define DEBUG_H2C 0
25 #endif
26
27 #define H2C_LOG(...) \
28 do { \
29 if (DEBUG_H2C) \
30 log_message(LOG_LEVEL_DEBUG, __VA_ARGS__); \
31 } while (0)
32
33 // Create a single backend connection
34 29 static backend_conn_t* create_backend_connection(backend_pool_t *pool, const backend_config_t *config)
35 {
36 29 backend_conn_t *conn = calloc(1, sizeof(backend_conn_t));
37
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 29 times.
 29 if (!conn) {
38 ✗ log_message(LOG_LEVEL_ERROR, "Failed to allocate backend connection");
39 ✗ return NULL;
40 }
41
42
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 29 times.
 29 if (pthread_mutex_init(&conn->lock, NULL) != 0) {
43 ✗ log_message(LOG_LEVEL_ERROR, "Failed to initialize connection mutex");
44 ✗ free(conn);
45 ✗ return NULL;
46 }
47
48 29 conn->pool = pool;
49 29 atomic_store(&conn->in_use, false);
50 29 atomic_store(&conn->last_used, time(NULL));
51 29 conn->health = BACKEND_HEALTH_UNKNOWN;
52 29 conn->consecutive_failures = 0;
53 29 conn->consecutive_successes = 0;
54
55
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 29 times.
 29 if (http2_client_init(&conn->client, config) != 0) {
56 ✗ log_message(LOG_LEVEL_ERROR, "Failed to initialize HTTP/2 client");
57 ✗ pthread_mutex_destroy(&conn->lock);
58 ✗ free(conn);
59 ✗ return NULL;
60 }
61
62 29 return conn;
63 }
64
65 // Destroy a single backend connection
66 29 static void destroy_backend_connection(backend_conn_t *conn)
67 {
68
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 29 times.
 29 if (!conn) return;
69
70 29 http2_client_cleanup(&conn->client);
71 29 pthread_mutex_destroy(&conn->lock);
72 29 free(conn);
73 }
74
75 // Connect a backend connection to the server
76 ✗ static int connect_backend_connection(backend_conn_t *conn, const backend_config_t *config)
77 {
78 ✗ if (!conn) return -1;
79 ✗ return http2_client_connect(&conn->client, config);
80 }
81
82 12 backend_pool_t* backend_pool_create(const char *host, int port,
83 bool tls_enabled, bool tls_verify,
84 int pool_size)
85 {
86
3/6
✓ Branch 0 taken 12 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 12 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 12 times.
 12 if (!host || pool_size <= 0 || pool_size > BACKEND_POOL_MAX_SIZE) {
87 ✗ log_message(LOG_LEVEL_ERROR, "Invalid backend pool parameters");
88 ✗ return NULL;
89 }
90
91 12 backend_pool_t *pool = calloc(1, sizeof(backend_pool_t));
92
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 12 times.
 12 if (!pool) {
93 ✗ log_message(LOG_LEVEL_ERROR, "Failed to allocate backend pool");
94 ✗ return NULL;
95 }
96
97
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 12 times.
 12 if (pthread_mutex_init(&pool->pool_lock, NULL) != 0) {
98 ✗ log_message(LOG_LEVEL_ERROR, "Failed to initialize pool mutex");
99 ✗ free(pool);
100 ✗ return NULL;
101 }
102
103 12 strncpy(pool->backend_host, host, sizeof(pool->backend_host) - 1);
104 12 pool->backend_port = port;
105 12 pool->tls_enabled = tls_enabled;
106 12 pool->tls_verify = tls_verify;
107 12 pool->idle_timeout_sec = BACKEND_POOL_IDLE_TIMEOUT_SEC;
108 12 atomic_store(&pool->size, 0);
109 12 atomic_store(&pool->active_count, 0);
110 12 atomic_store(&pool->idle_count, 0);
111 12 atomic_store(&pool->healthy_count, 0);
112
113 12 pool->config.host[sizeof(pool->config.host) - 1] = '\0';
114 12 strncpy(pool->config.host, host, sizeof(pool->config.host) - 1);
115 12 pool->config.port = port;
116 12 pool->config.tls_enabled = tls_enabled;
117 12 pool->config.tls_verify = tls_verify;
118
119 // Pre-create connections
120
2/2
✓ Branch 0 taken 29 times.
✓ Branch 1 taken 12 times.
 41 for (int i = 0; i < pool_size; i++) {
121 29 backend_conn_t *conn = create_backend_connection(pool, &pool->config);
122
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 29 times.
 29 if (!conn) {
123 ✗ log_message(LOG_LEVEL_WARN, "Failed to create connection %d for %s:%d",
124 i, host, port);
125 ✗ break;
126 }
127
128 29 pool->connections[i] = conn;
129 29 atomic_fetch_add(&pool->size, 1);
130 29 atomic_fetch_add(&pool->idle_count, 1);
131
132 29 log_message(LOG_LEVEL_INFO, "Created backend connection %d for %s:%d",
133 i, host, port);
134 }
135
136
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 12 times.
 12 log_message(LOG_LEVEL_INFO, "Backend pool created for %s:%d (size=%d, tls=%s)",
137 12 host, port, atomic_load(&pool->size), tls_enabled ? "yes" : "no");
138
139 12 return pool;
140 }
141
142 12 void backend_pool_destroy(backend_pool_t *pool)
143 {
144
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 12 times.
 12 if (!pool) return;
145
146 12 log_message(LOG_LEVEL_INFO, "Destroying backend pool for %s:%d",
147 12 pool->backend_host, pool->backend_port);
148
149 12 pthread_mutex_lock(&pool->pool_lock);
150
151 12 int size = atomic_load(&pool->size);
152
2/2
✓ Branch 0 taken 29 times.
✓ Branch 1 taken 12 times.
 41 for (int i = 0; i < size; i++) {
153
1/2
✓ Branch 0 taken 29 times.
✗ Branch 1 not taken.
 29 if (pool->connections[i]) {
154 29 destroy_backend_connection(pool->connections[i]);
155 29 pool->connections[i] = NULL;
156 }
157 }
158
159 12 pthread_mutex_unlock(&pool->pool_lock);
160 12 pthread_mutex_destroy(&pool->pool_lock);
161 12 free(pool);
162 }
163
164 8 backend_conn_t* backend_pool_acquire(backend_pool_t *pool)
165 {
166
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8 times.
 8 if (!pool) return NULL;
167
168 8 pthread_mutex_lock(&pool->pool_lock);
169
170 8 int size = atomic_load(&pool->size);
171 8 time_t now = time(NULL);
172
173 // Find an idle, healthy connection
174
2/2
✓ Branch 0 taken 11 times.
✓ Branch 1 taken 1 times.
 12 for (int i = 0; i < size; i++) {
175 11 backend_conn_t *conn = pool->connections[i];
176
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 11 times.
 11 if (!conn) continue;
177
178 11 pthread_mutex_lock(&conn->lock);
179
180 11 bool in_use = atomic_load(&conn->in_use);
181
3/4
✓ Branch 0 taken 7 times.
✓ Branch 1 taken 4 times.
✓ Branch 2 taken 7 times.
✗ Branch 3 not taken.
 11 if (!in_use && conn->health != BACKEND_HEALTH_UNHEALTHY) {
182 // Check idle timeout
183 7 long last_used = atomic_load(&conn->last_used);
184
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7 times.
 7 if (now - last_used > pool->idle_timeout_sec) {
185 ✗ log_message(LOG_LEVEL_INFO, "Connection %d idle timeout, reconnecting", i);
186 ✗ http2_client_cleanup(&conn->client);
187 ✗ if (http2_client_init(&conn->client, &pool->config) == 0) {
188 ✗ if (connect_backend_connection(conn, &pool->config) == 0) {
189 ✗ atomic_store(&conn->last_used, now);
190 } else {
191 ✗ conn->health = BACKEND_HEALTH_UNHEALTHY;
192 ✗ pthread_mutex_unlock(&conn->lock);
193 ✗ continue;
194 }
195 } else {
196 ✗ conn->health = BACKEND_HEALTH_UNHEALTHY;
197 ✗ pthread_mutex_unlock(&conn->lock);
198 ✗ continue;
199 }
200 }
201
202 7 atomic_store(&conn->in_use, true);
203 7 atomic_fetch_add(&pool->active_count, 1);
204 7 atomic_fetch_sub(&pool->idle_count, 1);
205
206 7 pthread_mutex_unlock(&conn->lock);
207 7 pthread_mutex_unlock(&pool->pool_lock);
208
209 H2C_LOG("backend_pool: acquired connection %d", i);
210 7 return conn;
211 }
212
213 4 pthread_mutex_unlock(&conn->lock);
214 }
215
216 1 pthread_mutex_unlock(&pool->pool_lock);
217
218 1 log_message(LOG_LEVEL_WARN, "No available connections in pool for %s:%d",
219 1 pool->backend_host, pool->backend_port);
220 1 return NULL;
221 }
222
223 7 void backend_pool_release(backend_conn_t *conn)
224 {
225
2/4
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 7 times.
 7 if (!conn || !conn->pool) return;
226
227 7 backend_pool_t *pool = conn->pool;
228
229 7 pthread_mutex_lock(&conn->lock);
230
231 7 atomic_store(&conn->in_use, false);
232 7 atomic_store(&conn->last_used, time(NULL));
233
234 7 pthread_mutex_unlock(&conn->lock);
235
236 7 atomic_fetch_sub(&pool->active_count, 1);
237 7 atomic_fetch_add(&pool->idle_count, 1);
238
239 H2C_LOG("backend_pool: released connection");
240 }
241
242 4 void backend_pool_mark_success(backend_conn_t *conn)
243 {
244
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 4 times.
 4 if (!conn) return;
245
246 4 pthread_mutex_lock(&conn->lock);
247
248 4 conn->consecutive_failures = 0;
249 4 conn->consecutive_successes++;
250
251
3/4
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 4 if (conn->consecutive_successes >= 2 && conn->health != BACKEND_HEALTH_HEALTHY) {
252 2 conn->health = BACKEND_HEALTH_HEALTHY;
253 2 log_message(LOG_LEVEL_INFO, "Backend connection marked healthy");
254 }
255
256 4 pthread_mutex_unlock(&conn->lock);
257 }
258
259 3 void backend_pool_mark_failure(backend_conn_t *conn)
260 {
261
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
 3 if (!conn) return;
262
263 3 pthread_mutex_lock(&conn->lock);
264
265 3 conn->consecutive_successes = 0;
266 3 conn->consecutive_failures++;
267
268
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 2 times.
 3 if (conn->consecutive_failures >= 3) {
269 1 conn->health = BACKEND_HEALTH_UNHEALTHY;
270 1 log_message(LOG_LEVEL_WARN, "Backend connection marked unhealthy after %d failures",
271 conn->consecutive_failures);
272 }
273
274 3 pthread_mutex_unlock(&conn->lock);
275 }
276
277 ✗ backend_health_t backend_pool_get_health(backend_conn_t *conn)
278 {
279 ✗ if (!conn) return BACKEND_HEALTH_UNKNOWN;
280
281 ✗ pthread_mutex_lock(&conn->lock);
282 ✗ backend_health_t health = conn->health;
283 ✗ pthread_mutex_unlock(&conn->lock);
284
285 ✗ return health;
286 }
287
288 ✗ int backend_pool_get_active_count(backend_pool_t *pool)
289 {
290 ✗ if (!pool) return 0;
291 ✗ return atomic_load(&pool->active_count);
292 }
293
294 ✗ int backend_pool_get_idle_count(backend_pool_t *pool)
295 {
296 ✗ if (!pool) return 0;
297 ✗ return atomic_load(&pool->idle_count);
298 }
299
300 3 int backend_pool_get_healthy_count(backend_pool_t *pool)
301 {
302
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
 3 if (!pool) return 0;
303
304 3 int healthy = 0;
305 3 int size = atomic_load(&pool->size);
306
307
2/2
✓ Branch 0 taken 9 times.
✓ Branch 1 taken 3 times.
 12 for (int i = 0; i < size; i++) {
308 9 backend_conn_t *conn = pool->connections[i];
309
3/4
✓ Branch 0 taken 9 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✓ Branch 3 taken 8 times.
 9 if (conn && conn->health == BACKEND_HEALTH_HEALTHY) {
310 1 healthy++;
311 }
312 }
313
314 3 return healthy;
315 }
316
317 ✗ bool backend_pool_has_healthy_connection(backend_pool_t *pool)
318 {
319 ✗ return backend_pool_get_healthy_count(pool) > 0;
320 }
321
322 ✗ static void handle_health_check_success(health_checker_t *checker, backend_conn_t *conn)
323 {
324 ✗ int prev_failures = atomic_load(&checker->consecutive_failures);
325 ✗ atomic_store(&checker->consecutive_failures, 0);
326 ✗ int successes = atomic_fetch_add(&checker->consecutive_successes, 1) + 1;
327
328 ✗ backend_pool_mark_success(conn);
329
330 ✗ if (prev_failures >= checker->config.unhealthy_threshold &&
331 ✗ successes >= checker->config.healthy_threshold) {
332 ✗ backend_health_t old_health = atomic_load(&checker->health);
333 ✗ if (old_health != BACKEND_HEALTH_HEALTHY) {
334 ✗ atomic_store(&checker->health, BACKEND_HEALTH_HEALTHY);
335 ✗ log_message(LOG_LEVEL_INFO, "Health check: backend marked HEALTHY");
336 }
337 }
338 ✗ }
339
340 ✗ static void handle_health_check_failure(health_checker_t *checker, backend_conn_t *conn)
341 {
342 ✗ int prev_successes = atomic_load(&checker->consecutive_successes);
343 ✗ atomic_store(&checker->consecutive_successes, 0);
344 ✗ int failures = atomic_fetch_add(&checker->consecutive_failures, 1) + 1;
345
346 ✗ backend_pool_mark_failure(conn);
347
348 ✗ if (prev_successes >= checker->config.healthy_threshold &&
349 ✗ failures >= checker->config.unhealthy_threshold) {
350 ✗ backend_health_t old_health = atomic_load(&checker->health);
351 ✗ if (old_health != BACKEND_HEALTH_UNHEALTHY) {
352 ✗ atomic_store(&checker->health, BACKEND_HEALTH_UNHEALTHY);
353 ✗ log_message(LOG_LEVEL_WARN, "Health check: backend marked UNHEALTHY after %d failures",
354 failures);
355 }
356 }
357 ✗ }
358
359 ✗ static bool perform_health_check(health_checker_t *checker, backend_conn_t *conn)
360 {
361 ✗ const char *health_path = checker->config.path;
362 ✗ if (health_path[0] == '\0') {
363 ✗ health_path = "/health";
364 }
365
366 ✗ int status = http2_client_send_request(&conn->client, "GET",
367 health_path, "health-check", NULL, 0);
368
369 ✗ if (status >= 0) {
370 ✗ int response_status = http2_client_recv_response(&conn->client);
371 ✗ if (response_status >= HTTP_STATUS_SUCCESS_MIN && response_status < HTTP_STATUS_CLIENT_ERROR_MIN) {
372 ✗ return true;
373 } else {
374 ✗ log_message(LOG_LEVEL_WARN, "Health check: backend returned status %d",
375 response_status);
376 }
377 } else {
378 ✗ log_message(LOG_LEVEL_WARN, "Health check: failed to send/receive");
379 }
380 ✗ return false;
381 }
382
383 ✗ static void *health_check_thread(void *arg)
384 {
385 ✗ health_checker_t *checker = (health_checker_t *)arg;
386 ✗ if (!checker || !checker->pool) {
387 ✗ return NULL;
388 }
389
390 ✗ log_message(LOG_LEVEL_DEBUG, "Health check thread started");
391
392 ✗ while (atomic_load(&checker->state) == HEALTH_CHECK_STATE_RUNNING) {
393 ✗ backend_conn_t *conn = backend_pool_acquire(checker->pool);
394 ✗ if (!conn) {
395 ✗ log_message(LOG_LEVEL_WARN, "Health check: failed to acquire connection");
396 ✗ atomic_fetch_add(&checker->consecutive_failures, 1);
397 ✗ atomic_store(&checker->consecutive_successes, 0);
398 ✗ atomic_store(&checker->health, BACKEND_HEALTH_UNHEALTHY);
399 ✗ atomic_fetch_add(&checker->failed_checks, 1);
400 ✗ atomic_fetch_add(&checker->total_checks, 1);
401 ✗ atomic_store(&checker->last_check_time, (long)time(NULL));
402 ✗ sleep(checker->config.interval_seconds);
403 ✗ continue;
404 }
405
406 ✗ bool check_success = perform_health_check(checker, conn);
407
408 ✗ if (check_success) {
409 ✗ handle_health_check_success(checker, conn);
410 } else {
411 ✗ handle_health_check_failure(checker, conn);
412 }
413
414 ✗ atomic_fetch_add(&checker->total_checks, 1);
415 ✗ if (!check_success) {
416 ✗ atomic_fetch_add(&checker->failed_checks, 1);
417 }
418 ✗ atomic_store(&checker->last_check_time, (long)time(NULL));
419
420 ✗ backend_pool_release(conn);
421 ✗ backend_pool_update_metrics(checker->pool);
422 ✗ sleep(checker->config.interval_seconds);
423 }
424
425 ✗ log_message(LOG_LEVEL_DEBUG, "Health check thread stopped");
426 ✗ return NULL;
427 }
428
429 ✗ static int health_checker_start(health_checker_t *checker, backend_pool_t *pool,
430 health_check_config_t *config)
431 {
432 ✗ if (!checker || !pool || !config || !config->enabled) {
433 ✗ return -1;
434 }
435
436 ✗ memset(checker, 0, sizeof(*checker));
437 ✗ checker->pool = pool;
438 ✗ memcpy(&checker->config, config, sizeof(*config));
439 ✗ atomic_store(&checker->state, HEALTH_CHECK_STATE_RUNNING);
440 ✗ atomic_store(&checker->health, BACKEND_HEALTH_UNKNOWN);
441 ✗ atomic_store(&checker->last_check_time, 0);
442
443 ✗ if (pthread_create(&checker->thread, NULL, health_check_thread, checker) != 0) {
444 ✗ log_message(LOG_LEVEL_ERROR, "Failed to create health check thread");
445 ✗ return -1;
446 }
447
448 ✗ log_message(LOG_LEVEL_INFO, "Health checker started (interval=%ds, path=%s)",
449 ✗ checker->config.interval_seconds, checker->config.path);
450 ✗ return 0;
451 }
452
453 ✗ static void health_checker_stop(health_checker_t *checker)
454 {
455 ✗ if (!checker) {
456 ✗ return;
457 }
458
459 ✗ atomic_store(&checker->state, HEALTH_CHECK_STATE_STOPPED);
460
461 ✗ if (checker->thread) {
462 ✗ pthread_join(checker->thread, NULL);
463 }
464
465 ✗ log_message(LOG_LEVEL_DEBUG, "Health checker stopped");
466 }
467
468 ✗ static backend_health_t health_checker_get_health(health_checker_t *checker)
469 {
470 ✗ if (!checker) {
471 ✗ return BACKEND_HEALTH_UNKNOWN;
472 }
473 ✗ return atomic_load(&checker->health);
474 }
475
476 ✗ int backend_pool_start_health_checker(backend_pool_t *pool, health_check_config_t *config)
477 {
478 ✗ if (!pool || !config || !config->enabled) {
479 ✗ return -1;
480 }
481
482 ✗ pool->health_check_enabled = true;
483
484 ✗ if (health_checker_start(&pool->health_checker, pool, config) != 0) {
485 ✗ log_message(LOG_LEVEL_ERROR, "Failed to start health checker for %s:%d",
486 ✗ pool->backend_host, pool->backend_port);
487 ✗ return -1;
488 }
489
490 ✗ return 0;
491 }
492
493 ✗ void backend_pool_stop_health_checker(backend_pool_t *pool)
494 {
495 ✗ if (!pool || !pool->health_check_enabled) {
496 ✗ return;
497 }
498
499 ✗ health_checker_stop(&pool->health_checker);
500 ✗ pool->health_check_enabled = false;
501 }
502
503 ✗ backend_health_t backend_pool_get_overall_health(backend_pool_t *pool)
504 {
505 ✗ if (!pool) {
506 ✗ return BACKEND_HEALTH_UNKNOWN;
507 }
508
509 ✗ if (pool->health_check_enabled) {
510 ✗ return health_checker_get_health(&pool->health_checker);
511 }
512
513 ✗ return backend_pool_get_healthy_count(pool) > 0 ?
514 ✗ BACKEND_HEALTH_HEALTHY : BACKEND_HEALTH_UNKNOWN;
515 }
516
517 1 void backend_pool_update_metrics(backend_pool_t *pool)
518 {
519
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (!pool) {
520 ✗ return;
521 }
522
523 1 int active = atomic_load(&pool->active_count);
524 1 int idle = atomic_load(&pool->idle_count);
525 1 int healthy = backend_pool_get_healthy_count(pool);
526
527 1 metrics_set_backend_pool_stats(pool->backend_host, active, idle, healthy);
528
529
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (pool->health_check_enabled) {
530 ✗ health_checker_t *hc = &pool->health_checker;
531 ✗ metrics_set_health_checker_stats(pool->backend_host,
532 ✗ (int)atomic_load(&hc->total_checks),
533 ✗ (int)atomic_load(&hc->failed_checks),
534 ✗ (int)atomic_load(&hc->health),
535 ✗ atomic_load(&hc->last_check_time));
536 }
537
538
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (pool->circuit_breaker_enabled) {
539 ✗ circuit_breaker_t *cb = &pool->circuit_breaker;
540 ✗ metrics_set_circuit_breaker_stats(pool->backend_host,
541 ✗ (int)atomic_load(&cb->state),
542 ✗ atomic_load(&cb->failure_count),
543 ✗ atomic_load(&cb->total_opens),
544 ✗ atomic_load(&cb->total_closes));
545 }
546 }
547
548 7 int backend_pool_init_circuit_breaker(backend_pool_t *pool, circuit_breaker_config_t *config)
549 {
550
4/6
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 7 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1 times.
✓ Branch 5 taken 6 times.
 7 if (!pool || !config || !config->enabled) {
551 1 return -1;
552 }
553
554 6 memset(&pool->circuit_breaker, 0, sizeof(pool->circuit_breaker));
555 6 memcpy(&pool->circuit_breaker.config, config, sizeof(*config));
556 6 atomic_store(&pool->circuit_breaker.state, CIRCUIT_BREAKER_CLOSED);
557 6 atomic_store(&pool->circuit_breaker.failure_count, 0);
558 6 atomic_store(&pool->circuit_breaker.success_count, 0);
559 6 atomic_store(&pool->circuit_breaker.last_failure_time, 0);
560 6 atomic_store(&pool->circuit_breaker.last_state_change, time(NULL));
561 6 atomic_store(&pool->circuit_breaker.total_opens, 0);
562 6 atomic_store(&pool->circuit_breaker.total_closes, 0);
563 6 pool->circuit_breaker_enabled = true;
564
565 6 log_message(LOG_LEVEL_INFO, "Circuit breaker initialized for %s:%d (threshold=%d, recovery=%ds)",
566 6 pool->backend_host, pool->backend_port,
567 config->failure_threshold, config->recovery_timeout_seconds);
568 6 return 0;
569 }
570
571 6 void backend_pool_destroy_circuit_breaker(backend_pool_t *pool)
572 {
573
2/4
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
 6 if (!pool || !pool->circuit_breaker_enabled) {
574 ✗ return;
575 }
576
577 6 pool->circuit_breaker_enabled = false;
578 6 log_message(LOG_LEVEL_DEBUG, "Circuit breaker destroyed for %s:%d",
579 6 pool->backend_host, pool->backend_port);
580 }
581
582 7 bool backend_pool_circuit_breaker_allow_request(backend_pool_t *pool)
583 {
584
3/4
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✓ Branch 3 taken 6 times.
 7 if (!pool || !pool->circuit_breaker_enabled) {
585 1 return true;
586 }
587
588 6 circuit_breaker_state_t state = atomic_load(&pool->circuit_breaker.state);
589
590
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 4 times.
 6 if (state == CIRCUIT_BREAKER_CLOSED) {
591 2 return true;
592 }
593
594
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 if (state == CIRCUIT_BREAKER_OPEN) {
595 4 long now = time(NULL);
596 4 long last_failure = atomic_load(&pool->circuit_breaker.last_failure_time);
597 4 int recovery_timeout = pool->circuit_breaker.config.recovery_timeout_seconds;
598
599
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 1 times.
 4 if (now - last_failure >= recovery_timeout) {
600 3 atomic_store(&pool->circuit_breaker.state, CIRCUIT_BREAKER_HALF_OPEN);
601 3 atomic_store(&pool->circuit_breaker.last_state_change, now);
602 3 log_message(LOG_LEVEL_INFO, "Circuit breaker transitioning to HALF-OPEN for %s:%d",
603 3 pool->backend_host, pool->backend_port);
604 3 return true;
605 }
606 1 return false;
607 }
608
609 ✗ if (state == CIRCUIT_BREAKER_HALF_OPEN) {
610 ✗ return true;
611 }
612
613 ✗ return false;
614 }
615
616 2 void backend_pool_circuit_breaker_record_success(backend_pool_t *pool)
617 {
618
2/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 2 times.
 2 if (!pool || !pool->circuit_breaker_enabled) {
619 ✗ return;
620 }
621
622 2 circuit_breaker_state_t state = atomic_load(&pool->circuit_breaker.state);
623
624
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 if (state == CIRCUIT_BREAKER_HALF_OPEN) {
625 2 int successes = atomic_fetch_add(&pool->circuit_breaker.success_count, 1) + 1;
626
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 1 times.
 2 if (successes >= 2) {
627 1 atomic_store(&pool->circuit_breaker.state, CIRCUIT_BREAKER_CLOSED);
628 1 atomic_store(&pool->circuit_breaker.failure_count, 0);
629 1 atomic_store(&pool->circuit_breaker.success_count, 0);
630 1 atomic_store(&pool->circuit_breaker.last_state_change, time(NULL));
631 1 atomic_fetch_add(&pool->circuit_breaker.total_closes, 1);
632 1 log_message(LOG_LEVEL_INFO, "Circuit breaker CLOSED for %s:%d after successful recovery",
633 1 pool->backend_host, pool->backend_port);
634 }
635 ✗ } else if (state == CIRCUIT_BREAKER_CLOSED) {
636 ✗ atomic_store(&pool->circuit_breaker.failure_count, 0);
637 }
638 }
639
640 12 void backend_pool_circuit_breaker_record_failure(backend_pool_t *pool)
641 {
642
3/4
✓ Branch 0 taken 12 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✓ Branch 3 taken 11 times.
 12 if (!pool || !pool->circuit_breaker_enabled) {
643 1 return;
644 }
645
646 11 circuit_breaker_state_t state = atomic_load(&pool->circuit_breaker.state);
647 11 long now = time(NULL);
648
649
3/4
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 10 times.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
 11 if (state == CIRCUIT_BREAKER_CLOSED || state == CIRCUIT_BREAKER_HALF_OPEN) {
650 11 int failures = atomic_fetch_add(&pool->circuit_breaker.failure_count, 1) + 1;
651 11 atomic_store(&pool->circuit_breaker.last_failure_time, now);
652
653
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 10 times.
 11 if (state == CIRCUIT_BREAKER_HALF_OPEN) {
654 1 atomic_store(&pool->circuit_breaker.state, CIRCUIT_BREAKER_OPEN);
655 1 atomic_store(&pool->circuit_breaker.last_state_change, now);
656 1 atomic_fetch_add(&pool->circuit_breaker.total_opens, 1);
657 1 log_message(LOG_LEVEL_WARN, "Circuit breaker OPENED for %s:%d (failure in half-open state)",
658 1 pool->backend_host, pool->backend_port);
659
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 6 times.
 10 } else if (failures >= pool->circuit_breaker.config.failure_threshold) {
660 4 atomic_store(&pool->circuit_breaker.state, CIRCUIT_BREAKER_OPEN);
661 4 atomic_store(&pool->circuit_breaker.last_state_change, now);
662 4 atomic_fetch_add(&pool->circuit_breaker.total_opens, 1);
663 4 log_message(LOG_LEVEL_WARN, "Circuit breaker OPENED for %s:%d after %d failures",
664 4 pool->backend_host, pool->backend_port, failures);
665 }
666 }
667 }
668
669 10 circuit_breaker_state_t backend_pool_circuit_breaker_get_state(backend_pool_t *pool)
670 {
671
3/4
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✓ Branch 3 taken 9 times.
 10 if (!pool || !pool->circuit_breaker_enabled) {
672 1 return CIRCUIT_BREAKER_CLOSED;
673 }
674 9 return atomic_load(&pool->circuit_breaker.state);
675 }
676
677 1 int backend_pool_circuit_breaker_get_failure_count(backend_pool_t *pool)
678 {
679
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (!pool) {
680 ✗ return 0;
681 }
682 1 return atomic_load(&pool->circuit_breaker.failure_count);
683 }
684
685 2 long backend_pool_circuit_breaker_get_total_opens(backend_pool_t *pool)
686 {
687
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 if (!pool) {
688 ✗ return 0;
689 }
690 2 return atomic_load(&pool->circuit_breaker.total_opens);
691 }
692
693 1 long backend_pool_circuit_breaker_get_total_closes(backend_pool_t *pool)
694 {
695
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (!pool) {
696 ✗ return 0;
697 }
698 1 return atomic_load(&pool->circuit_breaker.total_closes);
699 }
700