254 lines
8.6 KiB
C
254 lines
8.6 KiB
C
/*
|
|
* Caching Inbox Poller
|
|
*
|
|
* Periodically dequeues events from the PostgreSQL `caching_event_inbox` table
|
|
* and feeds them through the shared ingest_event() pipeline with
|
|
* EVENT_SOURCE_CACHING_INBOX. The poller runs on the main libwebsockets
|
|
* service thread (no separate thread) and is compiled only for the PostgreSQL
|
|
* backend. For SQLite builds every entry point is a no-op.
|
|
*
|
|
* The poller maintains a simple two-state machine:
|
|
* POLLER_IDLE - slow poll interval (no rows were found last cycle)
|
|
* POLLER_ACTIVE - fast poll interval (a full batch was dequeued last cycle)
|
|
*/
|
|
|
|
#include "caching_inbox_poller.h"
|
|
|
|
#include <stddef.h>
|
|
#include <string.h>
|
|
#include <time.h>
|
|
|
|
#include "debug.h"
|
|
#include "config.h"
|
|
#include "db_ops.h"
|
|
#include "main.h"
|
|
|
|
#ifdef DB_BACKEND_POSTGRES
|
|
#include <cjson/cJSON.h>
|
|
#endif
|
|
|
|
// ---- Configuration keys (read each tick) -------------------------------------
|
|
|
|
#define CFG_KEY_ENABLED "caching_inbox_enabled"
|
|
#define CFG_KEY_BATCH_SIZE "caching_inbox_batch_size"
|
|
#define CFG_KEY_ACTIVE_POLL_MS "caching_inbox_active_poll_ms"
|
|
#define CFG_KEY_IDLE_POLL_MS "caching_inbox_idle_poll_ms"
|
|
|
|
// ---- Defaults ----------------------------------------------------------------
|
|
|
|
#define DEFAULT_BATCH_SIZE 150
|
|
#define DEFAULT_ACTIVE_POLL_MS 200
|
|
#define DEFAULT_IDLE_POLL_MS 5000
|
|
|
|
// ---- State machine -----------------------------------------------------------
|
|
|
|
typedef enum {
|
|
POLLER_IDLE = 0,
|
|
POLLER_ACTIVE = 1
|
|
} poller_state_t;
|
|
|
|
// ---- Module state (PostgreSQL only; trivial for SQLite) ----------------------
|
|
|
|
#ifdef DB_BACKEND_POSTGRES
|
|
static poller_state_t g_state = POLLER_IDLE;
|
|
static long long g_last_poll_ms = 0; // monotonic-ish ms timestamp of last poll
|
|
static int g_initialized = 0;
|
|
|
|
// Statistics counters
|
|
static int g_total_dequeued = 0;
|
|
static int g_total_accepted = 0;
|
|
static int g_total_rejected = 0;
|
|
static int g_total_duplicates = 0;
|
|
static int g_last_batch_size = 0;
|
|
#endif
|
|
|
|
// ---- Helpers -----------------------------------------------------------------
|
|
|
|
#ifdef DB_BACKEND_POSTGRES
|
|
// Return a coarse millisecond timestamp using time(NULL) scaled to ms.
|
|
// We avoid clock_gettime(CLOCK_MONOTONIC) portability concerns and keep the
|
|
// resolution sufficient for the configured poll intervals (>= 200ms).
|
|
static long long now_ms(void) {
|
|
return (long long)time(NULL) * 1000LL;
|
|
}
|
|
#endif
|
|
|
|
// ---- Public API --------------------------------------------------------------
|
|
|
|
int caching_inbox_poller_init(void) {
|
|
#ifdef DB_BACKEND_POSTGRES
|
|
g_state = POLLER_IDLE;
|
|
g_last_poll_ms = 0;
|
|
g_total_dequeued = 0;
|
|
g_total_accepted = 0;
|
|
g_total_rejected = 0;
|
|
g_total_duplicates = 0;
|
|
g_last_batch_size = 0;
|
|
g_initialized = 1;
|
|
DEBUG_LOG("caching_inbox_poller: initialized (PostgreSQL backend)");
|
|
#else
|
|
// SQLite build: no-op. The inbox feature is PostgreSQL-only.
|
|
#endif
|
|
return 0;
|
|
}
|
|
|
|
int caching_inbox_poller_tick(void) {
|
|
#ifdef DB_BACKEND_POSTGRES
|
|
if (!g_initialized) {
|
|
return 0;
|
|
}
|
|
|
|
// Read configuration each tick (kept simple; values are cheap to read).
|
|
int enabled = get_config_bool(CFG_KEY_ENABLED, 0);
|
|
if (!enabled) {
|
|
return 0;
|
|
}
|
|
|
|
int batch_size = get_config_int(CFG_KEY_BATCH_SIZE, DEFAULT_BATCH_SIZE);
|
|
if (batch_size <= 0) {
|
|
batch_size = DEFAULT_BATCH_SIZE;
|
|
}
|
|
int active_poll_ms = get_config_int(CFG_KEY_ACTIVE_POLL_MS, DEFAULT_ACTIVE_POLL_MS);
|
|
if (active_poll_ms < 0) {
|
|
active_poll_ms = DEFAULT_ACTIVE_POLL_MS;
|
|
}
|
|
int idle_poll_ms = get_config_int(CFG_KEY_IDLE_POLL_MS, DEFAULT_IDLE_POLL_MS);
|
|
if (idle_poll_ms < 0) {
|
|
idle_poll_ms = DEFAULT_IDLE_POLL_MS;
|
|
}
|
|
|
|
// Throttle: only poll when the configured interval has elapsed.
|
|
long long now = now_ms();
|
|
long long interval_ms = (g_state == POLLER_ACTIVE) ? active_poll_ms : idle_poll_ms;
|
|
if (g_last_poll_ms != 0 && (now - g_last_poll_ms) < interval_ms) {
|
|
return 0;
|
|
}
|
|
g_last_poll_ms = now;
|
|
|
|
// Dequeue a bounded batch in a single SQL transaction.
|
|
int count = 0;
|
|
cJSON* rows = db_caching_inbox_dequeue_batch(batch_size, &count);
|
|
if (count <= 0 || rows == NULL) {
|
|
// Empty queue (or dequeue failure). Back off to idle mode.
|
|
if (rows != NULL) {
|
|
// Defensive: count==0 but a non-null array was returned.
|
|
cJSON_Delete(rows);
|
|
}
|
|
if (g_state == POLLER_ACTIVE) {
|
|
DEBUG_TRACE("caching_inbox_poller: queue empty, switching to IDLE");
|
|
}
|
|
g_state = POLLER_IDLE;
|
|
g_last_batch_size = 0;
|
|
// Note: db_caching_inbox_dequeue_batch logs its own errors on failure;
|
|
// a NULL return with count==0 is the normal "empty queue" case.
|
|
return 0;
|
|
}
|
|
|
|
DEBUG_LOG("caching_inbox_poller: dequeued %d event(s) (batch_size=%d, state=%s)",
|
|
count, batch_size, g_state == POLLER_ACTIVE ? "ACTIVE" : "IDLE");
|
|
|
|
g_last_batch_size = count;
|
|
g_total_dequeued += count;
|
|
|
|
// Feed each dequeued event through the shared ingestion pipeline.
|
|
cJSON* row = NULL;
|
|
cJSON_ArrayForEach(row, rows) {
|
|
cJSON* event_json_obj = cJSON_GetObjectItemCaseSensitive(row, "event_json");
|
|
if (!event_json_obj || !cJSON_IsString(event_json_obj)) {
|
|
DEBUG_WARN("caching_inbox_poller: row missing event_json string; skipping");
|
|
g_total_rejected++;
|
|
continue;
|
|
}
|
|
|
|
const char* event_json = cJSON_GetStringValue(event_json_obj);
|
|
size_t event_json_len = strlen(event_json);
|
|
|
|
int rc = ingest_event(event_json, event_json_len,
|
|
EVENT_SOURCE_CACHING_INBOX, NULL, NULL);
|
|
if (rc == 0) {
|
|
g_total_accepted++;
|
|
} else {
|
|
g_total_rejected++;
|
|
DEBUG_WARN("caching_inbox_poller: ingest_event rejected an event (rc=%d)", rc);
|
|
}
|
|
}
|
|
|
|
cJSON_Delete(rows);
|
|
|
|
// State transition: a full batch suggests more work is waiting.
|
|
if (count >= batch_size) {
|
|
if (g_state != POLLER_ACTIVE) {
|
|
DEBUG_TRACE("caching_inbox_poller: full batch, switching to ACTIVE");
|
|
}
|
|
g_state = POLLER_ACTIVE;
|
|
} else {
|
|
if (g_state != POLLER_IDLE) {
|
|
DEBUG_TRACE("caching_inbox_poller: partial batch, switching to IDLE");
|
|
}
|
|
g_state = POLLER_IDLE;
|
|
}
|
|
|
|
return 0;
|
|
#else
|
|
// SQLite build: no-op.
|
|
return 0;
|
|
#endif
|
|
}
|
|
|
|
void caching_inbox_poller_shutdown(void) {
|
|
#ifdef DB_BACKEND_POSTGRES
|
|
DEBUG_LOG("caching_inbox_poller: shutdown (dequeued=%d accepted=%d rejected=%d)",
|
|
g_total_dequeued, g_total_accepted, g_total_rejected);
|
|
g_initialized = 0;
|
|
g_state = POLLER_IDLE;
|
|
g_last_poll_ms = 0;
|
|
g_last_batch_size = 0;
|
|
#else
|
|
// SQLite build: no-op.
|
|
#endif
|
|
}
|
|
|
|
void caching_inbox_poller_get_stats(int* out_total_dequeued,
|
|
int* out_total_accepted,
|
|
int* out_total_rejected,
|
|
int* out_total_duplicates,
|
|
int* out_last_batch_size,
|
|
int* out_pending_live,
|
|
int* out_pending_backfill,
|
|
int* out_oldest_age_seconds) {
|
|
#ifdef DB_BACKEND_POSTGRES
|
|
if (out_total_dequeued) *out_total_dequeued = g_total_dequeued;
|
|
if (out_total_accepted) *out_total_accepted = g_total_accepted;
|
|
if (out_total_rejected) *out_total_rejected = g_total_rejected;
|
|
if (out_total_duplicates) *out_total_duplicates = g_total_duplicates;
|
|
if (out_last_batch_size) *out_last_batch_size = g_last_batch_size;
|
|
|
|
// Pending counts and oldest age come from the database; query on demand.
|
|
int live = 0, backfill = 0, oldest = 0;
|
|
if (out_pending_live || out_pending_backfill) {
|
|
if (db_caching_inbox_pending_counts(&live, &backfill) != 0) {
|
|
live = 0;
|
|
backfill = 0;
|
|
}
|
|
}
|
|
if (out_pending_live) *out_pending_live = live;
|
|
if (out_pending_backfill) *out_pending_backfill = backfill;
|
|
|
|
if (out_oldest_age_seconds) {
|
|
if (db_caching_inbox_oldest_age(&oldest) != 0) {
|
|
oldest = 0;
|
|
}
|
|
*out_oldest_age_seconds = oldest;
|
|
}
|
|
#else
|
|
if (out_total_dequeued) *out_total_dequeued = 0;
|
|
if (out_total_accepted) *out_total_accepted = 0;
|
|
if (out_total_rejected) *out_total_rejected = 0;
|
|
if (out_total_duplicates) *out_total_duplicates = 0;
|
|
if (out_last_batch_size) *out_last_batch_size = 0;
|
|
if (out_pending_live) *out_pending_live = 0;
|
|
if (out_pending_backfill) *out_pending_backfill = 0;
|
|
if (out_oldest_age_seconds) *out_oldest_age_seconds = 0;
|
|
#endif
|
|
}
|