fix(cron): deliver other due jobs while one remains unacked

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,15 +478,50 @@ 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)
+int cron_job_list_due(long long now, cron_due_key_t *out, int max_count)
 {
-	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 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)
+{
+	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);
+	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 +533,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;

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.
@@ -192,12 +213,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.
  *

src/tools/cron.c

@@ -178,37 +178,143 @@ 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
+#define CRON_DUE_SCAN_MAX 64
 
-static void cron_wait_if_reoffer(const char *job_id, int timeout_ms)
+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];
+
+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_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;
+	if (timeout_ms <= 0 || !job_id)
+		return 0;
+	if (!cron_offer_age_ms(job_id, now, &age))
+		return 0;
+	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;
 	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);
 }
 
-static void cron_mark_offered(const char *job_id)
+static void cron_mark_offered(const char *job_id, int timeout_ms)
 {
-	if (!job_id)
+	struct timespec now;
+	int i;
+	int slot = -1;
+	if (!job_id || job_id[0] == '\0')
 		return;
-	snprintf(s_offered_id, sizeof(s_offered_id), "%s", job_id);
-	clock_gettime(CLOCK_MONOTONIC, &s_offered_mono);
+	if (clock_gettime(CLOCK_MONOTONIC, &now) != 0)
+		return;
+	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 && cron_slot_reusable(&s_offers[i], &now))
+			slot = i;
+	}
+	if (slot < 0)
+		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;
+	int n;
+	int i;
+
+	if (clock_gettime(CLOCK_MONOTONIC, &mono) != 0)
+		memset(&mono, 0, sizeof(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;
+	}
+	if (cron_job_get_by_id(keys[0].id, 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 +325,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",
@@ -230,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);
@@ -280,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 = {