From 0e5bc515e3519df859b0cb3221161f62ca8d66e6 Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Sereno Date: Wed, 23 Sep 2026 21:04:48 -0300 Subject: [PATCH 1/2] fix(cron): deliver other due jobs while one remains unacked Deferred ack kept the earliest due row at the head of the queue, so a failed send or agent run blocked every later reminder until that job succeeded. --- src/core/memory.c | 35 ++++++++++++-- src/core/memory.h | 18 +++++++ src/tools/cron.c | 116 +++++++++++++++++++++++++++++++++++++++------- tests/test_cron.c | 48 +++++++++++++++++++ 4 files changed, 196 insertions(+), 21 deletions(-) diff --git a/src/core/memory.c b/src/core/memory.c index 6757d57..1c28d61 100644 --- a/src/core/memory.c +++ b/src/core/memory.c @@ -475,15 +475,29 @@ int cron_job_list(cron_job_row_t *out, int max_count) return count; } -int cron_job_get_next_due(long long now, cron_job_row_t *out) +static int cron_job_select_due(long long now, int use_cursor, long long after_run, + const char *after_id, cron_job_row_t *out) { - if (!g_db || !out) return -1; - const char *sql = "SELECT id, schedule, message, channel, recipient, next_run, enabled FROM cron_jobs " - "WHERE next_run <= ?1 AND enabled = 1 ORDER BY next_run ASC LIMIT 1"; + const char *sql; sqlite3_stmt *stmt = NULL; + int ret = 0; + + if (!g_db || !out) return -1; + if (use_cursor && (!after_id || after_id[0] == '\0')) return -1; + sql = use_cursor + ? "SELECT id, schedule, message, channel, recipient, next_run, enabled FROM cron_jobs " + "WHERE next_run <= ?1 AND enabled = 1 " + "AND (next_run > ?2 OR (next_run = ?2 AND id > ?3)) " + "ORDER BY next_run ASC, id ASC LIMIT 1" + : "SELECT id, schedule, message, channel, recipient, next_run, enabled FROM cron_jobs " + "WHERE next_run <= ?1 AND enabled = 1 " + "ORDER BY next_run ASC, id ASC LIMIT 1"; if (sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1; sqlite3_bind_int64(stmt, 1, now); - int ret = 0; + if (use_cursor) { + sqlite3_bind_int64(stmt, 2, after_run); + sqlite3_bind_text(stmt, 3, after_id, -1, SQLITE_TRANSIENT); + } if (sqlite3_step(stmt) == SQLITE_ROW) { if (fill_cron_job_row(stmt, out) != 0) { sqlite3_finalize(stmt); @@ -495,6 +509,17 @@ int cron_job_get_next_due(long long now, cron_job_row_t *out) return ret; } +int cron_job_get_next_due(long long now, cron_job_row_t *out) +{ + return cron_job_select_due(now, 0, 0, NULL, out); +} + +int cron_job_get_next_due_after(long long now, long long after_run, const char *after_id, + cron_job_row_t *out) +{ + return cron_job_select_due(now, 1, after_run, after_id, out); +} + int cron_job_get_by_id(const char *id, cron_job_row_t *out) { const char *sql; diff --git a/src/core/memory.h b/src/core/memory.h index 676a547..99cd0e1 100644 --- a/src/core/memory.h +++ b/src/core/memory.h @@ -192,12 +192,30 @@ int cron_job_list(cron_job_row_t *out, int max_count); /** * Get the next due job (next_run <= now, enabled). * + * Ties on next_run are ordered by id ascending. + * * @param now Current Unix timestamp. * @param out Filled with job data if found. Caller must cron_job_row_free(). * @return 1 if found, 0 if none, -1 on error. */ int cron_job_get_next_due(long long now, cron_job_row_t *out); +/** + * Get the next due job after a (next_run, id) cursor. + * + * Uses the same order as cron_job_get_next_due(). Example: + * if (cron_job_get_next_due_after(now, row.next_run, row.id, &next) == 1) + * cron_job_row_free(&next); + * + * @param now Current Unix timestamp. + * @param after_run next_run of the last row already visited. + * @param after_id Id of the last row already visited (must be non-empty). + * @param out Filled with job data if found. Caller must cron_job_row_free(). + * @return 1 if found, 0 if none, -1 on error. + */ +int cron_job_get_next_due_after(long long now, long long after_run, const char *after_id, + cron_job_row_t *out); + /** * Load a cron job by id. * diff --git a/src/tools/cron.c b/src/tools/cron.c index 6114621..9074954 100644 --- a/src/tools/cron.c +++ b/src/tools/cron.c @@ -178,26 +178,56 @@ static int cron_init(const config_t *cfg) return 0; } -static char s_offered_id[128]; -static struct timespec s_offered_mono; +#define CRON_OFFER_TRACK 16 -static void cron_wait_if_reoffer(const char *job_id, int timeout_ms) +typedef struct cron_offer_slot { + char id[128]; + struct timespec mono; +} cron_offer_slot_t; + +static cron_offer_slot_t s_offers[CRON_OFFER_TRACK]; + +static long cron_elapsed_ms(const struct timespec *then, const struct timespec *now) +{ + return (now->tv_sec - then->tv_sec) * 1000L + + (now->tv_nsec - then->tv_nsec) / 1000000L; +} + +static int cron_offer_age_ms(const char *job_id, const struct timespec *now, long *age_out) +{ + int i; + for (i = 0; i < CRON_OFFER_TRACK; i++) { + if (s_offers[i].id[0] == '\0' || strcmp(s_offers[i].id, job_id) != 0) + continue; + *age_out = cron_elapsed_ms(&s_offers[i].mono, now); + return 1; + } + return 0; +} + +static int cron_offer_is_hot(const char *job_id, int timeout_ms, const struct timespec *now) +{ + long age = 0; + if (timeout_ms <= 0 || !job_id) + return 0; + if (!cron_offer_age_ms(job_id, now, &age)) + return 0; + return age < (long)timeout_ms; +} + +static void cron_wait_remaining(const char *job_id, int timeout_ms) { struct timespec now; struct timespec remain; - long elapsed_ms; + long age = 0; long wait_ms; - if (timeout_ms <= 0 || !job_id || job_id[0] == '\0') - return; - if (s_offered_id[0] == '\0' || strcmp(s_offered_id, job_id) != 0) + if (timeout_ms <= 0 || !job_id) return; if (clock_gettime(CLOCK_MONOTONIC, &now) != 0) return; - elapsed_ms = (now.tv_sec - s_offered_mono.tv_sec) * 1000L - + (now.tv_nsec - s_offered_mono.tv_nsec) / 1000000L; - if (elapsed_ms >= timeout_ms) + if (!cron_offer_age_ms(job_id, &now, &age) || age >= (long)timeout_ms) return; - wait_ms = timeout_ms - elapsed_ms; + wait_ms = (long)timeout_ms - age; remain.tv_sec = wait_ms / 1000; remain.tv_nsec = (wait_ms % 1000) * 1000000L; nanosleep(&remain, NULL); @@ -205,10 +235,64 @@ static void cron_wait_if_reoffer(const char *job_id, int timeout_ms) static void cron_mark_offered(const char *job_id) { - if (!job_id) + struct timespec now; + int i; + int slot = -1; + int oldest = 0; + if (!job_id || job_id[0] == '\0') + return; + if (clock_gettime(CLOCK_MONOTONIC, &now) != 0) return; - snprintf(s_offered_id, sizeof(s_offered_id), "%s", job_id); - clock_gettime(CLOCK_MONOTONIC, &s_offered_mono); + for (i = 0; i < CRON_OFFER_TRACK; i++) { + if (strcmp(s_offers[i].id, job_id) == 0) { + s_offers[i].mono = now; + return; + } + if (slot < 0 && s_offers[i].id[0] == '\0') + slot = i; + if (cron_elapsed_ms(&s_offers[i].mono, &now) > + cron_elapsed_ms(&s_offers[oldest].mono, &now)) + oldest = i; + } + if (slot < 0) + slot = oldest; + snprintf(s_offers[slot].id, sizeof(s_offers[slot].id), "%s", job_id); + s_offers[slot].mono = now; +} + +/** Prefer a due job outside its re-offer window so one stuck job cannot hide the rest. */ +static int cron_pick_due_row(long long now, int timeout_ms, cron_job_row_t *row) +{ + struct timespec mono; + long long cursor_run = 0; + char cursor_id[128]; + int have_cursor = 0; + + if (clock_gettime(CLOCK_MONOTONIC, &mono) != 0) + memset(&mono, 0, sizeof(mono)); + cursor_id[0] = '\0'; + for (;;) { + int rc = have_cursor + ? cron_job_get_next_due_after(now, cursor_run, cursor_id, row) + : cron_job_get_next_due(now, row); + if (rc != 1) + break; + if (have_cursor && (row->next_run < cursor_run || + (row->next_run == cursor_run && strcmp(row->id, cursor_id) <= 0))) { + cron_job_row_free(row); + break; + } + if (!cron_offer_is_hot(row->id, timeout_ms, &mono)) + return 1; + cursor_run = row->next_run; + snprintf(cursor_id, sizeof(cursor_id), "%s", row->id); + have_cursor = 1; + cron_job_row_free(row); + } + if (cron_job_get_next_due(now, row) != 1) + return 0; + cron_wait_remaining(row->id, timeout_ms); + return 1; } static int cron_poll(channel_incoming_msg_t *out, int timeout_ms) @@ -219,8 +303,8 @@ static int cron_poll(channel_incoming_msg_t *out, int timeout_ms) if (!out) return -1; now = (long long)time(NULL); memset(&row, 0, sizeof(row)); - if (cron_job_get_next_due(now, &row) != 1) return 0; - cron_wait_if_reoffer(row.id, timeout_ms); + if (!cron_pick_due_row(now, timeout_ms, &row)) + return 0; memset(out, 0, sizeof(*out)); snprintf(session_id, sizeof(session_id), "%s:%s", row.channel[0] ? row.channel : "cli", diff --git a/tests/test_cron.c b/tests/test_cron.c index d213768..4589dd5 100644 --- a/tests/test_cron.c +++ b/tests/test_cron.c @@ -449,6 +449,53 @@ static int test_cron_job_rejects_oversized_text(void) return 0; } +/* An unacked earlier job must not hide another job that is also due. */ +static int test_cron_poll_offers_sibling_while_earlier_unacked(void) +{ + const char *path = "/tmp/shellclaw_test_cron_sibling.db"; + const channel_t *cron_ch; + channel_incoming_msg_t msg; + cron_job_row_t row; + struct timespec t0; + struct timespec t1; + long elapsed_ms; + long long now; + + remove(path); + ASSERT(memory_init(path) == 0); + now = (long long)time(NULL); + ASSERT(cron_job_create("job_a", "interval:60", "A", "cli", "default", now - 10, 1) == 0); + ASSERT(cron_job_create("job_b", "interval:60", "B", "cli", "default", now - 5, 1) == 0); + cron_ch = channel_cron_get(); + ASSERT(cron_ch != NULL && cron_ch->poll != NULL); + memset(&msg, 0, sizeof(msg)); + ASSERT(cron_ch->poll(&msg, 0) == 1); + ASSERT(msg.user_id != NULL); + ASSERT(strcmp(msg.user_id, "job_a") == 0); + channel_incoming_msg_clear(&msg); + memset(&msg, 0, sizeof(msg)); + ASSERT(clock_gettime(CLOCK_MONOTONIC, &t0) == 0); + ASSERT(cron_ch->poll(&msg, 2000) == 1); + ASSERT(clock_gettime(CLOCK_MONOTONIC, &t1) == 0); + elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000L + + (t1.tv_nsec - t0.tv_nsec) / 1000000L; + ASSERT(msg.user_id != NULL); + ASSERT(strcmp(msg.user_id, "job_b") == 0); + ASSERT(elapsed_ms < 400); + channel_incoming_msg_clear(&msg); + memset(&row, 0, sizeof(row)); + ASSERT(cron_job_get_by_id("job_a", &row) == 1); + ASSERT(row.next_run == now - 10); + cron_job_row_free(&row); + memset(&row, 0, sizeof(row)); + ASSERT(cron_job_get_by_id("job_b", &row) == 1); + ASSERT(row.next_run == now - 5); + cron_job_row_free(&row); + memory_cleanup(); + remove(path); + return 0; +} + int main(void) { RUN(test_interval_next_run()); @@ -469,6 +516,7 @@ int main(void) RUN(test_cron_ack_fail_closed_on_parse_error()); RUN(test_cron_poll_waits_before_reoffer()); RUN(test_cron_job_rejects_oversized_text()); + RUN(test_cron_poll_offers_sibling_while_earlier_unacked()); printf("test_cron: all tests passed\n"); return 0; } From e09b2e94ae856f1879a587d6aed509d1f32342cd Mon Sep 17 00:00:00 2001 From: Adrianno Esnarriaga Sereno Date: Wed, 23 Sep 2026 21:24:20 -0300 Subject: [PATCH 2/2] fix(cron): keep the re-offer wait when the offer table is full A full ring used to drop the oldest id, and the next poll treated that job as cold, so a down channel tight-looped agent_run. Reuse a slot only after its recorded window has elapsed, and wait on the earliest due job when every slot is still hot. --- src/core/memory.c | 26 +++++++++++- src/core/memory.h | 23 ++++++++++- src/tools/cron.c | 81 +++++++++++++++++++++++------------- tests/test_cron.c | 102 ++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 201 insertions(+), 31 deletions(-) diff --git a/src/core/memory.c b/src/core/memory.c index 1c28d61..9edaa7e 100644 --- a/src/core/memory.c +++ b/src/core/memory.c @@ -398,6 +398,8 @@ int cron_job_create(const char *id, const char *schedule, const char *message, const char *channel, const char *recipient, long long next_run, int enabled) { if (!g_db || !id || !schedule || !message) return -1; + /* fill_cron_job_row truncates id[128]; a longer id breaks the due cursor. */ + if (strlen(id) >= sizeof(((cron_job_row_t *)0)->id)) return -1; if (strlen(schedule) > (size_t)CRON_JOB_TEXT_MAX) return -1; if (strlen(message) > (size_t)CRON_JOB_TEXT_MAX) return -1; const char *ch = channel ? channel : ""; @@ -458,7 +460,8 @@ int cron_job_update_next_run(const char *id, long long next_run) int cron_job_list(cron_job_row_t *out, int max_count) { if (!g_db || !out || max_count <= 0) return -1; - const char *sql = "SELECT id, schedule, message, channel, recipient, next_run, enabled FROM cron_jobs ORDER BY next_run ASC"; + const char *sql = "SELECT id, schedule, message, channel, recipient, next_run, enabled " + "FROM cron_jobs ORDER BY next_run ASC, id ASC"; sqlite3_stmt *stmt = NULL; if (sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1; int count = 0; @@ -475,6 +478,27 @@ int cron_job_list(cron_job_row_t *out, int max_count) return count; } +int cron_job_list_due(long long now, cron_due_key_t *out, int max_count) +{ + const char *sql; + sqlite3_stmt *stmt = NULL; + int count = 0; + + if (!g_db || !out || max_count <= 0) return -1; + sql = "SELECT id, next_run FROM cron_jobs WHERE next_run <= ?1 AND enabled = 1 " + "ORDER BY next_run ASC, id ASC"; + if (sqlite3_prepare_v2(g_db, sql, -1, &stmt, NULL) != SQLITE_OK) return -1; + sqlite3_bind_int64(stmt, 1, now); + while (count < max_count && sqlite3_step(stmt) == SQLITE_ROW) { + copy_str_bounded(out[count].id, sizeof(out[count].id), + (const char *)sqlite3_column_text(stmt, 0)); + out[count].next_run = sqlite3_column_int64(stmt, 1); + count++; + } + sqlite3_finalize(stmt); + return count; +} + static int cron_job_select_due(long long now, int use_cursor, long long after_run, const char *after_id, cron_job_row_t *out) { diff --git a/src/core/memory.h b/src/core/memory.h index 99cd0e1..b266660 100644 --- a/src/core/memory.h +++ b/src/core/memory.h @@ -134,7 +134,7 @@ int config_kv_set(const char *key, const char *value); * cron_job_row_free(&row); */ typedef struct cron_job_row { - char id[128]; + char id[128]; /* stored ids must fit with a trailing NUL (127 chars) */ char *schedule; char *message; char channel[64]; @@ -179,9 +179,30 @@ int cron_job_toggle(const char *id); */ int cron_job_update_next_run(const char *id, long long next_run); +/** + * Key of a due cron job. No heap fields. Example: + * cron_due_key_t keys[8]; + * int n = cron_job_list_due(now, keys, 8); + */ +typedef struct cron_due_key { + char id[128]; + long long next_run; +} cron_due_key_t; + +/** + * List enabled jobs with next_run <= now, ordered by next_run then id. + * + * @param out Array to fill (caller-allocated). + * @param max_count Maximum keys to return. + * @return Number of keys written, or -1 on error. + */ +int cron_job_list_due(long long now, cron_due_key_t *out, int max_count); + /** * List cron jobs into output array. * + * Ties on next_run are ordered by id ascending, matching the due poller. + * * @param out Array to fill (caller-allocated). Heap fields are owned * by the caller on success; on -1, no row is owned. * @param max_count Maximum jobs to return. diff --git a/src/tools/cron.c b/src/tools/cron.c index 9074954..c6df97c 100644 --- a/src/tools/cron.c +++ b/src/tools/cron.c @@ -179,10 +179,12 @@ static int cron_init(const config_t *cfg) } #define CRON_OFFER_TRACK 16 +#define CRON_DUE_SCAN_MAX 64 typedef struct cron_offer_slot { char id[128]; struct timespec mono; + int timeout_ms; } cron_offer_slot_t; static cron_offer_slot_t s_offers[CRON_OFFER_TRACK]; @@ -205,6 +207,27 @@ static int cron_offer_age_ms(const char *job_id, const struct timespec *now, lon return 0; } +static int cron_slot_reusable(const cron_offer_slot_t *slot, const struct timespec *now) +{ + long age; + if (slot->id[0] == '\0') + return 1; + if (slot->timeout_ms <= 0) + return 0; + age = cron_elapsed_ms(&slot->mono, now); + return age >= (long)slot->timeout_ms; +} + +static int cron_offer_has_room(const struct timespec *now) +{ + int i; + for (i = 0; i < CRON_OFFER_TRACK; i++) { + if (cron_slot_reusable(&s_offers[i], now)) + return 1; + } + return 0; +} + static int cron_offer_is_hot(const char *job_id, int timeout_ms, const struct timespec *now) { long age = 0; @@ -215,6 +238,17 @@ static int cron_offer_is_hot(const char *job_id, int timeout_ms, const struct ti return age < (long)timeout_ms; } +/** A missing id is deliverable only when a cold slot can remember the offer. */ +static int cron_due_is_returnable(const char *job_id, int timeout_ms, const struct timespec *now) +{ + long age = 0; + if (cron_offer_is_hot(job_id, timeout_ms, now)) + return 0; + if (!cron_offer_age_ms(job_id, now, &age) && !cron_offer_has_room(now)) + return 0; + return 1; +} + static void cron_wait_remaining(const char *job_id, int timeout_ms) { struct timespec now; @@ -233,12 +267,11 @@ static void cron_wait_remaining(const char *job_id, int timeout_ms) nanosleep(&remain, NULL); } -static void cron_mark_offered(const char *job_id) +static void cron_mark_offered(const char *job_id, int timeout_ms) { struct timespec now; int i; int slot = -1; - int oldest = 0; if (!job_id || job_id[0] == '\0') return; if (clock_gettime(CLOCK_MONOTONIC, &now) != 0) @@ -246,50 +279,39 @@ static void cron_mark_offered(const char *job_id) for (i = 0; i < CRON_OFFER_TRACK; i++) { if (strcmp(s_offers[i].id, job_id) == 0) { s_offers[i].mono = now; + s_offers[i].timeout_ms = timeout_ms; return; } - if (slot < 0 && s_offers[i].id[0] == '\0') + if (slot < 0 && cron_slot_reusable(&s_offers[i], &now)) slot = i; - if (cron_elapsed_ms(&s_offers[i].mono, &now) > - cron_elapsed_ms(&s_offers[oldest].mono, &now)) - oldest = i; } if (slot < 0) - slot = oldest; + return; snprintf(s_offers[slot].id, sizeof(s_offers[slot].id), "%s", job_id); s_offers[slot].mono = now; + s_offers[slot].timeout_ms = timeout_ms; } /** Prefer a due job outside its re-offer window so one stuck job cannot hide the rest. */ static int cron_pick_due_row(long long now, int timeout_ms, cron_job_row_t *row) { + cron_due_key_t keys[CRON_DUE_SCAN_MAX]; struct timespec mono; - long long cursor_run = 0; - char cursor_id[128]; - int have_cursor = 0; + int n; + int i; if (clock_gettime(CLOCK_MONOTONIC, &mono) != 0) memset(&mono, 0, sizeof(mono)); - cursor_id[0] = '\0'; - for (;;) { - int rc = have_cursor - ? cron_job_get_next_due_after(now, cursor_run, cursor_id, row) - : cron_job_get_next_due(now, row); - if (rc != 1) - break; - if (have_cursor && (row->next_run < cursor_run || - (row->next_run == cursor_run && strcmp(row->id, cursor_id) <= 0))) { - cron_job_row_free(row); - break; - } - if (!cron_offer_is_hot(row->id, timeout_ms, &mono)) + n = cron_job_list_due(now, keys, CRON_DUE_SCAN_MAX); + if (n <= 0) + return 0; + for (i = 0; i < n; i++) { + if (!cron_due_is_returnable(keys[i].id, timeout_ms, &mono)) + continue; + if (cron_job_get_by_id(keys[i].id, row) == 1) return 1; - cursor_run = row->next_run; - snprintf(cursor_id, sizeof(cursor_id), "%s", row->id); - have_cursor = 1; - cron_job_row_free(row); } - if (cron_job_get_next_due(now, row) != 1) + if (cron_job_get_by_id(keys[0].id, row) != 1) return 0; cron_wait_remaining(row->id, timeout_ms); return 1; @@ -314,7 +336,7 @@ static int cron_poll(channel_incoming_msg_t *out, int timeout_ms) out->text = strdup(row.message ? row.message : ""); out->attachments = NULL; out->attachments_count = 0; - cron_mark_offered(row.id); + cron_mark_offered(row.id, timeout_ms); cron_job_row_free(&row); if (!out->session_id || !out->user_id || !out->text) { channel_incoming_msg_clear(out); @@ -364,6 +386,7 @@ static int cron_send(const char *recipient, const char *text, static void cron_cleanup(void) { + memset(s_offers, 0, sizeof(s_offers)); } static const channel_t cron_channel = { diff --git a/tests/test_cron.c b/tests/test_cron.c index 4589dd5..28c9cf5 100644 --- a/tests/test_cron.c +++ b/tests/test_cron.c @@ -15,6 +15,13 @@ #define ASSERT(c) do { if (!(c)) { fprintf(stderr, "FAIL: %s:%d %s\n", __FILE__, __LINE__, #c); return 1; } } while (0) #define RUN(t) do { int r = (t); if (r) return r; } while (0) +static void cron_test_reset_offers(void) +{ + const channel_t *cron_ch = channel_cron_get(); + if (cron_ch && cron_ch->cleanup) + cron_ch->cleanup(); +} + static int test_interval_next_run(void) { long long now = 1700000000; @@ -205,6 +212,7 @@ static int test_one_shot_detection(void) static int test_due_job_delivers_full_message(void) { + cron_test_reset_offers(); const char *path = "/tmp/shellclaw_test_cron_long_message.db"; remove(path); ASSERT(memory_init(path) == 0); @@ -235,6 +243,7 @@ static int test_due_job_delivers_full_message(void) static int test_long_interval_schedule_roundtrips(void) { + cron_test_reset_offers(); const char *path = "/tmp/shellclaw_test_cron_long_schedule.db"; const channel_t *ch; channel_incoming_msg_t msg; @@ -314,6 +323,7 @@ static int test_cron_ack_delivery_deferred(void) static int test_cron_poll_keeps_job_until_ack(void) { + cron_test_reset_offers(); const channel_t *cron_ch = channel_cron_get(); ASSERT(cron_ch != NULL); const char *path = "/tmp/shellclaw_test_cron_poll.db"; @@ -340,6 +350,7 @@ static int test_cron_poll_keeps_job_until_ack(void) static int test_cron_ack_advances_recurring_past_due_minute(void) { + cron_test_reset_offers(); const char *path = "/tmp/shellclaw_test_cron_ack_advance.db"; const channel_t *cron_ch; channel_incoming_msg_t msg; @@ -404,6 +415,7 @@ static int test_cron_ack_fail_closed_on_parse_error(void) static int test_cron_poll_waits_before_reoffer(void) { + cron_test_reset_offers(); const char *path = "/tmp/shellclaw_test_cron_reoffer.db"; const channel_t *cron_ch; channel_incoming_msg_t msg; @@ -443,6 +455,11 @@ static int test_cron_job_rejects_oversized_text(void) too_big[CRON_JOB_TEXT_MAX + 1] = '\0'; ASSERT(cron_job_create("bigmsg", "interval:60", too_big, "cli", "default", 1, 1) != 0); ASSERT(cron_job_create("bigsched", too_big, "tick", "cli", "default", 1, 1) != 0); + memset(too_big, 'i', 128); + too_big[128] = '\0'; + ASSERT(cron_job_create(too_big, "interval:60", "tick", "cli", "default", 1, 1) != 0); + too_big[127] = '\0'; + ASSERT(cron_job_create(too_big, "interval:60", "tick", "cli", "default", 1, 1) == 0); free(too_big); memory_cleanup(); remove(path); @@ -461,6 +478,7 @@ static int test_cron_poll_offers_sibling_while_earlier_unacked(void) long elapsed_ms; long long now; + cron_test_reset_offers(); remove(path); ASSERT(memory_init(path) == 0); now = (long long)time(NULL); @@ -491,6 +509,88 @@ static int test_cron_poll_offers_sibling_while_earlier_unacked(void) ASSERT(cron_job_get_by_id("job_b", &row) == 1); ASSERT(row.next_run == now - 5); cron_job_row_free(&row); + memset(&msg, 0, sizeof(msg)); + ASSERT(clock_gettime(CLOCK_MONOTONIC, &t0) == 0); + ASSERT(cron_ch->poll(&msg, 2000) == 1); + ASSERT(clock_gettime(CLOCK_MONOTONIC, &t1) == 0); + elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000L + + (t1.tv_nsec - t0.tv_nsec) / 1000000L; + ASSERT(msg.user_id != NULL); + ASSERT(strcmp(msg.user_id, "job_a") == 0); + ASSERT(elapsed_ms >= 400); + channel_incoming_msg_clear(&msg); + memory_cleanup(); + remove(path); + return 0; +} + +/* A full offer table must wait instead of re-offering an evicted id immediately. */ +static int test_cron_poll_waits_when_offer_table_is_full(void) +{ + const char *path = "/tmp/shellclaw_test_cron_offer_full.db"; + const channel_t *cron_ch; + channel_incoming_msg_t msg; + struct timespec t0; + struct timespec t1; + char id[8]; + long elapsed_ms; + long long now; + int i; + + cron_test_reset_offers(); + remove(path); + ASSERT(memory_init(path) == 0); + now = (long long)time(NULL); + for (i = 0; i < 17; i++) { + snprintf(id, sizeof(id), "j%02d", i); + ASSERT(cron_job_create(id, "interval:60", id, "cli", "default", now - (20 - i), 1) == 0); + } + cron_ch = channel_cron_get(); + for (i = 0; i < 16; i++) { + memset(&msg, 0, sizeof(msg)); + ASSERT(clock_gettime(CLOCK_MONOTONIC, &t0) == 0); + ASSERT(cron_ch->poll(&msg, 400) == 1); + ASSERT(clock_gettime(CLOCK_MONOTONIC, &t1) == 0); + elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000L + + (t1.tv_nsec - t0.tv_nsec) / 1000000L; + snprintf(id, sizeof(id), "j%02d", i); + ASSERT(msg.user_id != NULL); + ASSERT(strcmp(msg.user_id, id) == 0); + ASSERT(elapsed_ms < 150); + channel_incoming_msg_clear(&msg); + } + memset(&msg, 0, sizeof(msg)); + ASSERT(clock_gettime(CLOCK_MONOTONIC, &t0) == 0); + ASSERT(cron_ch->poll(&msg, 400) == 1); + ASSERT(clock_gettime(CLOCK_MONOTONIC, &t1) == 0); + elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000L + + (t1.tv_nsec - t0.tv_nsec) / 1000000L; + ASSERT(msg.user_id != NULL); + ASSERT(strcmp(msg.user_id, "j00") == 0); + ASSERT(elapsed_ms >= 200); + channel_incoming_msg_clear(&msg); + memory_cleanup(); + remove(path); + return 0; +} + +static int test_cron_list_orders_ties_by_id(void) +{ + const char *path = "/tmp/shellclaw_test_cron_list_order.db"; + cron_job_row_t rows[4]; + long long now; + + remove(path); + ASSERT(memory_init(path) == 0); + now = (long long)time(NULL); + ASSERT(cron_job_create("b_job", "interval:60", "B", "cli", "default", now, 1) == 0); + ASSERT(cron_job_create("a_job", "interval:60", "A", "cli", "default", now, 1) == 0); + memset(rows, 0, sizeof(rows)); + ASSERT(cron_job_list(rows, 4) == 2); + ASSERT(strcmp(rows[0].id, "a_job") == 0); + ASSERT(strcmp(rows[1].id, "b_job") == 0); + cron_job_row_free(&rows[0]); + cron_job_row_free(&rows[1]); memory_cleanup(); remove(path); return 0; @@ -517,6 +617,8 @@ int main(void) RUN(test_cron_poll_waits_before_reoffer()); RUN(test_cron_job_rejects_oversized_text()); RUN(test_cron_poll_offers_sibling_while_earlier_unacked()); + RUN(test_cron_poll_waits_when_offer_table_is_full()); + RUN(test_cron_list_orders_ties_by_id()); printf("test_cron: all tests passed\n"); return 0; }