feat: add test_dst_backpressure to verify outbox behavior under 503 errors

Author: FridayCandour

CMakeLists.txt

@@ -245,7 +245,8 @@ if(ARKILIAN_BUILD_TESTS)
         endforeach()
 
         if(NOT WIN32)
-            foreach(t test_kill_switch test_kill_resilience test_load_contention)
+            foreach(t test_kill_switch test_kill_resilience test_load_contention
+                     test_dst_backpressure)
                 ark_add_test(${t} tests/${t}.c)
             endforeach()
         endif()

tests/run_all.sh

@@ -62,8 +62,9 @@ run      test_basic           tests/test_basic.c
 run      test_interception    tests/test_interception.c
 run      test_regressions     tests/test_regressions.c
 run      test_kill_switch     tests/test_kill_switch.c
-run      test_load_contention tests/test_load_contention.c
 run      test_kill_resilience tests/test_kill_resilience.c
+run      test_load_contention tests/test_load_contention.c
+run      test_dst_backpressure tests/test_dst_backpressure.c
 run      test_monitoring      tests/test_monitoring.c
 run      test_virtual_tables  tests/test_virtual_tables.c
 run      test_deterministic   tests/test_deterministic.c

tests/test_dst_backpressure.c

@@ -0,0 +1,349 @@
+// Arkilian Destination Backpressure Tests — 5,000-business launch
+// verification (Checklist #1: "If the backend returns 429/503 under load,
+// verify that client nodes gracefully buffer up to ARKILIAN_MAX_QUEUE_DEPTH
+// without dropping local database performance".)
+//
+// A mock HTTP server that always returns 503 Service Unavailable stands in
+// for an overwhelmed control plane. The test proves:
+//
+//   1. Every application write succeeds (spec §0: backup never breaks the
+//      app) — even while the flush thread is retrying against the 503.
+//   2. Captured rows accumulate in _pending_backup (nothing is deleted on
+//      a non-2xx response — at-least-once delivery, no silent data loss).
+//   3. Attempts climb but rows are NOT dead-lettered while the outage is
+//      transient — they stay queued and retry with exponential backoff.
+//   4. The outbox cap (ARKILIAN_MAX_QUEUE_DEPTH) is respected: capture
+//      pauses at the cap, but application writes keep succeeding.
+//   5. db_backup_is_healthy() == 0 (red) during the outage — the failure
+//      is visible to monitoring, not silent.
+//
+// POSIX-only: uses BSD sockets for the mock server (same pattern as
+// test_kill_resilience.c). Not compiled on MinGW (no <sys/socket.h>).
+//
+// Compile (macOS/Linux):
+//   cc tests/test_dst_backpressure.c src/class.c src/deps/sqlite/sqlite3.c \
+//      -Isrc -Isrc/deps/sqlite -lcurl -lpthread -o test_dst_backpressure
+
+#ifndef _POSIX_C_SOURCE
+#define _POSIX_C_SOURCE 200809L
+#endif
+#ifdef __APPLE__
+#ifndef _DARWIN_C_SOURCE
+#define _DARWIN_C_SOURCE
+#endif
+#endif
+#ifndef _DEFAULT_SOURCE
+#define _DEFAULT_SOURCE
+#endif
+
+#include "class.h"
+#include <assert.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <pthread.h>
+#include <arpa/inet.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+
+static int tests_run = 0;
+static int tests_passed = 0;
+
+#define RUN_TEST(fn)                                                           \
+  do {                                                                         \
+    tests_run++;                                                                \
+    printf("  [%02d] %-52s ", tests_run, #fn);                                 \
+    fflush(stdout);                                                             \
+    fn();                                                                       \
+    tests_passed++;                                                             \
+    printf("PASS\n");                                                           \
+  } while (0)
+
+static void cleanup(const char *path) {
+  remove(path);
+  char side[256];
+  snprintf(side, sizeof(side), "%s-wal", path); remove(side);
+  snprintf(side, sizeof(side), "%s-shm", path); remove(side);
+  snprintf(side, sizeof(side), "%s-journal", path); remove(side);
+}
+
+// ── Mock 503 destination ────────────────────────────────────────────
+// Always returns 503 Service Unavailable, simulating an overwhelmed
+// control-plane ingestion layer. Records request count so the test can
+// verify the flush thread IS retrying (not silently dead).
+
+typedef struct {
+  int listen_fd;
+  int port;
+  pthread_t thread;
+  volatile int stop;
+  volatile int requests;
+  volatile int return_503; // 1 = 503, 0 = 200 (for flip-to-healthy test)
+} mock_503_server;
+
+static void *mock_503_run(void *arg) {
+  mock_503_server *s = (mock_503_server *)arg;
+  for (;;) {
+    int fd = accept(s->listen_fd, NULL, NULL);
+    if (fd < 0) break;
+#ifdef SO_NOSIGPIPE
+    int on = 1;
+    setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &on, sizeof(on));
+#endif
+    char buf[16384];
+    ssize_t n = recv(fd, buf, sizeof(buf) - 1, 0);
+    if (s->stop) { close(fd); break; }
+    if (n > 0) {
+      buf[n] = '\0';
+      // Drain the full request body (Content-Length)
+      long body_len = 0;
+      char *cl = strstr(buf, "Content-Length:");
+      if (cl) body_len = atol(cl + 15);
+      char *hdr_end = strstr(buf, "\r\n\r\n");
+      long have = hdr_end ? n - (hdr_end + 4 - buf) : 0;
+      while (have < body_len) {
+        n = recv(fd, buf, sizeof(buf) - 1, 0);
+        if (n <= 0) break;
+        have += n;
+      }
+      s->requests++;
+      const char *resp;
+      if (s->return_503) {
+        resp = "HTTP/1.1 503 Service Unavailable\r\n"
+               "Content-Length: 0\r\n"
+               "Connection: close\r\n"
+               "Retry-After: 1\r\n"
+               "\r\n";
+      } else {
+        resp = "HTTP/1.1 200 OK\r\n"
+               "Content-Length: 2\r\n"
+               "Connection: close\r\n"
+               "\r\nOK";
+      }
+      send(fd, resp, strlen(resp), 0);
+    }
+    close(fd);
+  }
+  return NULL;
+}
+
+static int mock_503_start(mock_503_server *s) {
+  memset(s, 0, sizeof(*s));
+  s->return_503 = 1;
+  s->listen_fd = socket(AF_INET, SOCK_STREAM, 0);
+  if (s->listen_fd < 0) return -1;
+  int one = 1;
+  setsockopt(s->listen_fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
+  struct sockaddr_in addr;
+  memset(&addr, 0, sizeof(addr));
+  addr.sin_family = AF_INET;
+  addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+  addr.sin_port = 0;
+  if (bind(s->listen_fd, (struct sockaddr *)&addr, sizeof(addr)) != 0) return -1;
+  socklen_t alen = sizeof(addr);
+  if (getsockname(s->listen_fd, (struct sockaddr *)&addr, &alen) != 0) return -1;
+  s->port = ntohs(addr.sin_port);
+  if (listen(s->listen_fd, 64) != 0) return -1;
+  if (pthread_create(&s->thread, NULL, mock_503_run, s) != 0) return -1;
+  return 0;
+}
+
+static void mock_503_stop(mock_503_server *s) {
+  s->stop = 1;
+  // Kick the accept() loop
+  int fd = socket(AF_INET, SOCK_STREAM, 0);
+  if (fd >= 0) {
+    struct sockaddr_in addr;
+    memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = htons((unsigned short)s->port);
+    connect(fd, (struct sockaddr *)&addr, sizeof(addr));
+    close(fd);
+  }
+  pthread_join(s->thread, NULL);
+  close(s->listen_fd);
+}
+
+// ── Tests ───────────────────────────────────────────────────────────
+
+// 1. Application writes survive a 503-spewing destination. Every INSERT
+//    returns SQLITE_OK; rows accumulate in _pending_backup (none deleted
+//    on a non-2xx); the flush thread IS retrying (requests > 0).
+static void test_writes_survive_503_backpressure(void) {
+  const char *db_path = "test_bp_503.db";
+  cleanup(db_path);
+
+  mock_503_server srv;
+  assert(mock_503_start(&srv) == 0);
+
+  char url[128];
+  snprintf(url, sizeof(url), "http://127.0.0.1:%d/push", srv.port);
+  setenv("ARKILIAN_ENABLE_BACKUP", "1", 1);
+  setenv("ARKILIAN_API_KEY", "test-key", 1);
+  setenv("ARKILIAN_SKIP_STARTUP_AUTH", "1", 1);
+  setenv("ARKILIAN_CONTROL_URL", url, 1);
+  setenv("ARKILIAN_BACKUP_INTERVAL", "3600", 1);
+  setenv("ARKILIAN_MAX_ATTEMPTS", "100", 1); // don't dead-letter during test
+
+  arkilian *db = NULL;
+  assert(db_init(&db, db_path) == 0);
+  assert(db_exec(db, "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)") == SQLITE_OK);
+
+  // Write 20 rows while the destination returns 503.
+  for (int i = 0; i < 20; i++) {
+    char sql[64];
+    snprintf(sql, sizeof(sql), "INSERT INTO t (v) VALUES ('row%d')", i);
+    assert(db_exec(db, sql) == SQLITE_OK);
+  }
+
+  // Let the flush thread retry for a bit.
+  sleep(2);
+
+  // Every application write succeeded (spec §0: backup never breaks the app).
+  db_prepare(db, "SELECT COUNT(*) FROM t");
+  assert(db_step(db) == SQLITE_ROW);
+  assert(db_column_int(db, 0) == 20);
+  db_finalize(db);
+
+  // The flush thread IS retrying against the 503 — it didn't silently die.
+  assert(srv.requests > 0);
+
+  // Rows are NOT deleted on a non-2xx: they accumulate in _pending_backup.
+  // (1 DDL capture + 20 inserts = 21 rows minimum; none shipped.)
+  int depth = db_backup_queue_depth(db);
+  assert(depth >= 21);
+
+  // Health is red — the outage is visible to monitoring, not silent.
+  assert(db_backup_is_healthy(db) == 0);
+
+  db_close(db);
+  mock_503_stop(&srv);
+  cleanup(db_path);
+  setenv("ARKILIAN_MAX_ATTEMPTS", "3", 1); // restore for other tests
+}
+
+// 2. The outbox cap (ARKILIAN_MAX_QUEUE_DEPTH) is respected under 503
+//    backpressure: capture pauses at the cap, but application writes keep
+//    succeeding. No data is lost — the uncaptured writes are in the table;
+//    the cap only prevents the outbox from growing without bound.
+static void test_outbox_cap_respected_under_503(void) {
+  const char *db_path = "test_bp_cap.db";
+  cleanup(db_path);
+
+  mock_503_server srv;
+  assert(mock_503_start(&srv) == 0);
+
+  char url[128];
+  snprintf(url, sizeof(url), "http://127.0.0.1:%d/push", srv.port);
+  setenv("ARKILIAN_ENABLE_BACKUP", "1", 1);
+  setenv("ARKILIAN_API_KEY", "test-key", 1);
+  setenv("ARKILIAN_SKIP_STARTUP_AUTH", "1", 1);
+  setenv("ARKILIAN_CONTROL_URL", url, 1);
+  setenv("ARKILIAN_BACKUP_INTERVAL", "3600", 1);
+  setenv("ARKILIAN_MAX_ATTEMPTS", "100", 1);
+  setenv("ARKILIAN_MAX_QUEUE_DEPTH", "10", 1); // tight cap
+
+  arkilian *db = NULL;
+  assert(db_init(&db, db_path) == 0);
+  assert(db_exec(db, "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)") == SQLITE_OK);
+
+  // Write well past the cap. Every INSERT must still succeed.
+  for (int i = 0; i < 50; i++) {
+    char sql[64];
+    snprintf(sql, sizeof(sql), "INSERT INTO t (v) VALUES ('row%d')", i);
+    assert(db_exec(db, sql) == SQLITE_OK);
+  }
+
+  // All 50 application writes succeeded (spec §0).
+  db_prepare(db, "SELECT COUNT(*) FROM t");
+  assert(db_step(db) == SQLITE_ROW);
+  assert(db_column_int(db, 0) == 50);
+  db_finalize(db);
+
+  // The outbox is hard-capped at 10.
+  int depth = db_backup_queue_depth(db);
+  assert(depth >= 1 && depth <= 10);
+
+  // Health is red at cap.
+  assert(db_backup_is_healthy(db) == 0);
+
+  db_close(db);
+  mock_503_stop(&srv);
+  cleanup(db_path);
+  setenv("ARKILIAN_MAX_QUEUE_DEPTH", "100000", 1); // restore default
+  setenv("ARKILIAN_MAX_ATTEMPTS", "3", 1);
+}
+
+// 3. When the destination recovers (flips from 503 to 200), the backlog
+//    drains. This proves the client is a self-healing buffer: no restart,
+//    no operator intervention — the queue clears when the backend is back.
+static void test_backlog_drains_on_recovery(void) {
+  const char *db_path = "test_bp_recover.db";
+  cleanup(db_path);
+
+  mock_503_server srv;
+  assert(mock_503_start(&srv) == 0);
+
+  char url[128];
+  snprintf(url, sizeof(url), "http://127.0.0.1:%d/push", srv.port);
+  setenv("ARKILIAN_ENABLE_BACKUP", "1", 1);
+  setenv("ARKILIAN_API_KEY", "test-key", 1);
+  setenv("ARKILIAN_SKIP_STARTUP_AUTH", "1", 1);
+  setenv("ARKILIAN_CONTROL_URL", url, 1);
+  setenv("ARKILIAN_BACKUP_INTERVAL", "3600", 1);
+  setenv("ARKILIAN_MAX_ATTEMPTS", "100", 1);
+
+  arkilian *db = NULL;
+  assert(db_init(&db, db_path) == 0);
+  assert(db_exec(db, "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)") == SQLITE_OK);
+
+  // Write 10 rows under 503 backpressure.
+  for (int i = 0; i < 10; i++) {
+    char sql[64];
+    snprintf(sql, sizeof(sql), "INSERT INTO t (v) VALUES ('row%d')", i);
+    assert(db_exec(db, sql) == SQLITE_OK);
+  }
+  sleep(1); // let the flush thread retry against 503
+
+  // Queue has accumulated (nothing shipped yet).
+  int depth_before = db_backup_queue_depth(db);
+  assert(depth_before >= 11); // 1 DDL + 10 inserts
+
+  // Destination recovers: flip the server from 503 to 200.
+  srv.return_503 = 0;
+  db_wal_flush(db); // wake the flush thread immediately
+
+  // Wait for the backlog to drain (at most ~15s with 100-row cap + backoff).
+  int waited = 0;
+  while (db_backup_queue_depth(db) > 0 && waited < 15000) {
+    usleep(200 * 1000);
+    waited += 200;
+  }
+  assert(db_backup_queue_depth(db) == 0);
+
+  // All rows were delivered (the mock server received them all).
+  assert(srv.requests > 10);
+
+  db_close(db);
+  mock_503_stop(&srv);
+  cleanup(db_path);
+  setenv("ARKILIAN_MAX_ATTEMPTS", "3", 1);
+}
+
+// ── Main ────────────────────────────────────────────────────────────
+
+int main(void) {
+  signal(SIGPIPE, SIG_IGN);
+  setenv("ARKILIAN_API_KEY", "test-key", 1);
+  setenv("ARKILIAN_SKIP_STARTUP_AUTH", "1", 1);
+  setenv("ARKILIAN_MAX_ATTEMPTS", "3", 1);
+
+  printf("=== Arkilian Destination Backpressure Tests ===\n\n");
+  RUN_TEST(test_writes_survive_503_backpressure);
+  RUN_TEST(test_outbox_cap_respected_under_503);
+  RUN_TEST(test_backlog_drains_on_recovery);
+  printf("\n=== Results: %d/%d passed ===\n", tests_passed, tests_run);
+  return (tests_passed == tests_run) ? 0 : 1;
+}