Govern LFTJ output growth

docs/MEMORY.md

@@ -544,8 +544,10 @@ advisory/unbounded. Explicit `0`, malformed values, overflow, and values below
 256 MiB are invalid. This resolver does not use physical RAM as an enforcing
 fallback. The legacy ledger worker-share hint remains separate from governor
 admission. Fixed eval-arena, delta-pool, compound-arena backing storage, and
-the covered join allocation paths use admission; LFTJ, auxiliary metadata,
-and eval-entry segment allocations remain follow-up work.
+the covered join allocation paths use admission. LFTJ iterator/sort workspace
+and operator staging use governor reservations; remaining LFTJ output/source
+boundaries, auxiliary metadata, and eval-entry segment allocations remain
+follow-up work tracked by #1978.
 
 The ledger's RELATION worker-share hint can still trigger join backpressure
 at 80% (`wl_mem_ledger_should_backpressure(RELATION, 80)`). The governor also

docs/THREADING.md

@@ -687,7 +687,47 @@ the committed token after publication and before a growth transaction.
 | `eval_dedup.c:wl_columnar_eval_dedup_test_fail_next_growth_alloc` | test-only fault flag | `atomic_store_explicit` | release | Arm one allocation refusal before a test invokes dedup growth; excluded from the production library |
 | `eval_dedup.c:wl_columnar_eval_dedup_set_grow` | test-only fault flag | `atomic_exchange_explicit` | acquire-release | Consume the one-shot fault safely when test workers grow dedup tables; excluded from the production library |
 
-The complete source audit now contains **220 atomic call sites**.
+### 5.19 `wirelog/columnar/lftj.c` — governed admission test hook (2 rows)
+
+The test-only admission hook temporarily lowers the governor limit to force
+the inner LFTJ reservation down its denial path. The load and store are
+excluded from the production library.
+
+| Anchor (file:function[#N]) | Field | Op | Order | Justification |
+|---|---|---|---|---|
+| `lftj.c:wl_columnar_lftj_join_typed_governed` | `memory_governor->usable_bytes` | `atomic_load_explicit` | relaxed | Save the configured admission limit before the test hook applies a temporary denial limit |
+| `lftj.c:wl_columnar_lftj_join_typed_governed#2` | `memory_governor->usable_bytes` | `atomic_store_explicit` | relaxed | Apply the test-only denial limit while this thread forces the inner reservation failure; excluded from the production library |
+
+The complete source audit now contains **222 atomic call sites**.
+
+### 5.20 `wirelog/columnar/ops.c` — LFTJ output growth admission test hook (3 rows)
+
+The test-only growth hook saves the configured governor limit, sets
+`usable_bytes` to the already-reserved total to force a growth denial, then
+restores the limit after the append attempt. These operations are excluded
+from the production library.
+
+| Anchor (file:function[#N]) | Field | Op | Order | Justification |
+|---|---|---|---|---|
+| `ops.c:lftj_test_before_output_growth` | `memory_governor->usable_bytes` | `atomic_load_explicit` | relaxed | Save the configured limit before the test hook forces output-growth admission to fail; test-only and excluded from the production library |
+| `ops.c:lftj_test_before_output_growth#2` | `memory_governor->usable_bytes` | `atomic_store_explicit` | relaxed | Temporarily set the limit to the current reserved total so the output-growth reservation is denied; test-only and excluded from the production library |
+| `ops.c:lftj_test_after_output_append` | `memory_governor->usable_bytes` | `atomic_store_explicit` | relaxed | Restore the saved governor limit after the append attempt; test-only and excluded from the production library |
+
+The complete source audit now contains **225 atomic call sites**.
+
+### 5.21 `wirelog/columnar/ops.c` — LFTJ output construction admission test hook (3 rows)
+
+The test-only constructor hook records and temporarily lowers the governor
+limit to exercise denied initial output allocation, then restores the limit.
+These operations are excluded from the production library.
+
+| Anchor (file:function[#N]) | Field | Op | Order | Justification |
+|---|---|---|---|---|
+| `ops.c:lftj_test_before_output_construction` | `memory_governor->usable_bytes` | `atomic_load_explicit` | relaxed | Save the configured limit before the test hook forces initial output-construction admission to fail; test-only and excluded from the production library |
+| `ops.c:lftj_test_before_output_construction#2` | `memory_governor->usable_bytes` | `atomic_store_explicit` | relaxed | Temporarily lower the limit so initial output construction is denied; test-only and excluded from the production library |
+| `ops.c:lftj_test_after_output_construction` | `memory_governor->usable_bytes` | `atomic_store_explicit` | relaxed | Restore the saved limit after the constructor attempt; test-only and excluded from the production library |
+
+The complete source audit now contains **228 atomic call sites**.
 
 ---
 

scripts/ci/check-threading-doc.sh

@@ -19,7 +19,7 @@ fi
 rows="$tmp_dir/rows"
 sed -nE 's/^\| `([^`]+:[A-Za-z_][A-Za-z0-9_]*(#[0-9]+)?)` \| [^|]* \| `([^`]*)` \|.*/\1\t\3/p' "$doc" >"$rows"
 row_count=$(wc -l <"$rows")
-expected_rows="${WIRELOG_THREADING_EXPECTED_ROWS:-220}"
+expected_rows="${WIRELOG_THREADING_EXPECTED_ROWS:-228}"
 [ "$row_count" -eq "$expected_rows" ] || {
     echo "check-threading-doc: FAIL: expected $expected_rows audit rows, found $row_count" >&2
     exit 1

tests/meson.build

@@ -180,7 +180,8 @@ testlib_prod_src = files(
   ) + thread_src
 testlib_prod_c_args = ['-DWL_SESSION_TEST_HOOKS=1', '-DWL_TEST_RELATION_RESIZE_HOOK=1',
                        '-DWL_TEST_JOIN_CACHE_HOOK=1',
-                       '-DWL_TEST_JOIN_BATCH_DESCRIPTOR_HOOKS=1']
+                       '-DWL_TEST_JOIN_BATCH_DESCRIPTOR_HOOKS=1',
+                       '-DWL_TEST_LFTJ_ADMISSION_HOOKS=1']
 
 testlib_prod = static_library(
   'testlib_prod',
@@ -2916,6 +2917,7 @@ test_lftj_exe = executable(
   include_directories: [wirelog_inc, wirelog_src_inc],
   dependencies: [nanoarrow_dep, threads_dep, xxhash_dep, mbedtls_dep, math_dep],
   link_with: [testlib_prod],
+  c_args: ['-DWL_TEST_LFTJ_ADMISSION_HOOKS=1'],
 )
 
 test('lftj', test_lftj_exe)
@@ -3924,6 +3926,7 @@ test_lftj_integration_exe = executable(
   include_directories: [wirelog_inc, wirelog_src_inc],
   dependencies: [nanoarrow_dep, threads_dep, xxhash_dep, mbedtls_dep, math_dep],
   link_with: [testlib_prod],
+  c_args: ['-DWL_TEST_LFTJ_ADMISSION_HOOKS=1'],
 )
 
 test('lftj_integration', test_lftj_integration_exe)

tests/test_lftj.c

@@ -24,13 +24,28 @@
 
 #include "../wirelog/columnar/lftj.h"
 
+#include <errno.h>
 #include <inttypes.h>
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 
+static wl_columnar_memory_governor_ref_t *
+test_governor(uint64_t usable_bytes)
+{
+    wl_columnar_memory_resolution_t resolution = {
+        .budget_bytes = usable_bytes,
+        .headroom_bytes = 0,
+        .usable_bytes = usable_bytes,
+        .mode = WL_COLUMNAR_MEMORY_MODE_ENFORCING,
+        .source = WL_COLUMNAR_MEMORY_SOURCE_ENV,
+        .status = WL_COLUMNAR_MEMORY_OK,
+    };
+    return wl_columnar_memory_governor_ref_create(&resolution);
+}
+
 /* ----------------------------------------------------------------
  * Test framework
  * ---------------------------------------------------------------- */
@@ -801,6 +816,104 @@ test_integer_negative_key_order(void)
     PASS();
 }
 
+static void
+test_governed_denial_and_retry(void)
+{
+    TEST("governed workspace denial returns ENOSPC and permits retry");
+    int64_t left[] = { 1 };
+    int64_t right[] = { 1 };
+    wl_lftj_input_t inputs[2] = {
+        { left, 1, 1, 0 },
+        { right, 1, 1, 0 },
+    };
+    wl_columnar_memory_governor_ref_t *ref = test_governor(1);
+    const char *failure = NULL;
+    int64_t count = 0;
+    int rc;
+
+    if (!ref) {
+        failure = "governor creation failed";
+        goto cleanup;
+    }
+    rc = wl_columnar_lftj_join_typed_governed(inputs, WIRELOG_TYPE_INT64,
+            2, count_cb, &count, ref);
+    if (rc != ENOSPC || count != 0
+        || wl_columnar_memory_reserved(
+            wl_columnar_memory_governor_ref_get(ref)) != 0) {
+        failure = "denial did not return ENOSPC without retaining bytes";
+        goto cleanup;
+    }
+    atomic_store_explicit(
+        &wl_columnar_memory_governor_ref_get(ref)->usable_bytes,
+        UINT64_C(1) << 20, memory_order_relaxed);
+    rc = wl_columnar_lftj_join_typed_governed(inputs, WIRELOG_TYPE_INT64,
+            2, count_cb, &count, ref);
+    if (rc != 0 || count != 1
+        || wl_columnar_memory_reserved(
+            wl_columnar_memory_governor_ref_get(ref)) != 0) {
+        failure = "successful retry did not release workspace";
+        goto cleanup;
+    }
+
+cleanup:
+    if (ref)
+        wl_columnar_memory_governor_ref_release(ref);
+    if (failure)
+        FAIL(failure);
+    PASS();
+}
+
+static void
+test_iters_allocation_failure_rollback(void)
+{
+    TEST("iterator allocation failure rolls back admitted workspace");
+    int64_t left[] = { 1 };
+    int64_t right[] = { 1 };
+    wl_lftj_input_t inputs[2] = {
+        { left, 1, 1, 0 },
+        { right, 1, 1, 0 },
+    };
+    wl_columnar_memory_governor_ref_t *ref = test_governor(UINT64_C(1) << 20);
+    const char *failure = NULL;
+    wl_columnar_lftj_test_hook_state_t hook_state = {0};
+    int64_t count = 0;
+    int rc;
+
+    if (!ref) {
+        failure = "governor creation failed";
+        goto cleanup;
+    }
+    wl_columnar_lftj_test_clear_hooks();
+    wl_columnar_lftj_test_fail_next_iters_alloc();
+    rc = wl_columnar_lftj_join_typed_governed(inputs, WIRELOG_TYPE_INT64,
+            2, count_cb, &count, ref);
+    wl_columnar_lftj_test_get_hook_state(&hook_state);
+    if (rc != ENOMEM || count != 0 || !hook_state.fail_iters_consumed
+        || hook_state.fail_iters_pending
+        || wl_columnar_memory_reserved(
+            wl_columnar_memory_governor_ref_get(ref)) != 0) {
+        failure = "injected failure did not roll back reservation";
+        goto cleanup;
+    }
+    count = 0;
+    rc = wl_columnar_lftj_join_typed_governed(inputs, WIRELOG_TYPE_INT64,
+            2, count_cb, &count, ref);
+    if (rc != 0 || count != 1
+        || wl_columnar_memory_reserved(
+            wl_columnar_memory_governor_ref_get(ref)) != 0) {
+        failure = "retry after allocation failure did not succeed cleanly";
+        goto cleanup;
+    }
+
+cleanup:
+    wl_columnar_lftj_test_clear_hooks();
+    if (ref)
+        wl_columnar_memory_governor_ref_release(ref);
+    if (failure)
+        FAIL(failure);
+    PASS();
+}
+
 /* ================================================================
  * main
  * ================================================================ */
@@ -827,6 +940,8 @@ main(void)
     test_einval_k_too_large();
     test_float_key_semantics();
     test_integer_negative_key_order();
+    test_governed_denial_and_retry();
+    test_iters_allocation_failure_rollback();
 
     printf("\nResults: %d/%d passed", pass_count, test_count);
     if (fail_count > 0)