fix(cypher): add a wall-clock execution deadline to abort runaway queries

An unbounded whole-graph OPTIONAL MATCH or a GROUP BY that produces roughly one
group per node does O(bindings x groups) work in execute_return_agg and can run
for minutes on a large graph before emitting a single row. The 100k row ceiling
never fires (no rows are produced), so query_graph just hangs with no partial
result and no error (#601).

Arm a monotonic wall-clock budget (default 30s) at query entry and check it
(throttled, every 1024 iterations) in the scan, relationship-expansion and
aggregation hot loops. On expiry the query aborts with a clear, actionable error
instead of hanging. A thread-local test hook forces the budget so the guard is
covered by a deterministic reproduce-first test; a companion test confirms the
default budget does not false-positive on a normal query.

Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>

src/cypher/cypher.c

@@ -2650,6 +2650,49 @@ static void rb_add_row(result_builder_t *rb, const char **values) {
  * Prevents accidental multi-GB JSON payloads from unbounded MATCH (n) RETURN n. */
 #define CYPHER_RESULT_CEILING 100000
 
+/* Wall-clock execution deadline (#601). The row ceiling above only fires once
+ * rows exist, but an unbounded `OPTIONAL MATCH` over the full node set (or a
+ * GROUP BY with ~one group per node) does O(bindings x groups) work and can run
+ * for minutes — exhausting RAM/CPU — before a single row is produced, so the
+ * ceiling never trips and the caller just hangs. A monotonic deadline aborts
+ * such runaway queries with a clear, actionable error. Checked (throttled) in
+ * the scan, expansion and aggregation hot loops. */
+#define CYPHER_DEADLINE_BUDGET_MS 30000  /* 30s: generous for legit heavy queries */
+#define CYPHER_DEADLINE_CHECK_MASK 0x3FF /* sample the clock every 1024 iterations */
+
+static _Thread_local uint64_t g_cypher_deadline_ms = 0; /* absolute; 0 = disarmed */
+static _Thread_local bool g_cypher_timed_out = false;
+static _Thread_local int64_t g_cypher_deadline_override_ms = -1; /* test hook; <0 = default */
+
+static void cypher_deadline_arm(void) {
+    g_cypher_timed_out = false;
+    int64_t budget = g_cypher_deadline_override_ms >= 0 ? g_cypher_deadline_override_ms
+                                                        : CYPHER_DEADLINE_BUDGET_MS;
+    g_cypher_deadline_ms = cbm_now_ms() + (uint64_t)budget;
+}
+
+/* True once the query has run past its wall-clock budget. Sticky: after the
+ * first trip every subsequent call returns true, so later loops short-circuit. */
+static bool cypher_deadline_exceeded(void) {
+    if (g_cypher_timed_out) {
+        return true;
+    }
+    if (g_cypher_deadline_ms == 0) {
+        return false;
+    }
+    if (cbm_now_ms() >= g_cypher_deadline_ms) {
+        g_cypher_timed_out = true;
+        return true;
+    }
+    return false;
+}
+
+/* Test-only: force the execution budget (ms) for subsequent queries on this
+ * thread. 0 = trip on the first hot-loop check; <0 restores the default. */
+void cbm_cypher_test_set_deadline_ms(int64_t budget_ms) {
+    g_cypher_deadline_override_ms = budget_ms;
+}
+
 /* ── Binding virtual variables (for WITH clause) ──────────────── */
 
 static const char *binding_get_virtual(binding_t *b, const char *var, const char *prop) {
@@ -3054,6 +3097,11 @@ static void expand_pattern_rels(cbm_store_t *store, cbm_pattern_t *pat, binding_
                                 int *bind_count, const int *bind_cap, const char **var_name,
                                 bool is_optional) {
     for (int ri = 0; ri < pat->rel_count; ri++) {
+        /* #601: stop expanding further hops once the wall-clock budget is spent
+         * (an unbounded expansion is exactly what blows up here). */
+        if (cypher_deadline_exceeded()) {
+            return;
+        }
         cbm_rel_pattern_t *rel = &pat->rels[ri];
         cbm_node_pattern_t *target_node = &pat->nodes[ri + SKIP_ONE];
         const char *to_var = target_node->variable ? target_node->variable : "_n_t";
@@ -3068,6 +3116,9 @@ static void expand_pattern_rels(cbm_store_t *store, cbm_pattern_t *pat, binding_
         int new_count = 0;
 
         for (int bi = 0; bi < *bind_count; bi++) {
+            if ((bi & CYPHER_DEADLINE_CHECK_MASK) == 0 && cypher_deadline_exceeded()) {
+                break;
+            }
             binding_t *b = &(*bindings)[bi];
             cbm_node_t *src = binding_get(b, *var_name);
             if (!src) {
@@ -4111,6 +4162,11 @@ static void execute_return_agg(cbm_return_clause_t *ret, binding_t *bindings, in
     int agg_count = 0;
 
     for (int bi = 0; bi < bind_count; bi++) {
+        /* #601: grouping is O(bindings x groups) — the dominant cost on a
+         * whole-graph GROUP BY. Abort if we blow the wall-clock budget. */
+        if ((bi & CYPHER_DEADLINE_CHECK_MASK) == 0 && cypher_deadline_exceeded()) {
+            break;
+        }
         char key[CBM_SZ_1K] = "";
         const char *vals[CBM_SZ_32];
         char valbufs[CBM_SZ_32][CBM_SZ_512];
@@ -4479,6 +4535,9 @@ static int execute_single(cbm_store_t *store, cbm_query_t *q, const char *projec
     const char *var_name = pat0->nodes[0].variable ? pat0->nodes[0].variable : "_n0";
 
     for (int i = 0; i < scan_count && bind_count < bind_cap; i++) {
+        if ((i & CYPHER_DEADLINE_CHECK_MASK) == 0 && cypher_deadline_exceeded()) {
+            break;
+        }
         binding_t b = {0};
         b.store = store;
         binding_set(&b, var_name, &scanned[i]);
@@ -4527,6 +4586,7 @@ int cbm_cypher_execute(cbm_store_t *store, const char *query, const char *projec
                        cbm_cypher_result_t *out) {
     memset(out, 0, sizeof(*out));
     g_cypher_depth_clamped = 0;
+    cypher_deadline_arm(); /* #601: start the wall-clock budget for this query */
     if (max_rows <= 0) {
         max_rows = CYPHER_RESULT_CEILING;
     }
@@ -4570,6 +4630,18 @@ int cbm_cypher_execute(cbm_store_t *store, const char *query, const char *projec
         rb_apply_distinct(&rb);
     }
 
+    /* #601: abort a runaway query that blew the wall-clock budget before it can
+     * return a misleading partial result. Checked before the row ceiling. */
+    if (g_cypher_timed_out) {
+        rb_free(&rb);
+        cbm_query_free(q);
+        out->error =
+            heap_strdup("query exceeded the execution time limit — narrow the pattern with a WHERE "
+                        "filter, use a directed MATCH instead of an unbounded OPTIONAL MATCH, or "
+                        "add LIMIT");
+        return CBM_NOT_FOUND;
+    }
+
     /* Check ceiling */
     if (rb.row_count >= CYPHER_RESULT_CEILING) {
         rb_free(&rb);

src/cypher/cypher.h

@@ -332,4 +332,9 @@ int cbm_cypher_parse(const char *query, cbm_query_t **out, char **error);
 /* Free a query AST. */
 void cbm_query_free(cbm_query_t *q);
 
+/* Test-only (#601): force the wall-clock execution budget in milliseconds for
+ * subsequent queries on the calling thread. 0 = trip on the first hot-loop
+ * check; a negative value restores the default budget. */
+void cbm_cypher_test_set_deadline_ms(int64_t budget_ms);
+
 #endif /* CBM_CYPHER_H */

tests/test_cypher.c

@@ -2827,6 +2827,54 @@ TEST(cypher_wide_return_projection_bounded) {
 #endif
 }
 
+/* #601: an unbounded whole-graph OPTIONAL MATCH / GROUP BY does
+ * O(bindings x groups) work and can run for minutes with no wall-clock guard —
+ * the 100k row ceiling never fires because no rows are produced, so query_graph
+ * just hangs. With the execution deadline armed to trip immediately (budget 0),
+ * the runaway query must abort with a clear error instead of returning a
+ * (misleading, possibly partial) result.
+ *
+ * RED on unfixed code: no deadline exists, so the query completes and returns
+ * rc==0 with rows and no error — the assertions below fail. */
+TEST(cypher_exec_deadline_aborts_runaway_query_issue601) {
+    cbm_store_t *s = setup_cypher_store();
+    cbm_cypher_result_t r = {0};
+
+    cbm_cypher_test_set_deadline_ms(0); /* trip on the first hot-loop check */
+    int rc = cbm_cypher_execute(
+        s, "MATCH (a) OPTIONAL MATCH (a)-[:CALLS]->(b) RETURN a.qualified_name, count(b)", "test",
+        0, &r);
+    cbm_cypher_test_set_deadline_ms(-1); /* restore default before asserting (thread-local) */
+
+    ASSERT_TRUE(rc != 0); /* CBM_NOT_FOUND (-1) — query aborted, not success */
+    ASSERT_NOT_NULL(r.error);
+    ASSERT_TRUE(strstr(r.error, "time limit") != NULL);
+    ASSERT_EQ(r.row_count, 0);
+
+    cbm_cypher_result_free(&r);
+    cbm_store_close(s);
+    PASS();
+}
+
+/* #601 companion: the default (ample) budget must NOT false-positive on a
+ * normal small query — it still returns its rows. */
+TEST(cypher_exec_deadline_allows_normal_query_issue601) {
+    cbm_store_t *s = setup_cypher_store();
+    cbm_cypher_result_t r = {0};
+
+    int rc = cbm_cypher_execute(
+        s, "MATCH (a) OPTIONAL MATCH (a)-[:CALLS]->(b) RETURN a.qualified_name, count(b)", "test",
+        0, &r);
+
+    ASSERT_EQ(rc, 0);
+    ASSERT_TRUE(r.error == NULL);
+    ASSERT_GT(r.row_count, 0);
+
+    cbm_cypher_result_free(&r);
+    cbm_store_close(s);
+    PASS();
+}
+
 /* ══════════════════════════════════════════════════════════════════ */
 
 SUITE(cypher) {
@@ -2859,6 +2907,8 @@ SUITE(cypher) {
     RUN_TEST(cypher_parse_inline_props);
     RUN_TEST(cypher_parse_error);
     /* Execution */
+    RUN_TEST(cypher_exec_deadline_aborts_runaway_query_issue601);
+    RUN_TEST(cypher_exec_deadline_allows_normal_query_issue601);
     RUN_TEST(cypher_exec_match_all_functions);
     RUN_TEST(cypher_issue240_labels_function);
     RUN_TEST(cypher_issue237_distinct_order_limit);