diff --git a/README.md b/README.md index f3bdffe8ac..843b78f8ba 100644 --- a/README.md +++ b/README.md @@ -1028,12 +1028,16 @@ codebase-memory-mcp config set index_max_files 250000 # optional per-index so codebase-memory-mcp config set index_max_source_mb 16384 # optional per-index source-size limit codebase-memory-mcp config set index_max_rss_mb 8192 # optional worker-tree charged-memory limit codebase-memory-mcp config set index_max_duration_seconds 3600 # optional per-request duration +codebase-memory-mcp config set index_cache_max_mb 32768 # optional projected cache-size limit +codebase-memory-mcp config set index_min_free_disk_mb 4096 # optional free-space reserve codebase-memory-mcp config reset auto_index # reset to default ``` -The four `index_max_*` settings default to `off`. Exceeding one fails the complete -index attempt rather than publishing a partial graph; an existing serving index -is preserved. See [Index resource limits](docs/INDEX_RESOURCE_LIMITS.md). +The six index resource settings default to `off`. Exceeding one fails the +complete index attempt rather than publishing a partial graph; an existing +serving index is preserved. Storage probes also fail closed when an enabled +measurement cannot be completed. See +[Index resource limits](docs/INDEX_RESOURCE_LIMITS.md). ### Environment Variables diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 67c624a602..68d648e513 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -97,6 +97,8 @@ Current keys: | `index_max_source_mb` | `off` | Optional maximum accepted source size in MiB in one discovery run. | | `index_max_rss_mb` | `off` | Optional maximum charged memory in MiB for the complete contained index-worker process tree (`64..1048576`). On macOS this is phys_footprint, not resident size. | | `index_max_duration_seconds` | `off` | Optional maximum wall-clock duration in seconds for the whole index request, including crash/hang recovery (`1..86400`). Like `index_max_rss_mb`, it stops a worker only before it starts to publish the new index; see [INDEX_RESOURCE_LIMITS.md](INDEX_RESOURCE_LIMITS.md). | +| `index_cache_max_mb` | `off` | Optional maximum projected cache size in MiB after publication. | +| `index_min_free_disk_mb` | `off` | Optional minimum free MiB reserved on the cache filesystem while indexing. | | `install_hooks` | `true` | Install CBM-managed client hooks. `install --no-hooks` persists `false`; `install --hooks` restores `true`. | Bare installs and updates respect the stored hook preference. MCP registration, skills, instructions, and profiles remain installed when hooks are disabled. @@ -141,13 +143,14 @@ has not run yet). > `auto_index` still runs, and `index_repository` stays available for manual > reindexing. -The four `index_max_*` settings are independent and disabled by default. They -apply to explicit indexing, automatic indexing, and watcher re-indexing, but not -to `cross-repo-intelligence`, which does not scan repository source files. -Equality is allowed; exceeding any setting fails the complete index request and -preserves any previously serving database. Worker RSS covers descendants and is -not the same as the internal `CBM_MEM_BUDGET_MB` allocation budget. Total -duration is independent of the existing 15-minute no-log-progress timeout. See +The six index resource settings are independent and disabled by default. +They apply to explicit indexing, automatic indexing, and watcher re-indexing, +but not to `cross-repo-intelligence`, which does not scan or publish repository +source indexes. Equality is allowed; exceeding a setting fails the complete +index request and preserves any previously serving database. Worker RSS covers +descendants and is not the same as the internal `CBM_MEM_BUDGET_MB` allocation +budget. Total duration is independent of the existing 15-minute no-log-progress +timeout. See [Index resource limits](INDEX_RESOURCE_LIMITS.md) for counting, validation, and error-response details. diff --git a/docs/INDEX_RESOURCE_LIMITS.md b/docs/INDEX_RESOURCE_LIMITS.md index d76b162d76..3967b6a0a4 100644 --- a/docs/INDEX_RESOURCE_LIMITS.md +++ b/docs/INDEX_RESOURCE_LIMITS.md @@ -119,6 +119,31 @@ request: it is true only when a servable index existed when the request began, no attempt of the request (including an earlier one that crashed and was recovered) claimed publication, and that index still serves. +## Storage settings + +| Key | Default | Accepted value | Protects | +|---|---:|---:|---| +| `index_cache_max_mb` | `off` | `off` or `1..1048576` | Projected cache bytes after replacement | +| `index_min_free_disk_mb` | `off` | `off` or `1..1048576` | Free bytes reserved on the cache filesystem | + +Set or reset these keys through the same `config set` and `config reset` +commands. With both keys `off`, indexing does not scan the cache tree or probe +filesystem capacity. + +Projected cache usage is the current cache size, minus the old project +database and SQLite sidecars only when that generation is confirmed valid and +replaceable, plus the current operation's staging artifacts. Other projects +and unrelated files always count toward the limit and are never evicted. + +Free space is checked before staging, after the staged build completes, and +immediately before atomic publication. Equality is allowed. An enabled +measurement that cannot be completed fails closed with +`code: "resource_probe_failed"` and +`stage: "storage"`. A limit breach uses `code: "resource_limit_exceeded"`. +Both cases preserve the old serving database. Staging files created by a +terminated supervised worker are tagged with a private task token and removed +after its process tree is quiescent; cleanup cannot match another attempt. + ## Trust and compatibility Limits are read from the CLI-managed `_config.db`; they are not MCP request @@ -129,5 +154,5 @@ parent policy. These settings do not replace or increase `auto_index_limit`, change the 512 MiB single-file cap, alter workspace-root authorization, or affect `cross-repo-intelligence`. With all settings `off`, discovery follows the -existing path and the supervisor performs no periodic RSS probe or total-duration -termination. +existing path, the supervisor performs no periodic RSS probe or total-duration +termination, and the pipeline performs no cache-tree or free-space probe. diff --git a/scripts/test-runtime.sh b/scripts/test-runtime.sh index e6e60b15a7..b7e3b885a3 100644 --- a/scripts/test-runtime.sh +++ b/scripts/test-runtime.sh @@ -113,7 +113,8 @@ _cbm_test_runtime_daemon() { # nothing else. cbm_test_index_worker_policy_json() { printf '%s' '"_cbm_index_policy":{"index_max_files":"off","index_max_source_mb":"off"' - printf '%s' ',"index_max_rss_mb":"off","index_max_duration_seconds":"off"}' + printf '%s' ',"index_max_rss_mb":"off","index_max_duration_seconds":"off"' + printf '%s' ',"index_cache_max_mb":"off","index_min_free_disk_mb":"off"}' } cbm_test_runtime_cleanup() { diff --git a/src/cli/cli.c b/src/cli/cli.c index 247c642160..01e61d975f 100644 --- a/src/cli/cli.c +++ b/src/cli/cli.c @@ -8105,6 +8105,9 @@ static const config_key_def_t CONFIG_KEYS[] = { {CBM_INDEX_CONFIG_MAX_RSS_MB, "off", "Max worker process-tree charged memory MiB, or off"}, {CBM_INDEX_CONFIG_MAX_DURATION_SECONDS, "off", "Max wall-clock seconds for the whole index request, or off"}, + {CBM_INDEX_CONFIG_CACHE_MAX_MB, "off", "Max projected cache MiB after publish, or off"}, + {CBM_INDEX_CONFIG_MIN_FREE_DISK_MB, "off", + "Minimum free cache-filesystem MiB during indexing, or off"}, }; /* #1558: ui_enabled and ui_port were reachable ONLY by hand-editing diff --git a/src/daemon/application.c b/src/daemon/application.c index 4c15a3e6ff..9b5382c95a 100644 --- a/src/daemon/application.c +++ b/src/daemon/application.c @@ -388,9 +388,12 @@ static int application_worker_start_default(void *context, const char *args_json (void)context; cbm_index_resource_policy_t resource_policy; char error[CBM_SZ_256] = {0}; + char task_db_path[CBM_SZ_1K]; if (!cbm_mcp_index_policy_from_internal_args(args_json, &resource_policy, error, - sizeof(error))) { - cbm_log_error("daemon.index.policy", "error", error); + sizeof(error)) || + !cbm_mcp_index_task_db_path(args_json, task_db_path, sizeof(task_db_path))) { + cbm_log_error("daemon.index.worker_policy", "error", + error[0] ? error : "could not resolve worker index path"); *worker_out = NULL; return -1; } @@ -403,9 +406,10 @@ static int application_worker_start_default(void *context, const char *args_json const char *root = yyjson_get_str(yyjson_obj_get(yyjson_doc_get_root(args_doc), "repo_path")); worker->project = application_index_project_key(root, args_json); yyjson_doc_free(args_doc); - int result = cbm_index_worker_start_with_policy_log( - args_json, memory_budget_bytes, &resource_policy, false, marker_file, quarantine_file, - application_worker_diagnostic, worker, duration_origin_ms, &worker->handle); + int result = cbm_index_worker_start_with_storage_policy( + args_json, memory_budget_bytes, &resource_policy, task_db_path, false, marker_file, + quarantine_file, application_worker_diagnostic, worker, duration_origin_ms, + &worker->handle); if (result != 0) { application_worker_destroy_default(NULL, worker); *worker_out = NULL; @@ -1559,7 +1563,7 @@ static void application_job_request_facts_begin(cbm_daemon_application_job_t *jo cbm_index_resource_policy_t policy; char error[CBM_SZ_256] = {0}; if (cbm_mcp_index_policy_from_internal_args(job->args_json, &policy, error, sizeof(error)) && - cbm_index_policy_worker_enabled(&policy)) { + (cbm_index_policy_worker_enabled(&policy) || policy.max_task_temp_bytes.enabled)) { cbm_mcp_index_request_facts_begin(job->args_json, &job->request_facts); } } diff --git a/src/daemon/bootstrap.c b/src/daemon/bootstrap.c index 617b9c06fc..12c5da3b3d 100644 --- a/src/daemon/bootstrap.c +++ b/src/daemon/bootstrap.c @@ -107,6 +107,21 @@ static bool bootstrap_worker_budget_valid(const char *text) { return value > 0; } +static bool bootstrap_worker_stage_token_valid(const char *token) { + size_t length = token ? strlen(token) : 0; + if (length < 6U || length >= 64U) { + return false; + } + for (size_t index = 0; index < length; index++) { + unsigned char ch = (unsigned char)token[index]; + if (!((ch >= '0' && ch <= '9') || (ch >= 'A' && ch <= 'Z') || (ch >= 'a' && ch <= 'z') || + ch == '-' || ch == '_')) { + return false; + } + } + return true; +} + /* Keep the bootstrap role boundary exact and fail closed before any client or * worker state is initialized. index_supervisor owns the matching builder and * performs the captured-build comparison after this syntax-only classification. */ @@ -147,6 +162,12 @@ static bool bootstrap_worker_argv_exact(int argc, char *const argv[]) { } next += 2; } + if (next < argc && bootstrap_arg_is(argv[next], "--index-worker-stage-token")) { + if (next + 1 >= argc || !bootstrap_worker_stage_token_valid(argv[next + 1])) { + return false; + } + next += 2; + } return next == argc; } diff --git a/src/foundation/compat_fs.c b/src/foundation/compat_fs.c index 604faa50ff..f1e0b03ab8 100644 --- a/src/foundation/compat_fs.c +++ b/src/foundation/compat_fs.c @@ -163,6 +163,24 @@ int cbm_path_info_utf8(const char *path, cbm_path_info_t *out) { return CBM_PATH_INFO_OK; } +bool cbm_filesystem_free_bytes(const char *path, uint64_t *bytes_out) { + if (!path || !bytes_out) { + return false; + } + wchar_t *wpath = cbm_path_to_wide(path); + if (!wpath) { + return false; + } + ULARGE_INTEGER available; + BOOL ok = GetDiskFreeSpaceExW(wpath, &available, NULL, NULL); + free(wpath); + if (!ok) { + return false; + } + *bytes_out = available.QuadPart; + return true; +} + void cbm_closedir(cbm_dir_t *d) { if (d) { if (d->find_handle != INVALID_HANDLE_VALUE) { @@ -786,6 +804,7 @@ int cbm_exec_no_shell(const char *const *argv) { #include #include #include +#include #include #include @@ -875,6 +894,23 @@ int cbm_path_info_utf8(const char *path, cbm_path_info_t *out) { return CBM_PATH_INFO_OK; } +bool cbm_filesystem_free_bytes(const char *path, uint64_t *bytes_out) { + if (!path || !bytes_out) { + return false; + } + struct statvfs status; + if (statvfs(path, &status) != 0) { + return false; + } + uint64_t blocks = (uint64_t)status.f_bavail; + uint64_t fragment = status.f_frsize != 0 ? (uint64_t)status.f_frsize : (uint64_t)status.f_bsize; + if (fragment == 0) { + return false; + } + *bytes_out = blocks > UINT64_MAX / fragment ? UINT64_MAX : blocks * fragment; + return true; +} + void cbm_closedir(cbm_dir_t *d) { if (d) { if (d->dir) { @@ -1258,6 +1294,281 @@ int cbm_exec_no_shell(const char *const *argv) { #endif /* _WIN32 */ +static uint64_t fs_add_saturated(uint64_t left, uint64_t right) { + return left > UINT64_MAX - right ? UINT64_MAX : left + right; +} + +static bool directory_size_recursive(const char *path, unsigned int depth, uint64_t *total) { + enum { CBM_DIRECTORY_SCAN_MAX_DEPTH = 1024 }; + if (!path || !total || depth > CBM_DIRECTORY_SCAN_MAX_DEPTH) { + return false; + } + cbm_dir_t *directory = cbm_opendir(path); + if (!directory) { + return false; + } + bool ok = true; + cbm_dirent_t *entry; + while (ok && (entry = cbm_readdir(directory)) != NULL) { + size_t path_len = strlen(path); + size_t name_len = strlen(entry->name); + bool separator_needed = + path_len > 0 && path[path_len - 1] != '/' && path[path_len - 1] != '\\'; + if (path_len > SIZE_MAX - name_len - (separator_needed ? 2U : 1U)) { + ok = false; + break; + } + size_t child_size = path_len + name_len + (separator_needed ? 2U : 1U); + char *child = (char *)malloc(child_size); + if (!child) { + ok = false; + break; + } + int written = + snprintf(child, child_size, separator_needed ? "%s/%s" : "%s%s", path, entry->name); + cbm_path_info_t info; + int status = written > 0 && (size_t)written < child_size ? cbm_path_info_utf8(child, &info) + : CBM_PATH_INFO_UNAVAILABLE; + if (status == CBM_PATH_INFO_UNAVAILABLE) { + ok = false; + } else if (status == CBM_PATH_INFO_OK && !info.is_symlink && info.is_regular) { + if (info.size < 0) { + ok = false; + } else { + *total = fs_add_saturated(*total, (uint64_t)info.size); + } + } else if (status == CBM_PATH_INFO_OK && !info.is_symlink && info.is_directory) { + ok = directory_size_recursive(child, depth + 1U, total); + } + free(child); + } + cbm_closedir(directory); + return ok; +} + +bool cbm_directory_size_bytes(const char *path, uint64_t *bytes_out) { + if (!path || !bytes_out) { + return false; + } + *bytes_out = 0; + return directory_size_recursive(path, 0, bytes_out); +} + +bool cbm_db_artifact_bytes(const char *db_path, uint64_t *bytes_out) { + if (!db_path || !bytes_out) { + return false; + } + *bytes_out = 0; + static const char *const suffixes[] = {"", "-wal", "-shm", "-journal"}; + size_t base_len = strlen(db_path); + for (size_t index = 0; index < sizeof(suffixes) / sizeof(suffixes[0]); index++) { + size_t suffix_len = strlen(suffixes[index]); + if (base_len > SIZE_MAX - suffix_len - 1U) { + return false; + } + size_t path_size = base_len + suffix_len + 1U; + char *path = (char *)malloc(path_size); + if (!path) { + return false; + } + (void)snprintf(path, path_size, "%s%s", db_path, suffixes[index]); + cbm_path_info_t info; + int status = cbm_path_info_utf8(path, &info); + free(path); + if (status == CBM_PATH_INFO_ABSENT) { + continue; + } + if (status != CBM_PATH_INFO_OK || !info.is_regular || info.is_symlink || info.size < 0) { + return false; + } + *bytes_out = fs_add_saturated(*bytes_out, (uint64_t)info.size); + } + return true; +} + +static bool fs_optional_regular_bytes(const char *path, uint64_t *total) { + if (!path || !path[0]) { + return true; + } + cbm_path_info_t info; + int status = cbm_path_info_utf8(path, &info); + if (status == CBM_PATH_INFO_ABSENT) { + return true; + } + if (status != CBM_PATH_INFO_OK || !info.is_regular || info.is_symlink || info.size < 0) { + return false; + } + *total = fs_add_saturated(*total, (uint64_t)info.size); + return true; +} + +bool cbm_index_task_temp_bytes(const char *final_db_path, const char *stage_token, + const char *log_path, const char *response_path, + uint64_t *bytes_out) { + if (!final_db_path || !final_db_path[0] || !stage_token || !stage_token[0] || !bytes_out) { + return false; + } + *bytes_out = 0; + if (!fs_optional_regular_bytes(log_path, bytes_out) || + !fs_optional_regular_bytes(response_path, bytes_out)) { + return false; + } + const char *basename = strrchr(final_db_path, '/'); +#ifdef _WIN32 + const char *backslash = strrchr(final_db_path, '\\'); + if (backslash && (!basename || backslash > basename)) { + basename = backslash; + } +#endif + basename = basename ? basename + 1 : final_db_path; + size_t directory_len = (size_t)(basename - final_db_path); + char *directory = NULL; + if (directory_len == 0) { + directory = strdup("."); + directory_len = 1; + } else { + directory = (char *)malloc(directory_len + 1U); + if (directory) { + memcpy(directory, final_db_path, directory_len); + while (directory_len > 1U && (directory[directory_len - 1U] == '/' || + directory[directory_len - 1U] == '\\')) { + directory_len--; + } + directory[directory_len] = '\0'; + } + } + size_t basename_len = strlen(basename); + size_t token_len = strlen(stage_token); + static const char stage_marker[] = ".stage."; + if (!directory || token_len > SIZE_MAX - sizeof(stage_marker) - 1U || + basename_len > SIZE_MAX - token_len - sizeof(stage_marker) - 1U) { + free(directory); + return false; + } + size_t prefix_size = basename_len + (sizeof(stage_marker) - 1U) + token_len + 2U; + char *prefix = (char *)malloc(prefix_size); + if (!prefix) { + free(directory); + return false; + } + (void)snprintf(prefix, prefix_size, "%s%s%s.", basename, stage_marker, stage_token); + cbm_dir_t *dir = cbm_opendir(directory); + if (!dir) { + free(prefix); + free(directory); + return false; + } + bool ok = true; + cbm_dirent_t *entry; + while (ok && (entry = cbm_readdir(dir)) != NULL) { + if (strncmp(entry->name, prefix, prefix_size - 1U) != 0) { + continue; + } + size_t name_len = strlen(entry->name); + if (directory_len > SIZE_MAX - name_len - 2U) { + ok = false; + break; + } + size_t path_size = directory_len + name_len + 2U; + char *path = (char *)malloc(path_size); + if (!path) { + ok = false; + break; + } + (void)snprintf(path, path_size, "%s/%s", directory, entry->name); + ok = fs_optional_regular_bytes(path, bytes_out); + free(path); + } + cbm_closedir(dir); + free(prefix); + free(directory); + return ok; +} + +bool cbm_index_staging_cleanup(const char *final_db_path, const char *stage_token) { + if (!final_db_path || !final_db_path[0] || !stage_token || !stage_token[0]) { + return false; + } + const char *basename = strrchr(final_db_path, '/'); +#ifdef _WIN32 + const char *backslash = strrchr(final_db_path, '\\'); + if (backslash && (!basename || backslash > basename)) { + basename = backslash; + } +#endif + basename = basename ? basename + 1 : final_db_path; + size_t directory_len = (size_t)(basename - final_db_path); + char *directory = NULL; + if (directory_len == 0) { + directory = strdup("."); + directory_len = 1; + } else { + directory = (char *)malloc(directory_len + 1U); + if (directory) { + memcpy(directory, final_db_path, directory_len); + while (directory_len > 1U && (directory[directory_len - 1U] == '/' || + directory[directory_len - 1U] == '\\')) { + directory_len--; + } + directory[directory_len] = '\0'; + } + } + static const char stage_marker[] = ".stage."; + size_t basename_len = strlen(basename); + size_t token_len = strlen(stage_token); + if (!directory || token_len > SIZE_MAX - sizeof(stage_marker) - 1U || + basename_len > SIZE_MAX - token_len - sizeof(stage_marker) - 1U) { + free(directory); + return false; + } + size_t prefix_size = basename_len + (sizeof(stage_marker) - 1U) + token_len + 2U; + char *prefix = (char *)malloc(prefix_size); + if (!prefix) { + free(directory); + return false; + } + (void)snprintf(prefix, prefix_size, "%s%s%s.", basename, stage_marker, stage_token); + cbm_dir_t *dir = cbm_opendir(directory); + if (!dir) { + free(prefix); + free(directory); + return false; + } + bool ok = true; + cbm_dirent_t *entry; + while ((entry = cbm_readdir(dir)) != NULL) { + if (strncmp(entry->name, prefix, prefix_size - 1U) != 0) { + continue; + } + size_t name_len = strlen(entry->name); + if (directory_len > SIZE_MAX - name_len - 2U) { + ok = false; + continue; + } + size_t path_size = directory_len + name_len + 2U; + char *path = (char *)malloc(path_size); + if (!path) { + ok = false; + continue; + } + (void)snprintf(path, path_size, "%s/%s", directory, entry->name); + cbm_path_info_t info; + int status = cbm_path_info_utf8(path, &info); + if (status == CBM_PATH_INFO_OK && !info.is_symlink && info.is_regular) { + if (cbm_unlink(path) != 0 && errno != ENOENT) { + ok = false; + } + } else if (status != CBM_PATH_INFO_ABSENT) { + ok = false; + } + free(path); + } + cbm_closedir(dir); + free(prefix); + free(directory); + return ok; +} + /* Canonicalize an EXISTING path (collapse `..`, resolve links/junctions): * realpath on POSIX; a final path queried from an opened handle on Windows. * The previous Windows callers used the ANSI CRT (_access/_fullpath) on UTF-8 diff --git a/src/foundation/compat_fs.h b/src/foundation/compat_fs.h index 1a572a4785..8264c795c8 100644 --- a/src/foundation/compat_fs.h +++ b/src/foundation/compat_fs.h @@ -47,6 +47,25 @@ enum { * their behavior. */ int cbm_path_info_utf8(const char *path, cbm_path_info_t *out); +/* Measure logical regular-file bytes beneath path without following symlinks + * or Windows reparse points. Saturates at UINT64_MAX. */ +bool cbm_directory_size_bytes(const char *path, uint64_t *bytes_out); + +/* Return bytes available to the current process on the filesystem containing + * an existing path (statvfs / GetDiskFreeSpaceExW). */ +bool cbm_filesystem_free_bytes(const char *path, uint64_t *bytes_out); + +/* Sum a SQLite main file and its -wal/-shm/-journal sidecars when present. + * Missing artifacts contribute zero; inspection errors fail the measurement. */ +bool cbm_db_artifact_bytes(const char *db_path, uint64_t *bytes_out); + +/* Sum this indexing attempt's explicit worker log/response plus every regular + * sibling whose name starts with ".stage.". */ +bool cbm_index_task_temp_bytes(const char *final_db_path, const char *stage_token, + const char *log_path, const char *response_path, + uint64_t *bytes_out); +bool cbm_index_staging_cleanup(const char *final_db_path, const char *stage_token); + /* Open a directory for iteration. Returns NULL on error. */ cbm_dir_t *cbm_opendir(const char *path); diff --git a/src/foundation/index_policy.c b/src/foundation/index_policy.c index 80aa2f8a65..9c0f22e9c0 100644 --- a/src/foundation/index_policy.c +++ b/src/foundation/index_policy.c @@ -21,6 +21,10 @@ static const index_policy_metadata_t INDEX_POLICY_METADATA[] = { CBM_INDEX_MIN_RSS_MB_VALUE, CBM_INDEX_MAX_RSS_MB_VALUE, CBM_INDEX_MIB_BYTES}, {CBM_INDEX_CONFIG_MAX_DURATION_SECONDS, offsetof(cbm_index_resource_policy_t, max_duration_ms), 1, CBM_INDEX_MAX_DURATION_SECONDS_VALUE, UINT64_C(1000)}, + {CBM_INDEX_CONFIG_CACHE_MAX_MB, offsetof(cbm_index_resource_policy_t, max_cache_bytes), 1, + CBM_INDEX_MAX_STORAGE_MB_VALUE, CBM_INDEX_MIB_BYTES}, + {CBM_INDEX_CONFIG_MIN_FREE_DISK_MB, offsetof(cbm_index_resource_policy_t, min_free_disk_bytes), + 1, CBM_INDEX_MAX_STORAGE_MB_VALUE, CBM_INDEX_MIB_BYTES}, }; static void set_error(char *error, size_t error_size, const char *key, uint64_t minimum, @@ -61,18 +65,8 @@ void cbm_index_policy_init(cbm_index_resource_policy_t *policy) { } bool cbm_index_policy_enabled(const cbm_index_resource_policy_t *policy) { - if (!policy) { - return false; - } - for (size_t index = 0; index < cbm_index_policy_key_count(); index++) { - const cbm_index_limit_u64_t *limit = - (const cbm_index_limit_u64_t *)((const unsigned char *)policy + - INDEX_POLICY_METADATA[index].field_offset); - if (limit->enabled) { - return true; - } - } - return false; + return cbm_index_policy_discovery_enabled(policy) || cbm_index_policy_worker_enabled(policy) || + cbm_index_policy_storage_enabled(policy); } bool cbm_index_policy_discovery_enabled(const cbm_index_resource_policy_t *policy) { @@ -83,6 +77,62 @@ bool cbm_index_policy_worker_enabled(const cbm_index_resource_policy_t *policy) return policy && (policy->max_rss_bytes.enabled || policy->max_duration_ms.enabled); } +bool cbm_index_policy_storage_enabled(const cbm_index_resource_policy_t *policy) { + return policy && (policy->max_cache_bytes.enabled || policy->min_free_disk_bytes.enabled || + policy->max_final_db_bytes.enabled || policy->max_staging_bytes.enabled || + policy->max_task_temp_bytes.enabled); +} + +static uint64_t add_saturated(uint64_t left, uint64_t right) { + return left > UINT64_MAX - right ? UINT64_MAX : left + right; +} + +static bool storage_within_max(cbm_index_resource_t resource, uint64_t observed, + const cbm_index_limit_u64_t *limit, + cbm_index_resource_violation_t *violation) { + if (!limit->enabled || observed <= limit->value) { + return true; + } + *violation = (cbm_index_resource_violation_t){ + .resource = resource, + .observed = observed, + .limit = limit->value, + }; + return false; +} + +bool cbm_index_policy_check_storage(const cbm_index_resource_policy_t *policy, + const cbm_index_storage_sample_t *sample, + cbm_index_resource_violation_t *violation) { + if (!policy || !sample || !violation) { + return false; + } + *violation = (cbm_index_resource_violation_t){0}; + uint64_t remaining_cache = sample->current_cache_bytes > sample->replaceable_old_bytes + ? sample->current_cache_bytes - sample->replaceable_old_bytes + : 0; + uint64_t projected_cache = add_saturated(remaining_cache, sample->operation_bytes); + if (!storage_within_max(CBM_INDEX_RESOURCE_CACHE_BYTES, projected_cache, + &policy->max_cache_bytes, violation)) { + return false; + } + if (policy->min_free_disk_bytes.enabled && + sample->free_disk_bytes < policy->min_free_disk_bytes.value) { + *violation = (cbm_index_resource_violation_t){ + .resource = CBM_INDEX_RESOURCE_FREE_DISK_BYTES, + .observed = sample->free_disk_bytes, + .limit = policy->min_free_disk_bytes.value, + }; + return false; + } + return storage_within_max(CBM_INDEX_RESOURCE_FINAL_DB_BYTES, sample->final_db_bytes, + &policy->max_final_db_bytes, violation) && + storage_within_max(CBM_INDEX_RESOURCE_STAGING_BYTES, sample->staging_bytes, + &policy->max_staging_bytes, violation) && + storage_within_max(CBM_INDEX_RESOURCE_TASK_TEMP_BYTES, sample->task_temp_bytes, + &policy->max_task_temp_bytes, violation); +} + size_t cbm_index_policy_key_count(void) { return sizeof(INDEX_POLICY_METADATA) / sizeof(INDEX_POLICY_METADATA[0]); } @@ -172,6 +222,16 @@ const char *cbm_index_resource_name(cbm_index_resource_t resource) { return "rss_bytes"; case CBM_INDEX_RESOURCE_DURATION_MS: return "duration_ms"; + case CBM_INDEX_RESOURCE_CACHE_BYTES: + return "cache_bytes"; + case CBM_INDEX_RESOURCE_FREE_DISK_BYTES: + return "free_disk_bytes"; + case CBM_INDEX_RESOURCE_FINAL_DB_BYTES: + return "final_db_bytes"; + case CBM_INDEX_RESOURCE_STAGING_BYTES: + return "staging_bytes"; + case CBM_INDEX_RESOURCE_TASK_TEMP_BYTES: + return "task_temp_bytes"; case CBM_INDEX_RESOURCE_NONE: default: return "unknown"; @@ -198,6 +258,14 @@ const char *cbm_index_resource_config_key(cbm_index_resource_t resource) { return CBM_INDEX_CONFIG_MAX_RSS_MB; case CBM_INDEX_RESOURCE_DURATION_MS: return CBM_INDEX_CONFIG_MAX_DURATION_SECONDS; + case CBM_INDEX_RESOURCE_CACHE_BYTES: + return CBM_INDEX_CONFIG_CACHE_MAX_MB; + case CBM_INDEX_RESOURCE_FREE_DISK_BYTES: + return CBM_INDEX_CONFIG_MIN_FREE_DISK_MB; + case CBM_INDEX_RESOURCE_FINAL_DB_BYTES: + case CBM_INDEX_RESOURCE_STAGING_BYTES: + case CBM_INDEX_RESOURCE_TASK_TEMP_BYTES: + return "index_resource_profile"; case CBM_INDEX_RESOURCE_NONE: default: return "index_resource_limit"; diff --git a/src/foundation/index_policy.h b/src/foundation/index_policy.h index 617a6ab7de..bac059f2e4 100644 --- a/src/foundation/index_policy.h +++ b/src/foundation/index_policy.h @@ -9,12 +9,15 @@ #define CBM_INDEX_CONFIG_MAX_SOURCE_MB "index_max_source_mb" #define CBM_INDEX_CONFIG_MAX_RSS_MB "index_max_rss_mb" #define CBM_INDEX_CONFIG_MAX_DURATION_SECONDS "index_max_duration_seconds" +#define CBM_INDEX_CONFIG_CACHE_MAX_MB "index_cache_max_mb" +#define CBM_INDEX_CONFIG_MIN_FREE_DISK_MB "index_min_free_disk_mb" #define CBM_INDEX_MAX_FILES_VALUE UINT64_C(10000000) #define CBM_INDEX_MAX_SOURCE_MB_VALUE UINT64_C(1048576) #define CBM_INDEX_MIN_RSS_MB_VALUE UINT64_C(64) #define CBM_INDEX_MAX_RSS_MB_VALUE UINT64_C(1048576) #define CBM_INDEX_MAX_DURATION_SECONDS_VALUE UINT64_C(86400) +#define CBM_INDEX_MAX_STORAGE_MB_VALUE UINT64_C(1048576) #define CBM_INDEX_MIB_BYTES UINT64_C(1048576) typedef struct { @@ -27,6 +30,15 @@ typedef struct { cbm_index_limit_u64_t max_source_bytes; cbm_index_limit_u64_t max_rss_bytes; cbm_index_limit_u64_t max_duration_ms; + cbm_index_limit_u64_t max_cache_bytes; + cbm_index_limit_u64_t min_free_disk_bytes; + /* Internal storage dimensions with no public config key. Nothing enables + * them on its own; they exist so that a single composed decision can bound + * the database, the staging artifacts and the task temporary directory + * together, which no individual key can express. */ + cbm_index_limit_u64_t max_final_db_bytes; + cbm_index_limit_u64_t max_staging_bytes; + cbm_index_limit_u64_t max_task_temp_bytes; } cbm_index_resource_policy_t; typedef enum { @@ -35,18 +47,38 @@ typedef enum { CBM_INDEX_RESOURCE_SOURCE_BYTES, CBM_INDEX_RESOURCE_RSS_BYTES, CBM_INDEX_RESOURCE_DURATION_MS, + CBM_INDEX_RESOURCE_CACHE_BYTES, + CBM_INDEX_RESOURCE_FREE_DISK_BYTES, + CBM_INDEX_RESOURCE_FINAL_DB_BYTES, + CBM_INDEX_RESOURCE_STAGING_BYTES, + CBM_INDEX_RESOURCE_TASK_TEMP_BYTES, } cbm_index_resource_t; typedef struct { cbm_index_resource_t resource; uint64_t observed; uint64_t limit; + bool probe_failed; } cbm_index_resource_violation_t; +typedef struct { + uint64_t current_cache_bytes; + uint64_t replaceable_old_bytes; + uint64_t operation_bytes; + uint64_t free_disk_bytes; + uint64_t final_db_bytes; + uint64_t staging_bytes; + uint64_t task_temp_bytes; +} cbm_index_storage_sample_t; + void cbm_index_policy_init(cbm_index_resource_policy_t *policy); bool cbm_index_policy_enabled(const cbm_index_resource_policy_t *policy); bool cbm_index_policy_discovery_enabled(const cbm_index_resource_policy_t *policy); bool cbm_index_policy_worker_enabled(const cbm_index_resource_policy_t *policy); +bool cbm_index_policy_storage_enabled(const cbm_index_resource_policy_t *policy); +bool cbm_index_policy_check_storage(const cbm_index_resource_policy_t *policy, + const cbm_index_storage_sample_t *sample, + cbm_index_resource_violation_t *violation); size_t cbm_index_policy_key_count(void); const char *cbm_index_policy_key_at(size_t index); diff --git a/src/main.c b/src/main.c index 3a7eed240b..6308d45631 100644 --- a/src/main.c +++ b/src/main.c @@ -899,6 +899,8 @@ static int run_cli(int argc, char **argv, cbm_project_lock_manager_t *project_lo cli_strip_flag_value(&argc, argv, CBM_INDEX_WORKER_QUARANTINE_ARG); const char *worker_publish_fence = cli_strip_flag_value(&argc, argv, CBM_INDEX_WORKER_PUBLISH_FENCE_ARG); + const char *worker_stage_token = + cli_strip_flag_value(&argc, argv, CBM_INDEX_WORKER_STAGE_TOKEN_ARG); if (index_worker) { /* The graph lives on this process's heaps: SQLite gets heaps of its * own here, and only here (cbm_sqlite_dedicated_heap). */ @@ -913,7 +915,7 @@ static int run_cli(int argc, char **argv, cbm_project_lock_manager_t *project_lo } cbm_index_set_worker_role_options(index_worker, response_out, worker_single_thread, worker_marker, worker_quarantine, - cbm_index_worker_memory_budget_bytes()); + cbm_index_worker_memory_budget_bytes(), worker_stage_token); cbm_index_set_worker_publish_fence(index_worker ? worker_publish_fence : NULL); if (argc < MAIN_MIN_ARGC) { @@ -3217,7 +3219,7 @@ int main(int argc, char **argv) { } cbm_index_set_worker_role_options(true, invocation.response_out, invocation.single_thread, invocation.marker_file, invocation.quarantine_file, - invocation.memory_budget_bytes); + invocation.memory_budget_bytes, invocation.stage_token); cbm_index_set_worker_publish_fence(invocation.publish_fence_file); #ifndef _WIN32 /* Split into three ordered steps rather than one condition, because the diff --git a/src/mcp/index_supervisor.c b/src/mcp/index_supervisor.c index 701ac57436..40fc624495 100644 --- a/src/mcp/index_supervisor.c +++ b/src/mcp/index_supervisor.c @@ -69,7 +69,7 @@ static void worker_set_local_env(const char *name, const char *value) { void cbm_index_set_worker_role_options(bool is_worker, const char *response_out, bool single_thread, const char *marker_file, const char *quarantine_file, - size_t memory_budget_bytes) { + size_t memory_budget_bytes, const char *stage_token) { cbm_index_set_worker_role(is_worker, response_out); g_worker_memory_budget_bytes = is_worker ? memory_budget_bytes : 0; if (!is_worker) { @@ -78,6 +78,7 @@ void cbm_index_set_worker_role_options(bool is_worker, const char *response_out, worker_set_local_env("CBM_INDEX_SINGLE_THREAD", single_thread ? "1" : NULL); worker_set_local_env("CBM_INDEX_MARKER_FILE", marker_file); worker_set_local_env("CBM_INDEX_QUARANTINE_FILE", quarantine_file); + worker_set_local_env("CBM_INDEX_STAGE_TOKEN", stage_token); } bool cbm_index_worker_active(void) { @@ -230,6 +231,21 @@ static bool worker_fingerprint_valid(const char *fingerprint) { return true; } +static bool worker_stage_token_valid(const char *token) { + size_t length = token ? strlen(token) : 0; + if (length < 6U || length >= CBM_SZ_64) { + return false; + } + for (size_t index = 0; index < length; index++) { + unsigned char ch = (unsigned char)token[index]; + if (!((ch >= '0' && ch <= '9') || (ch >= 'A' && ch <= 'Z') || (ch >= 'a' && ch <= 'z') || + ch == '-' || ch == '_')) { + return false; + } + } + return true; +} + static bool worker_parse_positive_size(const char *text, size_t *value_out) { if (!text || !text[0] || !value_out) { return false; @@ -314,6 +330,13 @@ cbm_index_worker_argv_status_t cbm_index_worker_parse_process_argv( parsed.publish_fence_file = argv[next + 1]; next += 2; } + if (next < argc && argv[next] && strcmp(argv[next], CBM_INDEX_WORKER_STAGE_TOKEN_ARG) == 0) { + if (next + 1 >= argc || !worker_stage_token_valid(argv[next + 1])) { + return CBM_INDEX_WORKER_ARGV_INVALID; + } + parsed.stage_token = argv[next + 1]; + next += 2; + } if (next != argc) { return CBM_INDEX_WORKER_ARGV_INVALID; } @@ -541,12 +564,13 @@ static void worker_log_last_phase(const char *log_path, char *out, size_t out_si enum { INDEX_WORKER_PATH_CAP = CBM_SZ_4K, - INDEX_WORKER_ARGV_CAP = 19, + INDEX_WORKER_ARGV_CAP = 21, INDEX_WORKER_SYNC_POLL_NS = 10000000, INDEX_WORKER_RELAY_LINES_PER_POLL = 64, INDEX_WORKER_RELAY_BYTES_PER_POLL = 64 * 1024, INDEX_WORKER_RSS_PROBE_FAILURE_LIMIT = 3, INDEX_WORKER_RSS_PROBE_INTERVAL_MS = 250, + INDEX_WORKER_TASK_TEMP_PROBE_INTERVAL_MS = 250, }; typedef enum { @@ -568,6 +592,8 @@ typedef enum { static cbm_index_supervisor_clock_fn g_resource_clock; static cbm_index_supervisor_rss_fn g_resource_rss; static void *g_resource_hook_context; +static cbm_index_supervisor_task_temp_fn g_task_temp_hook; +static void *g_task_temp_hook_context; void cbm_index_supervisor_set_resource_hooks_for_testing(cbm_index_supervisor_clock_fn clock_fn, cbm_index_supervisor_rss_fn rss_fn, @@ -582,6 +608,17 @@ void cbm_index_supervisor_reset_resource_hooks_for_testing(void) { g_resource_rss = NULL; g_resource_hook_context = NULL; } + +void cbm_index_supervisor_set_task_temp_hook_for_testing( + cbm_index_supervisor_task_temp_fn task_temp_fn, void *context) { + g_task_temp_hook = task_temp_fn; + g_task_temp_hook_context = context; +} + +void cbm_index_supervisor_reset_task_temp_hook_for_testing(void) { + g_task_temp_hook = NULL; + g_task_temp_hook_context = NULL; +} #endif static uint64_t worker_resource_now_ms(void) { @@ -613,10 +650,14 @@ struct cbm_index_worker_handle { bool process_terminal; cbm_proc_result_t process_result; cbm_index_resource_policy_t resource_policy; + char task_db_path[INDEX_WORKER_PATH_CAP]; + char stage_token[CBM_SZ_64]; uint64_t started_ms; uint64_t last_rss_probe_ms; + uint64_t last_task_temp_probe_ms; unsigned int rss_probe_failures; bool rss_probe_started; + bool task_temp_probe_started; /* Publication fence: a never-created path next to a reserved unique file * (both removed at terminal). Empty when no worker limit is enabled. */ char publish_fence_reserve[INDEX_WORKER_PATH_CAP]; @@ -945,7 +986,7 @@ static void worker_request_resource_termination(cbm_index_worker_handle_t *handl return; } handle->result.resource_violation = (cbm_index_resource_violation_t){ - .resource = resource, .observed = observed, .limit = limit}; + .resource = resource, .observed = observed, .limit = limit, .probe_failed = probe_failed}; handle->result.resource_probe_failed = probe_failed; if (!cbm_subprocess_request_cancel(handle->process)) { handle->result.resource_violation = (cbm_index_resource_violation_t){0}; @@ -1050,9 +1091,18 @@ static const char *worker_artifacts_create(cbm_index_worker_handle_t *handle) { if (!worker_unique_file(handle->log_path, sizeof(handle->log_path), "log")) { return "log"; } - /* Opt-in: only an enabled worker limit can stop a worker, so only then is - * there a publication to coordinate. */ - if (cbm_index_policy_worker_enabled(&handle->resource_policy) && + const char *random_suffix = strrchr(handle->response_path, '-'); + int token_written = random_suffix ? snprintf(handle->stage_token, sizeof(handle->stage_token), + "%ld%s", (long)worker_getpid(), random_suffix) + : -1; + if (token_written <= 0 || token_written >= (int)sizeof(handle->stage_token) || + !worker_stage_token_valid(handle->stage_token)) { + return "stage_token"; + } + /* Opt-in: only an enabled worker or task-temp limit can stop a worker, so + * only then is there a publication to coordinate. */ + if ((cbm_index_policy_worker_enabled(&handle->resource_policy) || + handle->resource_policy.max_task_temp_bytes.enabled) && !worker_publish_fence_create(handle)) { return "publish_fence"; } @@ -1082,11 +1132,49 @@ static bool worker_publication_committed_at_terminal(cbm_index_worker_handle_t * return true; } +static bool worker_task_temp_bytes(cbm_index_worker_handle_t *handle, uint64_t *bytes_out) { +#ifdef CBM_ENABLE_TEST_SEAMS + if (g_task_temp_hook) { + return g_task_temp_hook(handle->task_db_path, handle->log_path, handle->response_path, + bytes_out, g_task_temp_hook_context); + } +#endif + return cbm_index_task_temp_bytes(handle->task_db_path, handle->stage_token, handle->log_path, + handle->response_path, bytes_out); +} + +static void worker_check_task_temp_limit(cbm_index_worker_handle_t *handle) { + const cbm_index_limit_u64_t *limit = &handle->resource_policy.max_task_temp_bytes; + if (!limit->enabled || + atomic_load_explicit(&handle->termination_reason, memory_order_acquire) != + INDEX_WORKER_TERMINATION_NONE || + cbm_subprocess_termination_pending(handle->process) || + !cbm_subprocess_supervision_active(handle->process)) { + return; + } + uint64_t now = worker_resource_now_ms(); + bool probe_due = + !handle->task_temp_probe_started || now < handle->last_task_temp_probe_ms || + now - handle->last_task_temp_probe_ms >= INDEX_WORKER_TASK_TEMP_PROBE_INTERVAL_MS; + if (!probe_due) { + return; + } + handle->task_temp_probe_started = true; + handle->last_task_temp_probe_ms = now; + uint64_t observed = 0; + bool measured = worker_task_temp_bytes(handle, &observed); + if (!measured || observed > limit->value) { + worker_request_resource_termination(handle, CBM_INDEX_RESOURCE_TASK_TEMP_BYTES, observed, + limit->value, !measured); + } +} + static int worker_start_internal(const char *args_json, size_t memory_budget_bytes, const cbm_index_resource_policy_t *resource_policy, - bool single_thread, const char *marker_file, - const char *quarantine_file, cbm_proc_log_cb log_callback, - void *log_context, uint64_t duration_origin_ms, + const char *task_db_path, bool single_thread, + const char *marker_file, const char *quarantine_file, + cbm_proc_log_cb log_callback, void *log_context, + uint64_t duration_origin_ms, cbm_index_worker_handle_t **handle_out) { if (handle_out) { *handle_out = NULL; @@ -1098,6 +1186,10 @@ static int worker_start_internal(const char *args_json, size_t memory_budget_byt if (!handle_out || !args_json || !args_json[0]) { return -1; } + if (resource_policy && resource_policy->max_task_temp_bytes.enabled && + (!task_db_path || !task_db_path[0] || strlen(task_db_path) >= INDEX_WORKER_PATH_CAP)) { + return -1; + } if (g_host_marked && !g_worker_active && supervisor_disable_requested()) { /* Keep the old variable as a deterministic refusal seam, but never as * a production escape hatch into unsafe in-process indexing. */ @@ -1123,6 +1215,9 @@ static int worker_start_internal(const char *args_json, size_t memory_budget_byt if (resource_policy) { handle->resource_policy = *resource_policy; } + if (handle->resource_policy.max_task_temp_bytes.enabled) { + (void)snprintf(handle->task_db_path, sizeof(handle->task_db_path), "%s", task_db_path); + } atomic_init(&handle->termination_reason, INDEX_WORKER_TERMINATION_NONE); atomic_init(&handle->terminal, false); handle->log_callback = log_callback; @@ -1173,6 +1268,8 @@ static int worker_start_internal(const char *args_json, size_t memory_budget_byt argv[argc++] = CBM_INDEX_WORKER_PUBLISH_FENCE_ARG; argv[argc++] = handle->publish_fence_path; } + argv[argc++] = CBM_INDEX_WORKER_STAGE_TOKEN_ARG; + argv[argc++] = handle->stage_token; argv[argc] = NULL; cbm_proc_opts_t options = {0}; @@ -1213,15 +1310,16 @@ int cbm_index_worker_start_with_log(const char *args_json, size_t memory_budget_ bool single_thread, const char *marker_file, const char *quarantine_file, cbm_proc_log_cb log_callback, void *log_context, cbm_index_worker_handle_t **handle_out) { - return worker_start_internal(args_json, memory_budget_bytes, NULL, single_thread, marker_file, - quarantine_file, log_callback, log_context, 0, handle_out); + return worker_start_internal(args_json, memory_budget_bytes, NULL, NULL, single_thread, + marker_file, quarantine_file, log_callback, log_context, 0, + handle_out); } int cbm_index_worker_start(const char *args_json, size_t memory_budget_bytes, bool single_thread, const char *marker_file, const char *quarantine_file, cbm_index_worker_handle_t **handle_out) { - return worker_start_internal(args_json, memory_budget_bytes, NULL, single_thread, marker_file, - quarantine_file, NULL, NULL, 0, handle_out); + return worker_start_internal(args_json, memory_budget_bytes, NULL, NULL, single_thread, + marker_file, quarantine_file, NULL, NULL, 0, handle_out); } int cbm_index_worker_start_with_policy(const char *args_json, size_t memory_budget_bytes, @@ -1229,9 +1327,9 @@ int cbm_index_worker_start_with_policy(const char *args_json, size_t memory_budg bool single_thread, const char *marker_file, const char *quarantine_file, uint64_t duration_origin_ms, cbm_index_worker_handle_t **handle_out) { - return worker_start_internal(args_json, memory_budget_bytes, resource_policy, single_thread, - marker_file, quarantine_file, NULL, NULL, duration_origin_ms, - handle_out); + return worker_start_internal(args_json, memory_budget_bytes, resource_policy, NULL, + single_thread, marker_file, quarantine_file, NULL, NULL, + duration_origin_ms, handle_out); } int cbm_index_worker_start_with_policy_log(const char *args_json, size_t memory_budget_bytes, @@ -1241,9 +1339,21 @@ int cbm_index_worker_start_with_policy_log(const char *args_json, size_t memory_ cbm_proc_log_cb log_callback, void *log_context, uint64_t duration_origin_ms, cbm_index_worker_handle_t **handle_out) { - return worker_start_internal(args_json, memory_budget_bytes, resource_policy, single_thread, - marker_file, quarantine_file, log_callback, log_context, - duration_origin_ms, handle_out); + return worker_start_internal(args_json, memory_budget_bytes, resource_policy, NULL, + single_thread, marker_file, quarantine_file, log_callback, + log_context, duration_origin_ms, handle_out); +} + +int cbm_index_worker_start_with_storage_policy(const char *args_json, size_t memory_budget_bytes, + const cbm_index_resource_policy_t *resource_policy, + const char *task_db_path, bool single_thread, + const char *marker_file, const char *quarantine_file, + cbm_proc_log_cb log_callback, void *log_context, + uint64_t duration_origin_ms, + cbm_index_worker_handle_t **handle_out) { + return worker_start_internal(args_json, memory_budget_bytes, resource_policy, task_db_path, + single_thread, marker_file, quarantine_file, log_callback, + log_context, duration_origin_ms, handle_out); } cbm_index_worker_poll_t cbm_index_worker_poll(cbm_index_worker_handle_t *handle, @@ -1261,6 +1371,7 @@ cbm_index_worker_poll_t cbm_index_worker_poll(cbm_index_worker_handle_t *handle, bool relay_caught_up = true; if (!handle->process_terminal) { bool rss_probe_failed = worker_check_resource_limits(handle); + worker_check_task_temp_limit(handle); cbm_proc_result_t process_result; cbm_proc_poll_t state = cbm_subprocess_poll(handle->process, &process_result); relay_caught_up = worker_relay_log(handle); @@ -1326,6 +1437,14 @@ cbm_index_worker_poll_t cbm_index_worker_poll(cbm_index_worker_handle_t *handle, handle->log_path); } } + if (termination_reason == INDEX_WORKER_TERMINATION_RESOURCE && + handle->result.resource_violation.resource == CBM_INDEX_RESOURCE_TASK_TEMP_BYTES) { + if (!handle->result.tree_quiesced || handle->result.supervision_failed) { + cbm_log_warn("index.supervisor.staging_cleanup_deferred", "path", handle->task_db_path); + } else if (!cbm_index_staging_cleanup(handle->task_db_path, handle->stage_token)) { + cbm_log_warn("index.supervisor.staging_cleanup_failed", "path", handle->task_db_path); + } + } (void)cbm_unlink(handle->response_path); worker_terminal_log(handle); atomic_store_explicit(&handle->terminal, true, memory_order_release); @@ -1390,18 +1509,19 @@ void cbm_index_worker_destroy(cbm_index_worker_handle_t *handle) { static int worker_spawn_internal(const char *args_json, const cbm_index_resource_policy_t *resource_policy, - bool single_thread, const char *marker_file, - const char *quarantine_file, cbm_proc_log_cb log_callback, - void *log_context, const atomic_int *cancel_requested, - uint64_t duration_origin_ms, cbm_index_worker_result_t *result) { + const char *task_db_path, bool single_thread, + const char *marker_file, const char *quarantine_file, + cbm_proc_log_cb log_callback, void *log_context, + const atomic_int *cancel_requested, uint64_t duration_origin_ms, + cbm_index_worker_result_t *result) { if (!result) { return -1; } worker_result_init(result); cbm_index_worker_handle_t *handle = NULL; - if (worker_start_internal(args_json, 0, resource_policy, single_thread, marker_file, - quarantine_file, log_callback, log_context, duration_origin_ms, - &handle) != 0) { + if (worker_start_internal(args_json, 0, resource_policy, task_db_path, single_thread, + marker_file, quarantine_file, log_callback, log_context, + duration_origin_ms, &handle) != 0) { return -1; } const cbm_index_worker_result_t *cached = NULL; @@ -1432,7 +1552,7 @@ int cbm_index_spawn_worker_with_log_cancel(const char *args_json, bool single_th cbm_proc_log_cb log_callback, void *log_context, const atomic_int *cancel_requested, cbm_index_worker_result_t *result) { - return worker_spawn_internal(args_json, NULL, single_thread, marker_file, quarantine_file, + return worker_spawn_internal(args_json, NULL, NULL, single_thread, marker_file, quarantine_file, log_callback, log_context, cancel_requested, 0, result); } @@ -1441,11 +1561,22 @@ int cbm_index_spawn_worker_with_policy_log_cancel( const char *marker_file, const char *quarantine_file, cbm_proc_log_cb log_callback, void *log_context, const atomic_int *cancel_requested, uint64_t duration_origin_ms, cbm_index_worker_result_t *result) { - return worker_spawn_internal(args_json, resource_policy, single_thread, marker_file, + return worker_spawn_internal(args_json, resource_policy, NULL, single_thread, marker_file, quarantine_file, log_callback, log_context, cancel_requested, duration_origin_ms, result); } +int cbm_index_spawn_worker_with_storage_policy_log_cancel( + const char *args_json, const cbm_index_resource_policy_t *resource_policy, + const char *task_db_path, bool single_thread, const char *marker_file, + const char *quarantine_file, cbm_proc_log_cb log_callback, void *log_context, + const atomic_int *cancel_requested, uint64_t duration_origin_ms, + cbm_index_worker_result_t *result) { + return worker_spawn_internal(args_json, resource_policy, task_db_path, single_thread, + marker_file, quarantine_file, log_callback, log_context, + cancel_requested, duration_origin_ms, result); +} + int cbm_index_spawn_worker_with_log(const char *args_json, bool single_thread, const char *marker_file, const char *quarantine_file, cbm_proc_log_cb log_callback, void *log_context, diff --git a/src/mcp/index_supervisor.h b/src/mcp/index_supervisor.h index 359a3e25e9..cd4b803953 100644 --- a/src/mcp/index_supervisor.h +++ b/src/mcp/index_supervisor.h @@ -43,11 +43,12 @@ void cbm_index_set_worker_role(bool is_worker, const char *response_out); #define CBM_INDEX_WORKER_MEMORY_BUDGET_ARG "--index-worker-memory-budget-bytes" #define CBM_INDEX_WORKER_BUILD_ARG "--index-worker-build" #define CBM_INDEX_WORKER_PUBLISH_FENCE_ARG "--index-worker-publish-fence" +#define CBM_INDEX_WORKER_STAGE_TOKEN_ARG "--index-worker-stage-token" #define CBM_INDEX_WORKER_BUILD_FINGERPRINT_LENGTH 64U #define CBM_INDEX_WORKER_BUILD_FINGERPRINT_SIZE 65U void cbm_index_set_worker_role_options(bool is_worker, const char *response_out, bool single_thread, const char *marker_file, const char *quarantine_file, - size_t memory_budget_bytes); + size_t memory_budget_bytes, const char *stage_token); bool cbm_index_worker_active(void); const char *cbm_index_worker_response_out(void); size_t cbm_index_worker_memory_budget_bytes(void); @@ -124,6 +125,7 @@ typedef struct { const char *quarantine_file; size_t memory_budget_bytes; const char *publish_fence_file; + const char *stage_token; } cbm_index_worker_invocation_t; typedef enum { @@ -222,6 +224,13 @@ int cbm_index_worker_start_with_policy(const char *args_json, size_t memory_budg bool single_thread, const char *marker_file, const char *quarantine_file, uint64_t duration_origin_ms, cbm_index_worker_handle_t **handle_out); +int cbm_index_worker_start_with_storage_policy(const char *args_json, size_t memory_budget_bytes, + const cbm_index_resource_policy_t *resource_policy, + const char *task_db_path, bool single_thread, + const char *marker_file, const char *quarantine_file, + cbm_proc_log_cb log_callback, void *log_context, + uint64_t duration_origin_ms, + cbm_index_worker_handle_t **handle_out); /* Policy plus the request-scoped log callback. The daemon default start path * needs both: a missing policy fails closed, and worker diagnostics still @@ -305,6 +314,12 @@ int cbm_index_spawn_worker_with_policy_log_cancel( const char *marker_file, const char *quarantine_file, cbm_proc_log_cb log_callback, void *log_context, const atomic_int *cancel_requested, uint64_t duration_origin_ms, cbm_index_worker_result_t *result); +int cbm_index_spawn_worker_with_storage_policy_log_cancel( + const char *args_json, const cbm_index_resource_policy_t *resource_policy, + const char *task_db_path, bool single_thread, const char *marker_file, + const char *quarantine_file, cbm_proc_log_cb log_callback, void *log_context, + const atomic_int *cancel_requested, uint64_t duration_origin_ms, + cbm_index_worker_result_t *result); void cbm_index_worker_result_free(cbm_index_worker_result_t *result); @@ -320,6 +335,12 @@ void cbm_index_supervisor_reset_resource_hooks_for_testing(void); /* Borrowed fence path of an attempt; NULL when it has no fence. */ const char *cbm_index_worker_publish_fence_path_for_testing( const cbm_index_worker_handle_t *handle); +typedef bool (*cbm_index_supervisor_task_temp_fn)(const char *final_db_path, const char *log_path, + const char *response_path, uint64_t *bytes_out, + void *context); +void cbm_index_supervisor_set_task_temp_hook_for_testing( + cbm_index_supervisor_task_temp_fn task_temp_fn, void *context); +void cbm_index_supervisor_reset_task_temp_hook_for_testing(void); #endif #endif /* CBM_INDEX_SUPERVISOR_H */ diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index 694867276e..b76aa2609c 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -12227,6 +12227,9 @@ void cbm_mcp_index_request_facts_record(cbm_mcp_index_request_facts_t *facts, char *cbm_mcp_index_worker_resource_response(const char *args, const cbm_index_worker_result_t *worker_result, const cbm_mcp_index_request_facts_t *facts) { + if (!worker_result || worker_result->resource_violation.resource == CBM_INDEX_RESOURCE_NONE) { + return NULL; + } char *repo_path = cbm_mcp_get_string_arg(args, "repo_path"); /* The publication fence makes a limit-stopped attempt and a publication by * that attempt mutually exclusive; an EARLIER attempt of the request may @@ -12237,23 +12240,23 @@ char *cbm_mcp_index_worker_resource_response(const char *args, !worker_result->publication_committed && index_args_serving_index_servable(args); const cbm_index_resource_violation_t *violation = &worker_result->resource_violation; + bool probe_failed = worker_result->resource_probe_failed || violation->probe_failed; + bool worker_stage = violation->resource == CBM_INDEX_RESOURCE_RSS_BYTES || + violation->resource == CBM_INDEX_RESOURCE_DURATION_MS; const char *config_key = cbm_index_resource_config_key(violation->resource); char message[CBM_SZ_256]; (void)snprintf(message, sizeof(message), - worker_result->resource_probe_failed - ? "Worker resource measurement failed for %s" - : "Index worker exceeded %s", + probe_failed ? "Index resource measurement failed for %s" : "Index exceeded %s", config_key); yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL); yyjson_mut_val *root = yyjson_mut_obj(doc); yyjson_mut_doc_set_root(doc, root); yyjson_mut_obj_add_str(doc, root, "status", "error"); yyjson_mut_obj_add_str(doc, root, "code", - worker_result->resource_probe_failed ? "resource_probe_failed" - : "resource_limit_exceeded"); - yyjson_mut_obj_add_str(doc, root, "stage", "worker"); + probe_failed ? "resource_probe_failed" : "resource_limit_exceeded"); + yyjson_mut_obj_add_str(doc, root, "stage", worker_stage ? "worker" : "storage"); yyjson_mut_obj_add_str(doc, root, "resource", cbm_index_resource_name(violation->resource)); - if (!worker_result->resource_probe_failed) { + if (!probe_failed) { yyjson_mut_obj_add_uint(doc, root, "observed", violation->observed); yyjson_mut_obj_add_uint(doc, root, "limit", violation->limit); yyjson_mut_obj_add_str(doc, root, "unit", cbm_index_resource_unit(violation->resource)); @@ -12448,6 +12451,22 @@ bool cbm_mcp_index_policy_from_internal_args(const char *args, cbm_index_resourc return valid; } +bool cbm_mcp_index_task_db_path(const char *args, char *path_out, size_t path_size) { + if (!path_out || path_size == 0) { + return false; + } + path_out[0] = '\0'; + char *repo_path = cbm_mcp_get_string_arg(args, "repo_path"); + char *name_override = cbm_mcp_get_string_arg(args, "name"); + char *project = + cbm_project_name_from_path(name_override && name_override[0] ? name_override : repo_path); + char cache_path[CBM_SZ_1K]; + cache_dir(cache_path, sizeof(cache_path)); + int written = project ? snprintf(path_out, path_size, "%s/%s.db", cache_path, project) : -1; + index_request_names_free(repo_path, name_override, project); + return written > 0 && (size_t)written < path_size; +} + static bool load_index_policy(cbm_mcp_server_t *srv, const char *args, cbm_index_resource_policy_t *policy, char *error, size_t error_size) { if (cbm_index_worker_active()) { @@ -12565,19 +12584,24 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, const cbm_index_resource_policy_t *resource_policy) { invalidate_cached_store(srv); uint64_t request_started_ms = cbm_index_worker_now_ms(); - /* Only an enabled worker limit can produce a worker resource failure, the - * one response these facts are reported in. */ + char task_db_path[CBM_SZ_1K]; + if (!resource_policy || !cbm_mcp_index_task_db_path(args, task_db_path, sizeof(task_db_path))) { + return cbm_mcp_text_result("could not resolve worker index path", true); + } + /* Only an enabled worker or task-temp limit can produce a worker resource + * failure, the one response these facts are reported in. */ cbm_mcp_index_request_facts_t facts = {0}; - if (cbm_index_policy_worker_enabled(resource_policy)) { + if (cbm_index_policy_worker_enabled(resource_policy) || + resource_policy->max_task_temp_bytes.enabled) { cbm_mcp_index_request_facts_begin(args, &facts); } /* First attempt: normal parallel run. */ cbm_index_worker_result_t wr; - int rc = cbm_index_spawn_worker_with_policy_log_cancel( - args, resource_policy, false, NULL, NULL, srv ? srv->index_log_callback : NULL, - srv ? srv->index_log_context : NULL, srv ? &srv->pipeline_cancel_requested : NULL, - request_started_ms, &wr); + int rc = cbm_index_spawn_worker_with_storage_policy_log_cancel( + args, resource_policy, task_db_path, false, NULL, NULL, + srv ? srv->index_log_callback : NULL, srv ? srv->index_log_context : NULL, + srv ? &srv->pipeline_cancel_requested : NULL, request_started_ms, &wr); cbm_mcp_supervised_result_disposition_t disposition = cbm_mcp_supervised_result_disposition(rc, &wr); cbm_mcp_index_request_facts_record(&facts, &wr); @@ -12616,7 +12640,6 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, invalidate_cached_store(srv); return resp; } - /* Crash / hang / nonzero exit → skip-and-continue recovery. Re-run the * worker PARALLEL (there are no sequential production runs) with the * per-file marker JOURNAL armed; after each failed run the journal's @@ -12657,10 +12680,11 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, bool terminal_cancelled = false; for (int i = 0; i < cap; i++) { cbm_index_worker_result_t wr2; - int rc2 = cbm_index_spawn_worker_with_policy_log_cancel( - args, resource_policy, /*single_thread=*/false, marker_path, quarantine_path, - srv ? srv->index_log_callback : NULL, srv ? srv->index_log_context : NULL, - srv ? &srv->pipeline_cancel_requested : NULL, request_started_ms, &wr2); + int rc2 = cbm_index_spawn_worker_with_storage_policy_log_cancel( + args, resource_policy, task_db_path, /*single_thread=*/false, marker_path, + quarantine_path, srv ? srv->index_log_callback : NULL, + srv ? srv->index_log_context : NULL, srv ? &srv->pipeline_cancel_requested : NULL, + request_started_ms, &wr2); cbm_mcp_supervised_result_disposition_t recovery_disposition = cbm_mcp_supervised_result_disposition(rc2, &wr2); cbm_mcp_index_request_facts_record(&facts, &wr2); @@ -12785,8 +12809,8 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, * so it cannot itself hang. Rare given monotonic progress. */ if (!resp && !unsafe_terminal && quarantined > 0) { cbm_index_worker_result_t wrp; - int rcp = cbm_index_spawn_worker_with_policy_log_cancel( - args, resource_policy, /*single_thread=*/false, NULL, quarantine_path, + int rcp = cbm_index_spawn_worker_with_storage_policy_log_cancel( + args, resource_policy, task_db_path, /*single_thread=*/false, NULL, quarantine_path, srv ? srv->index_log_callback : NULL, srv ? srv->index_log_context : NULL, srv ? &srv->pipeline_cancel_requested : NULL, request_started_ms, &wrp); cbm_mcp_supervised_result_disposition_t partial_disposition = @@ -13636,11 +13660,18 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { } else if (rc == CBM_PIPELINE_RESOURCE_LIMIT && resource_violation.resource != CBM_INDEX_RESOURCE_NONE) { const char *config_key = cbm_index_resource_config_key(resource_violation.resource); + bool discovery_resource = resource_violation.resource == CBM_INDEX_RESOURCE_FILES || + resource_violation.resource == CBM_INDEX_RESOURCE_SOURCE_BYTES; char message[CBM_SZ_256]; - (void)snprintf(message, sizeof(message), "Index discovery exceeded %s", config_key); + (void)snprintf(message, sizeof(message), + resource_violation.probe_failed ? "Index resource probe failed for %s" + : "Index exceeded %s", + config_key); yyjson_mut_obj_add_str(doc, root, "status", "error"); - yyjson_mut_obj_add_str(doc, root, "code", "resource_limit_exceeded"); - yyjson_mut_obj_add_str(doc, root, "stage", "discovery"); + yyjson_mut_obj_add_str(doc, root, "code", + resource_violation.probe_failed ? "resource_probe_failed" + : "resource_limit_exceeded"); + yyjson_mut_obj_add_str(doc, root, "stage", discovery_resource ? "discovery" : "storage"); yyjson_mut_obj_add_str(doc, root, "resource", cbm_index_resource_name(resource_violation.resource)); yyjson_mut_obj_add_uint(doc, root, "observed", resource_violation.observed); @@ -22137,11 +22168,18 @@ static void *autoindex_thread(void *arg) { return NULL; } + cbm_index_resource_policy_t resource_policy; + char policy_error[CBM_SZ_256] = {0}; + if (!load_index_policy(srv, NULL, &resource_policy, policy_error, sizeof(policy_error))) { + cbm_log_warn("autoindex.err", "msg", "resource_policy_load_failed", "error", policy_error); + return NULL; + } cbm_pipeline_t *p = cbm_pipeline_new(srv->session_root, NULL, CBM_MODE_FULL); if (!p) { cbm_log_warn("autoindex.err", "msg", "pipeline_create_failed"); return NULL; } + cbm_pipeline_set_resource_policy(p, &resource_policy); index_doc_candidates_apply(p, index_doc_candidates_choice(srv, NULL)); cbm_pipeline_set_background(p, project_has_committed_index(cbm_pipeline_project_name(p))); diff --git a/src/mcp/mcp_internal.h b/src/mcp/mcp_internal.h index 5df164b1a0..e5c1e5978e 100644 --- a/src/mcp/mcp_internal.h +++ b/src/mcp/mcp_internal.h @@ -62,6 +62,7 @@ bool cbm_mcp_index_policy_add_to_args(yyjson_mut_doc *doc, yyjson_mut_val *root, const cbm_index_resource_policy_t *policy); bool cbm_mcp_index_policy_from_internal_args(const char *args, cbm_index_resource_policy_t *policy, char *error, size_t error_size); +bool cbm_mcp_index_task_db_path(const char *args, char *path_out, size_t path_size); /* Request-scoped facts a worker resource failure is reported against. One * supervised request can run several worker attempts (crash/hang recovery), and diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 3f61e122d3..aeeaec2b1c 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -217,6 +217,10 @@ struct cbm_pipeline { cbm_index_resource_policy_t resource_policy; cbm_index_resource_violation_t resource_violation; bool background; /* reserve CPU headroom when no user is waiting */ +#ifdef CBM_ENABLE_TEST_SEAMS + cbm_pipeline_storage_probe_fn storage_probe; + void *storage_probe_context; +#endif /* Snapshot of the artifact export failure of THIS run (set only by * export_after_publish failure, zeroed at run start, cleared on success). @@ -985,6 +989,17 @@ void cbm_pipeline_set_background(cbm_pipeline_t *p, bool background) { } } +#ifdef CBM_ENABLE_TEST_SEAMS +void cbm_pipeline_set_storage_probe_for_testing(cbm_pipeline_t *p, + cbm_pipeline_storage_probe_fn probe, + void *context) { + if (p) { + p->storage_probe = probe; + p->storage_probe_context = context; + } +} +#endif + bool cbm_pipeline_set_project_name(cbm_pipeline_t *p, const char *name) { if (!p || !name || !name[0]) { return false; @@ -3774,7 +3789,6 @@ static int publish_generation_impl(const cbm_pipeline_generation_t *generation, if (generation->cancelled && atomic_load(generation->cancelled)) { return CBM_PIPELINE_ABORT_PRESERVE_DB; } - /* The staging name must be unpredictable and created exclusively. It used * to be ".stage..", which any other process can compute * in advance; this path is then unlinked and written, so in a @@ -4614,6 +4628,216 @@ static int cbm_pipeline_run_staged(cbm_pipeline_t *p) { return rc; } +static const cbm_index_limit_u64_t *storage_limit_for_resource( + const cbm_index_resource_policy_t *policy, cbm_index_resource_t resource) { + if (!policy) { + return NULL; + } + switch (resource) { + case CBM_INDEX_RESOURCE_CACHE_BYTES: + return &policy->max_cache_bytes; + case CBM_INDEX_RESOURCE_FREE_DISK_BYTES: + return &policy->min_free_disk_bytes; + case CBM_INDEX_RESOURCE_FINAL_DB_BYTES: + return &policy->max_final_db_bytes; + case CBM_INDEX_RESOURCE_STAGING_BYTES: + return &policy->max_staging_bytes; + case CBM_INDEX_RESOURCE_TASK_TEMP_BYTES: + return &policy->max_task_temp_bytes; + default: + return NULL; + } +} + +static cbm_index_resource_t storage_enabled_probe_resource( + const cbm_index_resource_policy_t *policy) { + if (policy->max_cache_bytes.enabled) { + return CBM_INDEX_RESOURCE_CACHE_BYTES; + } + if (policy->min_free_disk_bytes.enabled) { + return CBM_INDEX_RESOURCE_FREE_DISK_BYTES; + } + if (policy->max_staging_bytes.enabled) { + return CBM_INDEX_RESOURCE_STAGING_BYTES; + } + if (policy->max_task_temp_bytes.enabled) { + return CBM_INDEX_RESOURCE_TASK_TEMP_BYTES; + } + return policy->max_final_db_bytes.enabled ? CBM_INDEX_RESOURCE_FINAL_DB_BYTES + : CBM_INDEX_RESOURCE_NONE; +} + +static char *storage_root_from_final_path(const char *final_path) { + if (!final_path) { + return NULL; + } + char *root = cbm_mem_strdup(CBM_MEM_CLASS_OTHER, final_path); + if (!root) { + return NULL; + } + char *slash = strrchr(root, '/'); +#ifdef _WIN32 + char *backslash = strrchr(root, '\\'); + if (backslash && (!slash || backslash > slash)) { + slash = backslash; + } +#endif + if (!slash) { + cbm_free(CBM_MEM_CLASS_OTHER, root); + return cbm_mem_strdup(CBM_MEM_CLASS_OTHER, "."); + } + if (slash == root) { + slash[1] = '\0'; + } else { + *slash = '\0'; + } + return root; +} + +static bool storage_regular_file_bytes(const char *path, uint64_t *bytes_out) { + if (!path || !bytes_out) { + return false; + } + cbm_path_info_t info; + int status = cbm_path_info_utf8(path, &info); + if (status == CBM_PATH_INFO_ABSENT) { + *bytes_out = 0; + return true; + } + if (status != CBM_PATH_INFO_OK || !info.is_regular || info.is_symlink || info.size < 0) { + return false; + } + *bytes_out = (uint64_t)info.size; + return true; +} + +static bool storage_replaceable_old_bytes(const char *final_path, uint64_t *bytes_out) { + if (!final_path || !bytes_out) { + return false; + } + *bytes_out = 0; + cbm_path_info_t info; + int status = cbm_path_info_utf8(final_path, &info); + if (status == CBM_PATH_INFO_ABSENT) { + return true; + } + if (status != CBM_PATH_INFO_OK || !info.is_regular || info.is_symlink) { + return false; + } + cbm_store_t *store = cbm_store_open_path_query(final_path); + if (!store) { + /* A file that cannot be opened is not proven replaceable. Keep its + * bytes in the projection rather than blocking repair of a confirmed + * corrupt/non-database destination. */ + return true; + } + cbm_integrity_verdict_t verdict = cbm_store_check_integrity_verdict(store); + cbm_store_close(store); + if (verdict == CBM_INTEGRITY_TRANSIENT) { + return false; + } + /* A generation that cannot be proven healthy remains in current_cache but is + * not deducted: publication may quarantine it instead of replacing it. */ + return verdict != CBM_INTEGRITY_OK || cbm_db_artifact_bytes(final_path, bytes_out); +} + +static bool storage_probe_default(const cbm_index_resource_policy_t *policy, + cbm_pipeline_storage_checkpoint_t checkpoint, + const char *final_path, const char *staging_path, + cbm_index_storage_sample_t *sample, + cbm_index_resource_t *failed_resource) { + *sample = (cbm_index_storage_sample_t){0}; + *failed_resource = CBM_INDEX_RESOURCE_NONE; + char *storage_root = NULL; + if (policy->max_cache_bytes.enabled || policy->min_free_disk_bytes.enabled) { + storage_root = storage_root_from_final_path(final_path); + if (!storage_root) { + *failed_resource = storage_enabled_probe_resource(policy); + return false; + } + } + uint64_t operation_bytes = 0; + bool need_operation = + staging_path && + (policy->max_cache_bytes.enabled || policy->max_staging_bytes.enabled || + policy->max_task_temp_bytes.enabled || + (checkpoint == CBM_PIPELINE_STORAGE_PREPUBLISH && policy->max_final_db_bytes.enabled)); + if (need_operation && !cbm_db_artifact_bytes(staging_path, &operation_bytes)) { + *failed_resource = + policy->max_staging_bytes.enabled + ? CBM_INDEX_RESOURCE_STAGING_BYTES + : (policy->max_task_temp_bytes.enabled + ? CBM_INDEX_RESOURCE_TASK_TEMP_BYTES + : (policy->max_final_db_bytes.enabled ? CBM_INDEX_RESOURCE_FINAL_DB_BYTES + : CBM_INDEX_RESOURCE_CACHE_BYTES)); + cbm_free(CBM_MEM_CLASS_OTHER, storage_root); + return false; + } + if (policy->max_cache_bytes.enabled) { + uint64_t cache_with_operation = 0; + if (!cbm_directory_size_bytes(storage_root, &cache_with_operation) || + !storage_replaceable_old_bytes(final_path, &sample->replaceable_old_bytes)) { + *failed_resource = CBM_INDEX_RESOURCE_CACHE_BYTES; + cbm_free(CBM_MEM_CLASS_OTHER, storage_root); + return false; + } + sample->current_cache_bytes = + cache_with_operation > operation_bytes ? cache_with_operation - operation_bytes : 0; + sample->operation_bytes = operation_bytes; + } + if (policy->min_free_disk_bytes.enabled && + !cbm_filesystem_free_bytes(storage_root, &sample->free_disk_bytes)) { + *failed_resource = CBM_INDEX_RESOURCE_FREE_DISK_BYTES; + cbm_free(CBM_MEM_CLASS_OTHER, storage_root); + return false; + } + cbm_free(CBM_MEM_CLASS_OTHER, storage_root); + if (checkpoint != CBM_PIPELINE_STORAGE_PREFLIGHT) { + sample->staging_bytes = operation_bytes; + sample->task_temp_bytes = operation_bytes; + } + if (checkpoint == CBM_PIPELINE_STORAGE_PREPUBLISH && policy->max_final_db_bytes.enabled && + !storage_regular_file_bytes(staging_path, &sample->final_db_bytes)) { + *failed_resource = CBM_INDEX_RESOURCE_FINAL_DB_BYTES; + return false; + } + return true; +} + +int cbm_pipeline_storage_admit(cbm_pipeline_t *p, cbm_pipeline_storage_checkpoint_t checkpoint, + const char *final_path, const char *staging_path) { + if (!p || !cbm_index_policy_storage_enabled(&p->resource_policy)) { + return 0; + } + cbm_index_storage_sample_t sample; + cbm_index_resource_t failed_resource = CBM_INDEX_RESOURCE_NONE; + bool measured; +#ifdef CBM_ENABLE_TEST_SEAMS + if (p->storage_probe) { + measured = p->storage_probe(checkpoint, final_path, staging_path, &sample, &failed_resource, + p->storage_probe_context); + } else +#endif + { + measured = storage_probe_default(&p->resource_policy, checkpoint, final_path, staging_path, + &sample, &failed_resource); + } + if (!measured) { + const cbm_index_limit_u64_t *limit = + storage_limit_for_resource(&p->resource_policy, failed_resource); + p->resource_violation = (cbm_index_resource_violation_t){ + .resource = failed_resource, + .limit = limit && limit->enabled ? limit->value : 0, + .probe_failed = true, + }; + return CBM_PIPELINE_RESOURCE_LIMIT; + } + if (!cbm_index_policy_check_storage(&p->resource_policy, &sample, &p->resource_violation)) { + return CBM_PIPELINE_RESOURCE_LIMIT; + } + return 0; +} + static void cleanup_staging_db(const char *path) { if (!path) { return; @@ -4648,66 +4872,122 @@ static bool ensure_db_parent(const char *path) { } /* Length of the path a stage was minted for: the input itself unless its - * basename has exactly the minted shape ".stage.<6 alphanumerics>", in - * which case the root is . The outer run rewrites the pipeline's db_path - * to its stage, so the inner publication (dump and delta clone) used to mint - * ITS stage from that stage: .stage.A.stage.B, with -wal/-shm beside it - * (#1839). Minting from the root keeps every generation's stage a sibling of - * the live database. Only the exact minted shape is recognised: a database - * named "x.stage.y.db" is not a stage and keeps its full name. */ + * basename has exactly the minted shape ".stage.<6 alphanumerics>" or + * ".stage..<6 alphanumerics>", in which case the root is . + * The outer run rewrites the pipeline's db_path to its stage, so the inner + * publication (dump and delta clone) used to mint ITS stage from that stage: + * .stage.A.stage.B, with -wal/-shm beside it (#1839). Minting from the + * root keeps every generation's stage a sibling of the live database. Only + * the exact minted shape is recognised: a database named "x.stage.y.db" is + * not a stage and keeps its full name. A worker stage token, when present, + * sits between the marker and the random suffix so the supervisor can find + * this attempt's temporary files. */ enum { CBM_STAGE_SUFFIX_RANDOM_CHARS = 6 }; static const char cbm_stage_marker[] = ".stage."; -static bool stage_suffix_at(const char *tail) { - if (strncmp(tail, cbm_stage_marker, sizeof(cbm_stage_marker) - 1) != 0) { +static bool stage_token_char(unsigned char ch) { + return (ch >= '0' && ch <= '9') || (ch >= 'A' && ch <= 'Z') || (ch >= 'a' && ch <= 'z') || + ch == '-' || ch == '_'; +} + +static bool stage_random_suffix(const char *text) { + for (int i = 0; i < CBM_STAGE_SUFFIX_RANDOM_CHARS; i++) { + if (!isalnum((unsigned char)text[i])) { + return false; + } + } + return text[CBM_STAGE_SUFFIX_RANDOM_CHARS] == '\0'; +} + +/* after ".stage.": 6 alphanumerics, or .<6 alphanumerics>, through end. */ +static bool stage_body_at(const char *after_marker) { + if (stage_random_suffix(after_marker)) { + return true; + } + const char *dot = strchr(after_marker, '.'); + if (!dot || dot == after_marker) { return false; } - const char *random = tail + sizeof(cbm_stage_marker) - 1; - for (int i = 0; i < CBM_STAGE_SUFFIX_RANDOM_CHARS; i++) { - if (!isalnum((unsigned char)random[i])) { + size_t token_len = (size_t)(dot - after_marker); + if (token_len < 6U || token_len >= CBM_SZ_64) { + return false; + } + for (size_t i = 0; i < token_len; i++) { + if (!stage_token_char((unsigned char)after_marker[i])) { return false; } } - return random[CBM_STAGE_SUFFIX_RANDOM_CHARS] == '\0'; + return stage_random_suffix(dot + 1); } static size_t stage_root_length(const char *path) { size_t len = strlen(path); - const size_t suffix_len = sizeof(cbm_stage_marker) - 1 + CBM_STAGE_SUFFIX_RANDOM_CHARS; - if (len <= suffix_len) { + const char *marker = NULL; + for (const char *next = strstr(path, cbm_stage_marker); next; + next = strstr(next + 1, cbm_stage_marker)) { + marker = next; + } + if (!marker) { return len; } - size_t root_len = len - suffix_len; - /* The marker must sit inside the basename, never span a separator. */ - for (size_t i = root_len; i < len; i++) { - if (path[i] == '/' + for (const char *scan = marker; *scan; scan++) { + if (*scan == '/' #ifdef _WIN32 - || path[i] == '\\' + || *scan == '\\' #endif ) { return len; } } - return stage_suffix_at(path + root_len) ? root_len : len; + if (!stage_body_at(marker + sizeof(cbm_stage_marker) - 1)) { + return len; + } + return (size_t)(marker - path); +} + +static bool staging_token_valid(const char *token) { + size_t length = token ? strlen(token) : 0; + if (length < 6U || length >= CBM_SZ_64) { + return false; + } + for (size_t index = 0; index < length; index++) { + if (!stage_token_char((unsigned char)token[index])) { + return false; + } + } + return true; } static char *stage_allocate_template(const char *final_path) { - if (!final_path) + if (!final_path) { + return NULL; + } + char suffix[CBM_SZ_128]; + const char *stage_token = getenv("CBM_INDEX_STAGE_TOKEN"); + if (!staging_token_valid(stage_token)) { + stage_token = NULL; + } + int suffix_length = stage_token && stage_token[0] + ? snprintf(suffix, sizeof(suffix), ".stage.%s.XXXXXX", stage_token) + : snprintf(suffix, sizeof(suffix), ".stage.XXXXXX"); + if (suffix_length <= 0 || (size_t)suffix_length >= sizeof(suffix)) { return NULL; - static const char suffix[] = ".stage.XXXXXX"; + } size_t final_len = stage_root_length(final_path); - if (final_len > SIZE_MAX - sizeof(suffix)) + if (final_len > SIZE_MAX - (size_t)suffix_length - 1U) { return NULL; - size_t path_size = final_len + sizeof(suffix); + } + size_t path_size = final_len + (size_t)suffix_length + 1U; #ifdef _WIN32 - if (path_size > CBM_SZ_4K) + if (path_size > CBM_SZ_4K) { return NULL; + } path_size = CBM_SZ_4K; #endif char *path = malloc(path_size); if (path) { memcpy(path, final_path, final_len); - memcpy(path + final_len, suffix, sizeof(suffix)); + memcpy(path + final_len, suffix, (size_t)suffix_length + 1U); } return path; } @@ -4761,7 +5041,7 @@ static char *create_staging_path_impl(const char *final_path, cbm_pipeline_t *fr char *path = stage_allocate_template(final_path); if (!path) return NULL; - char *random_at = path + stage_root_length(final_path) + sizeof(cbm_stage_marker) - 1; + char *random_at = path + strlen(path) - CBM_STAGE_SUFFIX_RANDOM_CHARS; for (int attempt = 0; attempt < 128; attempt++) { int rc = stage_create_attempt(path, random_at, frozen); if (rc > 0) @@ -4927,6 +5207,34 @@ static size_t stage_entry_stage_length_any_base(const char *name) { return stage_entry_stage_length(name, name, (size_t)(marker - name)); } +static const char *stage_name_after_marker(const char *after_marker) { + for (int i = 0; i < CBM_STAGE_SUFFIX_RANDOM_CHARS; i++) { + if (!isalnum((unsigned char)after_marker[i])) { + const char *dot = strchr(after_marker, '.'); + if (!dot || dot == after_marker) { + return NULL; + } + size_t token_len = (size_t)(dot - after_marker); + if (token_len < 6U || token_len >= CBM_SZ_64) { + return NULL; + } + for (size_t n = 0; n < token_len; n++) { + if (!stage_token_char((unsigned char)after_marker[n])) { + return NULL; + } + } + after_marker = dot + 1; + for (int j = 0; j < CBM_STAGE_SUFFIX_RANDOM_CHARS; j++) { + if (!isalnum((unsigned char)after_marker[j])) { + return NULL; + } + } + return after_marker + CBM_STAGE_SUFFIX_RANDOM_CHARS; + } + } + return after_marker + CBM_STAGE_SUFFIX_RANDOM_CHARS; +} + static size_t stage_entry_stage_length(const char *name, const char *base, size_t base_len) { if (strncmp(name, base, base_len) != 0) { return 0; @@ -4935,14 +5243,10 @@ static size_t stage_entry_stage_length(const char *name, const char *base, size_ if (strncmp(at, cbm_stage_marker, sizeof(cbm_stage_marker) - 1) != 0) { return 0; } - at += sizeof(cbm_stage_marker) - 1; - for (int i = 0; i < CBM_STAGE_SUFFIX_RANDOM_CHARS; i++) { - /* NUL is not alphanumeric, so a short name fails here too. */ - if (!isalnum((unsigned char)at[i])) { - return 0; - } + at = stage_name_after_marker(at + sizeof(cbm_stage_marker) - 1); + if (!at) { + return 0; } - at += CBM_STAGE_SUFFIX_RANDOM_CHARS; for (size_t i = 0; i < sizeof(cbm_stage_sidecar_tails) / sizeof(cbm_stage_sidecar_tails[0]); i++) { if (strcmp(at, cbm_stage_sidecar_tails[i]) == 0) { @@ -5144,10 +5448,15 @@ int cbm_pipeline_run(cbm_pipeline_t *p) { cbm_log_debug("pipeline.setup", "step", "stage"); pipeline_release_test_config(p); p->export_error[0] = '\0'; + p->resource_violation = (cbm_index_resource_violation_t){0}; char *final_path = resolve_db_path(p); - if (!final_path || !ensure_db_parent(final_path)) { + int storage_rc = + final_path && ensure_db_parent(final_path) + ? cbm_pipeline_storage_admit(p, CBM_PIPELINE_STORAGE_PREFLIGHT, final_path, NULL) + : CBM_NOT_FOUND; + if (storage_rc != 0) { free(final_path); - return CBM_NOT_FOUND; + return storage_rc; } struct stat final_st; bool final_existed = stat(final_path, &final_st) == 0; @@ -5192,6 +5501,9 @@ int cbm_pipeline_run(cbm_pipeline_t *p) { * abort is genuinely non-destructive -- a caller that cannot tell an * aborted run from a failed persist cannot tell whether its data survived. * The staging file is discarded on every one of these paths. */ + if (rc == 0) { + rc = cbm_pipeline_storage_admit(p, CBM_PIPELINE_STORAGE_GROWTH, final_path, staging_path); + } if (rc != 0) { cleanup_staging_db(staging_path); free(staging_path); @@ -5232,12 +5544,16 @@ int cbm_pipeline_run(cbm_pipeline_t *p) { /* A test hook may inspect the DB through SQLite and re-enable WAL mode; * seal once more before installing the standalone main file. */ - if (seal_staging_db(staging_path) != 0) { + int post_hook_rc = seal_staging_db(staging_path) == 0 + ? cbm_pipeline_storage_admit(p, CBM_PIPELINE_STORAGE_PREPUBLISH, + final_path, staging_path) + : CBM_PIPELINE_PERSIST_FAILED; + if (post_hook_rc != 0) { cleanup_staging_db(staging_path); free(staging_path); free(final_path); pipeline_release_test_config(p); - return CBM_PIPELINE_PERSIST_FAILED; + return post_hook_rc; } /* Check after the before-publish hook and the final seal, while the diff --git a/src/pipeline/pipeline.h b/src/pipeline/pipeline.h index 20717275bc..7b1b4722e2 100644 --- a/src/pipeline/pipeline.h +++ b/src/pipeline/pipeline.h @@ -107,6 +107,21 @@ int cbm_pipeline_worker_count_test_last(void); * cbm_pipeline_free(). Returns "" for NULL p. */ const char *cbm_pipeline_export_error(const cbm_pipeline_t *p); +typedef enum { + CBM_PIPELINE_STORAGE_PREFLIGHT = 0, + CBM_PIPELINE_STORAGE_GROWTH, + CBM_PIPELINE_STORAGE_PREPUBLISH, +} cbm_pipeline_storage_checkpoint_t; +#ifdef CBM_ENABLE_TEST_SEAMS +typedef bool (*cbm_pipeline_storage_probe_fn)(cbm_pipeline_storage_checkpoint_t checkpoint, + const char *final_db_path, + const char *staging_db_path, + cbm_index_storage_sample_t *sample, + cbm_index_resource_t *failed_resource, void *context); +void cbm_pipeline_set_storage_probe_for_testing(cbm_pipeline_t *p, + cbm_pipeline_storage_probe_fn probe, void *context); +#endif + /* Free a pipeline and all its internal state. NULL-safe. */ void cbm_pipeline_free(cbm_pipeline_t *p); diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index ca0f0ec358..72a4860422 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -1731,6 +1731,9 @@ typedef struct { bool doc_candidates_in_place; } cbm_pipeline_generation_t; +int cbm_pipeline_storage_admit(cbm_pipeline_t *p, cbm_pipeline_storage_checkpoint_t checkpoint, + const char *final_path, const char *staging_path); + /* Serialize and fully populate a sibling staging database, then atomically * replace final_db_path. The old generation is untouched on every failure or * cancellation observed before the rename commit point. */ diff --git a/tests/test_daemon_application.c b/tests/test_daemon_application.c index efbaee5905..f4397d4b8d 100644 --- a/tests/test_daemon_application.c +++ b/tests/test_daemon_application.c @@ -1730,6 +1730,7 @@ static int app_fake_worker_start(void *opaque, const char *args_json, size_t mem atomic_init(&worker->cancelled, false); worker->result.exit_code = -1; if (worker->attempt < APP_FAKE_MAX_ATTEMPTS) { + worker->result.resource_violation = context->resource_violations[worker->attempt]; (void)snprintf(context->args_json[worker->attempt], APP_FAKE_ARGS_CAP, "%s", args_json ? args_json : ""); atomic_store(&context->args_captured[worker->attempt], true); @@ -1763,8 +1764,8 @@ static bool app_wait_for_atomic_bool(atomic_bool *value, bool expected); static bool app_fake_worker_policy_equals(app_fake_worker_context_t *context, int attempt, const char *max_files, const char *max_source_mb, - const char *max_rss_mb, - const char *max_duration_seconds) { + const char *max_rss_mb, const char *max_duration_seconds, + const char *cache_max_mb, const char *min_free_disk_mb) { if (!context || attempt < 0 || attempt >= APP_FAKE_MAX_ATTEMPTS) { return false; } @@ -1787,12 +1788,21 @@ static bool app_fake_worker_policy_equals(app_fake_worker_context_t *context, in yyjson_val *duration = policy && yyjson_is_obj(policy) ? yyjson_obj_get(policy, CBM_INDEX_CONFIG_MAX_DURATION_SECONDS) : NULL; - bool equal = files && bytes && rss && duration && yyjson_is_str(files) && + yyjson_val *cache = policy && yyjson_is_obj(policy) + ? yyjson_obj_get(policy, CBM_INDEX_CONFIG_CACHE_MAX_MB) + : NULL; + yyjson_val *free_disk = policy && yyjson_is_obj(policy) + ? yyjson_obj_get(policy, CBM_INDEX_CONFIG_MIN_FREE_DISK_MB) + : NULL; + bool equal = files && bytes && rss && duration && cache && free_disk && yyjson_is_str(files) && yyjson_is_str(bytes) && yyjson_is_str(rss) && yyjson_is_str(duration) && + yyjson_is_str(cache) && yyjson_is_str(free_disk) && strcmp(yyjson_get_str(files), max_files) == 0 && strcmp(yyjson_get_str(bytes), max_source_mb) == 0 && strcmp(yyjson_get_str(rss), max_rss_mb) == 0 && - strcmp(yyjson_get_str(duration), max_duration_seconds) == 0; + strcmp(yyjson_get_str(duration), max_duration_seconds) == 0 && + strcmp(yyjson_get_str(cache), cache_max_mb) == 0 && + strcmp(yyjson_get_str(free_disk), min_free_disk_mb) == 0; yyjson_doc_free(document); return equal; } @@ -2510,7 +2520,7 @@ TEST(daemon_application_initialize_coalesces_auto_index_for_full_sessions) { app_wait_for_subscribers(application, project, 1) && app_wait_for_atomic_int(&fake.starts, 1); bool auto_policy_propagated = - first_owned && app_fake_worker_policy_equals(&fake, 0, "3", "4", "64", "7"); + first_owned && app_fake_worker_policy_equals(&fake, 0, "3", "4", "64", "7", "off", "off"); bool second_initialized = app_test_initialize_profile(&callbacks, sessions[1], root, CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL); bool coalesced = first_owned && second_initialized && project && @@ -2976,7 +2986,9 @@ TEST(daemon_application_programmatic_index_injects_resource_policy) { stored_config && cbm_config_set(stored_config, CBM_INDEX_CONFIG_MAX_FILES, "5") == 0 && cbm_config_set(stored_config, CBM_INDEX_CONFIG_MAX_SOURCE_MB, "6") == 0 && cbm_config_set(stored_config, CBM_INDEX_CONFIG_MAX_RSS_MB, "64") == 0 && - cbm_config_set(stored_config, CBM_INDEX_CONFIG_MAX_DURATION_SECONDS, "7") == 0; + cbm_config_set(stored_config, CBM_INDEX_CONFIG_MAX_DURATION_SECONDS, "7") == 0 && + cbm_config_set(stored_config, CBM_INDEX_CONFIG_CACHE_MAX_MB, "8") == 0 && + cbm_config_set(stored_config, CBM_INDEX_CONFIG_MIN_FREE_DISK_MB, "9") == 0; app_fake_worker_context_t fake; app_fake_worker_context_init(&fake); @@ -2997,7 +3009,7 @@ TEST(daemon_application_programmatic_index_injects_resource_policy) { int index_rc = application ? cbm_daemon_application_index(application, "policy-programmatic", root) : -1; bool policy_propagated = - index_rc == 0 && app_fake_worker_policy_equals(&fake, 0, "5", "6", "64", "7"); + index_rc == 0 && app_fake_worker_policy_equals(&fake, 0, "5", "6", "64", "7", "8", "9"); bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS); cbm_daemon_application_free(application); @@ -3163,6 +3175,40 @@ TEST(daemon_application_worker_resource_failure_is_structured_and_not_retried) { PASS(); } +TEST(daemon_application_worker_storage_resource_failure_is_structured_and_not_retried) { + app_watch_race_fixture_t fixture; + bool ready = app_watch_race_fixture_init(&fixture, 77); + atomic_store(&fixture.fake.scripted, true); + fixture.fake.outcomes[0] = CBM_PROC_KILLED; + fixture.fake.resource_violations[0] = (cbm_index_resource_violation_t){ + .resource = CBM_INDEX_RESOURCE_TASK_TEMP_BYTES, .observed = 2048, .limit = 1024}; + + char args[APP_TEST_PATH_CAP + 32]; + (void)snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", fixture.root); + uint8_t *request = NULL; + uint32_t request_length = 0; + uint8_t *response = NULL; + uint32_t response_length = 0; + bool encoded = + ready && app_test_tool_request("index_repository", args, &request, &request_length); + bool completed = encoded && app_test_request(&fixture.callbacks, fixture.session, request, + request_length, &response, &response_length) == + CBM_DAEMON_RUNTIME_APPLICATION_OK; + bool structured = + completed && response && strstr((char *)response, "resource_limit_exceeded") && + strstr((char *)response, "task_temp_bytes") && strstr((char *)response, "storage"); + int starts = atomic_load(&fixture.fake.starts); + free(request); + free(response); + bool cleaned = app_watch_race_fixture_finish(&fixture); + + ASSERT_TRUE(ready); + ASSERT_TRUE(completed); + ASSERT_TRUE(structured); + ASSERT_EQ(starts, 1); + ASSERT_TRUE(cleaned); + PASS(); +} /* #1818: the daemon.autoindex.skipped warning must name the effective limit, * like the MCP autoindex.skip event does (#1466). The warning is emitted on the @@ -8232,6 +8278,7 @@ SUITE(daemon_application) { RUN_TEST(daemon_application_programmatic_index_injects_resource_policy); RUN_TEST(daemon_application_programmatic_index_forwards_set_doc_candidates); RUN_TEST(daemon_application_worker_resource_failure_is_structured_and_not_retried); + RUN_TEST(daemon_application_worker_storage_resource_failure_is_structured_and_not_retried); RUN_TEST(daemon_application_auto_index_honors_tracked_file_limit); RUN_TEST(daemon_application_auto_index_file_count_handles_literal_metacharacter_path); RUN_TEST(daemon_application_auto_index_file_count_supports_non_git_roots); diff --git a/tests/test_index_policy.c b/tests/test_index_policy.c index b33d64cdb0..dc362f6b5b 100644 --- a/tests/test_index_policy.c +++ b/tests/test_index_policy.c @@ -3,6 +3,7 @@ #include "cli/cli.h" #include "foundation/compat.h" +#include "foundation/constants.h" #include "foundation/index_policy.h" #include "mcp/index_supervisor.h" #include "mcp/mcp.h" @@ -23,13 +24,21 @@ TEST(index_policy_defaults_are_disabled) { ASSERT_FALSE(policy.max_source_bytes.enabled); ASSERT_FALSE(policy.max_rss_bytes.enabled); ASSERT_FALSE(policy.max_duration_ms.enabled); + ASSERT_FALSE(policy.max_cache_bytes.enabled); + ASSERT_FALSE(policy.min_free_disk_bytes.enabled); + ASSERT_FALSE(policy.max_final_db_bytes.enabled); + ASSERT_FALSE(policy.max_staging_bytes.enabled); + ASSERT_FALSE(policy.max_task_temp_bytes.enabled); ASSERT_FALSE(cbm_index_policy_enabled(&policy)); ASSERT_FALSE(cbm_index_policy_discovery_enabled(&policy)); ASSERT_FALSE(cbm_index_policy_worker_enabled(&policy)); + ASSERT_FALSE(cbm_index_policy_storage_enabled(&policy)); ASSERT_STR_EQ(cbm_index_policy_default_value(CBM_INDEX_CONFIG_MAX_FILES), "off"); ASSERT_STR_EQ(cbm_index_policy_default_value(CBM_INDEX_CONFIG_MAX_SOURCE_MB), "off"); ASSERT_STR_EQ(cbm_index_policy_default_value(CBM_INDEX_CONFIG_MAX_RSS_MB), "off"); ASSERT_STR_EQ(cbm_index_policy_default_value(CBM_INDEX_CONFIG_MAX_DURATION_SECONDS), "off"); + ASSERT_STR_EQ(cbm_index_policy_default_value(CBM_INDEX_CONFIG_CACHE_MAX_MB), "off"); + ASSERT_STR_EQ(cbm_index_policy_default_value(CBM_INDEX_CONFIG_MIN_FREE_DISK_MB), "off"); PASS(); } @@ -98,6 +107,70 @@ TEST(index_policy_worker_limits_validate_and_convert_units) { PASS(); } +TEST(index_policy_storage_limits_validate_and_convert_units) { + cbm_index_resource_policy_t policy; + cbm_index_policy_init(&policy); + char error[256]; + char value[64]; + + ASSERT_TRUE(cbm_index_policy_set(&policy, CBM_INDEX_CONFIG_CACHE_MAX_MB, "131072", error, + sizeof(error))); + ASSERT_TRUE(cbm_index_policy_set(&policy, CBM_INDEX_CONFIG_MIN_FREE_DISK_MB, "4096", error, + sizeof(error))); + ASSERT_EQ(policy.max_cache_bytes.value, UINT64_C(131072) * CBM_INDEX_MIB_BYTES); + ASSERT_EQ(policy.min_free_disk_bytes.value, UINT64_C(4096) * CBM_INDEX_MIB_BYTES); + ASSERT_TRUE(cbm_index_policy_storage_enabled(&policy)); + ASSERT_TRUE( + cbm_index_policy_format(&policy, CBM_INDEX_CONFIG_CACHE_MAX_MB, value, sizeof(value))); + ASSERT_STR_EQ(value, "131072"); + ASSERT_TRUE( + cbm_index_policy_set(&policy, CBM_INDEX_CONFIG_CACHE_MAX_MB, "off", error, sizeof(error))); + ASSERT_FALSE(policy.max_cache_bytes.enabled); + PASS(); +} + +TEST(index_policy_storage_projection_boundaries_are_overflow_safe) { + cbm_index_resource_policy_t policy; + cbm_index_policy_init(&policy); + policy.max_cache_bytes = (cbm_index_limit_u64_t){.enabled = true, .value = 100}; + policy.min_free_disk_bytes = (cbm_index_limit_u64_t){.enabled = true, .value = 40}; + policy.max_final_db_bytes = (cbm_index_limit_u64_t){.enabled = true, .value = 60}; + policy.max_staging_bytes = (cbm_index_limit_u64_t){.enabled = true, .value = 70}; + policy.max_task_temp_bytes = (cbm_index_limit_u64_t){.enabled = true, .value = 80}; + cbm_index_storage_sample_t sample = { + .current_cache_bytes = 130, + .replaceable_old_bytes = 50, + .operation_bytes = 20, + .free_disk_bytes = 40, + .final_db_bytes = 60, + .staging_bytes = 70, + .task_temp_bytes = 80, + }; + cbm_index_resource_violation_t violation = {0}; + + ASSERT_TRUE(cbm_index_policy_check_storage(&policy, &sample, &violation)); + sample.current_cache_bytes++; + ASSERT_FALSE(cbm_index_policy_check_storage(&policy, &sample, &violation)); + ASSERT_EQ(violation.resource, CBM_INDEX_RESOURCE_CACHE_BYTES); + ASSERT_EQ(violation.observed, 101); + ASSERT_EQ(violation.limit, 100); + sample.current_cache_bytes--; + sample.free_disk_bytes--; + ASSERT_FALSE(cbm_index_policy_check_storage(&policy, &sample, &violation)); + ASSERT_EQ(violation.resource, CBM_INDEX_RESOURCE_FREE_DISK_BYTES); + ASSERT_EQ(violation.observed, 39); + sample = (cbm_index_storage_sample_t){ + .current_cache_bytes = UINT64_MAX, + .replaceable_old_bytes = 0, + .operation_bytes = UINT64_MAX, + .free_disk_bytes = UINT64_MAX, + }; + ASSERT_FALSE(cbm_index_policy_check_storage(&policy, &sample, &violation)); + ASSERT_EQ(violation.resource, CBM_INDEX_RESOURCE_CACHE_BYTES); + ASSERT_EQ(violation.observed, UINT64_MAX); + PASS(); +} + TEST(index_policy_invalid_value_is_rejected_atomically) { static const char *const invalid[] = {"", "0", "-1", "1MB", "1 ", "+1", "10000001", "18446744073709551616"}; @@ -149,6 +222,9 @@ TEST(index_policy_format_round_trips_public_values) { TEST(index_policy_violation_metadata_is_stable) { ASSERT_STR_EQ(cbm_index_resource_name(CBM_INDEX_RESOURCE_FILES), "files"); ASSERT_STR_EQ(cbm_index_resource_name(CBM_INDEX_RESOURCE_SOURCE_BYTES), "source_bytes"); + ASSERT_STR_EQ(cbm_index_resource_name(CBM_INDEX_RESOURCE_CACHE_BYTES), "cache_bytes"); + ASSERT_STR_EQ(cbm_index_resource_name(CBM_INDEX_RESOURCE_FREE_DISK_BYTES), "free_disk_bytes"); + ASSERT_STR_EQ(cbm_index_resource_name(CBM_INDEX_RESOURCE_STAGING_BYTES), "staging_bytes"); ASSERT_STR_EQ(cbm_index_resource_unit(CBM_INDEX_RESOURCE_FILES), "files"); ASSERT_STR_EQ(cbm_index_resource_unit(CBM_INDEX_RESOURCE_SOURCE_BYTES), "bytes"); ASSERT_STR_EQ(cbm_index_resource_name(CBM_INDEX_RESOURCE_RSS_BYTES), "rss_bytes"); @@ -163,6 +239,10 @@ TEST(index_policy_violation_metadata_is_stable) { CBM_INDEX_CONFIG_MAX_RSS_MB); ASSERT_STR_EQ(cbm_index_resource_config_key(CBM_INDEX_RESOURCE_DURATION_MS), CBM_INDEX_CONFIG_MAX_DURATION_SECONDS); + ASSERT_STR_EQ(cbm_index_resource_config_key(CBM_INDEX_RESOURCE_CACHE_BYTES), + CBM_INDEX_CONFIG_CACHE_MAX_MB); + ASSERT_STR_EQ(cbm_index_resource_config_key(CBM_INDEX_RESOURCE_FREE_DISK_BYTES), + CBM_INDEX_CONFIG_MIN_FREE_DISK_MB); PASS(); } @@ -180,11 +260,15 @@ TEST(index_policy_config_loads_defaults_values_and_rejects_corruption) { ASSERT_EQ(cbm_config_set(config, CBM_INDEX_CONFIG_MAX_SOURCE_MB, "3"), 0); ASSERT_EQ(cbm_config_set(config, CBM_INDEX_CONFIG_MAX_RSS_MB, "64"), 0); ASSERT_EQ(cbm_config_set(config, CBM_INDEX_CONFIG_MAX_DURATION_SECONDS, "7"), 0); + ASSERT_EQ(cbm_config_set(config, CBM_INDEX_CONFIG_CACHE_MAX_MB, "17"), 0); + ASSERT_EQ(cbm_config_set(config, CBM_INDEX_CONFIG_MIN_FREE_DISK_MB, "5"), 0); ASSERT_TRUE(cbm_config_load_index_policy(config, &policy, error, sizeof(error))); ASSERT_EQ(policy.max_files.value, 9); ASSERT_EQ(policy.max_source_bytes.value, UINT64_C(3) * CBM_INDEX_MIB_BYTES); ASSERT_EQ(policy.max_rss_bytes.value, UINT64_C(64) * CBM_INDEX_MIB_BYTES); ASSERT_EQ(policy.max_duration_ms.value, UINT64_C(7000)); + ASSERT_EQ(policy.max_cache_bytes.value, UINT64_C(17) * CBM_INDEX_MIB_BYTES); + ASSERT_EQ(policy.min_free_disk_bytes.value, UINT64_C(5) * CBM_INDEX_MIB_BYTES); ASSERT_EQ(cbm_config_set(config, CBM_INDEX_CONFIG_MAX_FILES, "corrupt"), 0); ASSERT_FALSE(cbm_config_load_index_policy(config, &policy, error, sizeof(error))); @@ -200,6 +284,8 @@ TEST(index_policy_cli_lists_all_operator_keys) { bool bytes_found = false; bool rss_found = false; bool duration_found = false; + bool cache_found = false; + bool free_found = false; for (size_t index = 0; index < cbm_cli_config_key_count_for_testing(); index++) { const char *key = cbm_cli_config_key_at_for_testing(index); files_found = files_found || (key && strcmp(key, CBM_INDEX_CONFIG_MAX_FILES) == 0); @@ -207,11 +293,15 @@ TEST(index_policy_cli_lists_all_operator_keys) { rss_found = rss_found || (key && strcmp(key, CBM_INDEX_CONFIG_MAX_RSS_MB) == 0); duration_found = duration_found || (key && strcmp(key, CBM_INDEX_CONFIG_MAX_DURATION_SECONDS) == 0); + cache_found = cache_found || (key && strcmp(key, CBM_INDEX_CONFIG_CACHE_MAX_MB) == 0); + free_found = free_found || (key && strcmp(key, CBM_INDEX_CONFIG_MIN_FREE_DISK_MB) == 0); } ASSERT_TRUE(files_found); ASSERT_TRUE(bytes_found); ASSERT_TRUE(rss_found); ASSERT_TRUE(duration_found); + ASSERT_TRUE(cache_found); + ASSERT_TRUE(free_found); PASS(); } @@ -454,11 +544,92 @@ TEST(index_policy_mcp_rejects_forged_override_and_preserves_serving_index) { PASS(); } +TEST(index_policy_storage_limit_preserves_old_index_and_unrelated_cache_files) { + char *repo = th_mktempdir("cbm_index_storage_repo"); + char *cache = th_mktempdir("cbm_index_storage_cache"); + ASSERT_NOT_NULL(repo); + ASSERT_NOT_NULL(cache); + ASSERT_EQ(th_write_file(TH_PATH(repo, "main.c"), "int original(void) { return 1; }\n"), 0); + const char *saved_cache = getenv("CBM_CACHE_DIR"); + char *saved_cache_copy = saved_cache ? strdup(saved_cache) : NULL; + (void)cbm_setenv("CBM_CACHE_DIR", cache, 1); + cbm_config_t *config = cbm_config_open(cache); + cbm_mcp_server_t *server = cbm_mcp_server_new(NULL); + if (server && config) { + cbm_mcp_server_set_config(server, config); + } + char args[CBM_SZ_4K]; + (void)snprintf(args, sizeof(args), + "{\"repo_path\":\"%s\",\"name\":\"StoragePolicyFixture\",\"mode\":\"fast\"}", + repo); + char *first = server ? cbm_mcp_handle_tool(server, "index_repository", args) : NULL; + bool first_ok = first && strstr(first, "\\\"status\\\":\\\"indexed\\\""); + free(first); + + char unrelated[CBM_SZ_4K]; + (void)snprintf(unrelated, sizeof(unrelated), "%s/unrelated.bin", cache); + FILE *large = cbm_fopen(unrelated, "wb"); + char block[4096] = {0}; + bool wrote_large = large != NULL; + for (int index = 0; wrote_large && index < 512; index++) { + wrote_large = fwrite(block, 1, sizeof(block), large) == sizeof(block); + } + if (large) { + wrote_large = fclose(large) == 0 && wrote_large; + } + bool configured = + config && cbm_config_set(config, CBM_INDEX_CONFIG_CACHE_MAX_MB, "1") == 0 && + th_write_file(TH_PATH(repo, "main.c"), "int changed(void) { return 2; }\n") == 0; + (void)snprintf(args, sizeof(args), + "{\"repo_path\":\"%s\",\"name\":\"StoragePolicyFixture\",\"mode\":\"fast\"," + "\"_cbm_index_policy\":{\"index_max_files\":\"off\"," + "\"index_max_source_mb\":\"off\",\"index_cache_max_mb\":\"off\"," + "\"index_min_free_disk_mb\":\"off\"}}", + repo); + char *limited = configured && wrote_large && server + ? cbm_mcp_handle_tool(server, "index_repository", args) + : NULL; + bool contract_ok = limited && strstr(limited, "resource_limit_exceeded") && + strstr(limited, "\\\"stage\\\":\\\"storage\\\"") && + strstr(limited, "\\\"resource\\\":\\\"cache_bytes\\\"") && + strstr(limited, "\\\"limit\\\":1048576") && + strstr(limited, "\\\"unit\\\":\\\"bytes\\\"") && + strstr(limited, "\\\"serving_index_preserved\\\":true"); + free(limited); + int64_t unrelated_size = cbm_file_size(unrelated); + char db_path[CBM_SZ_4K]; + (void)snprintf(db_path, sizeof(db_path), "%s/StoragePolicyFixture.db", cache); + cbm_store_t *store = cbm_store_open_path_query(db_path); + bool old_queryable = store && cbm_store_count_nodes(store, "StoragePolicyFixture") > 0; + cbm_store_close(store); + + cbm_mcp_server_free(server); + cbm_config_close(config); + th_cleanup(repo); + th_cleanup(cache); + if (saved_cache_copy) { + (void)cbm_setenv("CBM_CACHE_DIR", saved_cache_copy, 1); + } else { + (void)cbm_unsetenv("CBM_CACHE_DIR"); + } + free(saved_cache_copy); + + ASSERT_TRUE(first_ok); + ASSERT_TRUE(wrote_large); + ASSERT_TRUE(configured); + ASSERT_TRUE(contract_ok); + ASSERT_EQ(unrelated_size, (int64_t)sizeof(block) * 512); + ASSERT_TRUE(old_queryable); + PASS(); +} + SUITE(index_policy) { RUN_TEST(index_policy_defaults_are_disabled); RUN_TEST(index_policy_file_limit_accepts_off_and_exact_range); RUN_TEST(index_policy_source_limit_converts_mib_without_overflow); RUN_TEST(index_policy_worker_limits_validate_and_convert_units); + RUN_TEST(index_policy_storage_limits_validate_and_convert_units); + RUN_TEST(index_policy_storage_projection_boundaries_are_overflow_safe); RUN_TEST(index_policy_invalid_value_is_rejected_atomically); RUN_TEST(index_policy_format_round_trips_public_values); RUN_TEST(index_policy_violation_metadata_is_stable); @@ -468,4 +639,5 @@ SUITE(index_policy) { RUN_TEST(index_policy_cli_set_reports_a_failed_write); RUN_TEST(index_policy_worker_rejects_missing_parent_policy); RUN_TEST(index_policy_mcp_rejects_forged_override_and_preserves_serving_index); + RUN_TEST(index_policy_storage_limit_preserves_old_index_and_unrelated_cache_files); } diff --git a/tests/test_index_supervisor.c b/tests/test_index_supervisor.c index f3218fdb14..50d4ccd9c8 100644 --- a/tests/test_index_supervisor.c +++ b/tests/test_index_supervisor.c @@ -124,6 +124,24 @@ static bool index_supervisor_test_append_log(const char *path, const char *line) return fclose(file) == 0 && written; } +typedef struct { + uint64_t bytes; + bool succeeds; + int calls; +} index_supervisor_task_temp_fake_t; + +static bool index_supervisor_task_temp_fake(const char *final_db_path, const char *log_path, + const char *response_path, uint64_t *bytes_out, + void *context) { + index_supervisor_task_temp_fake_t *fake = context; + fake->calls++; + if (!final_db_path || !log_path || !response_path || !fake->succeeds) { + return false; + } + *bytes_out = fake->bytes; + return true; +} + static bool index_supervisor_test_append_terminal_backlog(const char *path) { FILE *file = cbm_fopen(path, "ab"); if (!file) { @@ -241,12 +259,14 @@ TEST(index_supervisor_worker_argv_requires_exact_build_bound_grammar) { "/tmp/q", CBM_INDEX_WORKER_PUBLISH_FENCE_ARG, "/tmp/f", + CBM_INDEX_WORKER_STAGE_TOKEN_ARG, + "task123", NULL, }; cbm_index_worker_invocation_t invocation; - ASSERT_EQ(cbm_index_worker_parse_process_argv(18, valid, &invocation), + ASSERT_EQ(cbm_index_worker_parse_process_argv(20, valid, &invocation), CBM_INDEX_WORKER_ARGV_VALID); - ASSERT_EQ(cbm_daemon_process_role(18, valid), CBM_DAEMON_PROCESS_WORKER); + ASSERT_EQ(cbm_daemon_process_role(20, valid), CBM_DAEMON_PROCESS_WORKER); ASSERT_STR_EQ(invocation.args_json, valid[6]); ASSERT_STR_EQ(invocation.response_out, valid[8]); ASSERT_EQ(invocation.memory_budget_bytes, 1024); @@ -254,11 +274,13 @@ TEST(index_supervisor_worker_argv_requires_exact_build_bound_grammar) { ASSERT_STR_EQ(invocation.marker_file, valid[13]); ASSERT_STR_EQ(invocation.quarantine_file, valid[15]); ASSERT_STR_EQ(invocation.publish_fence_file, valid[17]); + ASSERT_STR_EQ(invocation.stage_token, valid[19]); /* Without a worker limit there is no fence: the shorter grammar stays valid. */ ASSERT_EQ(cbm_index_worker_parse_process_argv(16, valid, &invocation), CBM_INDEX_WORKER_ARGV_VALID); ASSERT_EQ(cbm_daemon_process_role(16, valid), CBM_DAEMON_PROCESS_WORKER); ASSERT_NULL(invocation.publish_fence_file); + ASSERT_NULL(invocation.stage_token); char *fence_before_quarantine[] = {"test-runner", "cli", @@ -355,6 +377,18 @@ TEST(index_supervisor_worker_argv_requires_exact_build_bound_grammar) { CBM_INDEX_WORKER_MEMORY_BUDGET_ARG, "184467440737095516160", NULL}; + char *invalid_stage_token[] = {"test-runner", + "cli", + "--index-worker", + CBM_INDEX_WORKER_BUILD_ARG, + (char *)captured, + "index_repository", + "{}", + "--response-out", + "/tmp/r", + CBM_INDEX_WORKER_STAGE_TOKEN_ARG, + "../escape", + NULL}; char *user_value[] = {"test-runner", "cli", "search_code", "--query", "--index-worker", NULL}; ASSERT_EQ(cbm_index_worker_parse_process_argv(7, missing_build, &invocation), CBM_INDEX_WORKER_ARGV_INVALID); @@ -371,6 +405,9 @@ TEST(index_supervisor_worker_argv_requires_exact_build_bound_grammar) { ASSERT_EQ(cbm_index_worker_parse_process_argv(11, overflow_budget, &invocation), CBM_INDEX_WORKER_ARGV_INVALID); ASSERT_EQ(cbm_daemon_process_role(11, overflow_budget), CBM_DAEMON_PROCESS_INVALID); + ASSERT_EQ(cbm_index_worker_parse_process_argv(11, invalid_stage_token, &invocation), + CBM_INDEX_WORKER_ARGV_INVALID); + ASSERT_EQ(cbm_daemon_process_role(11, invalid_stage_token), CBM_DAEMON_PROCESS_INVALID); ASSERT_EQ(cbm_index_worker_parse_process_argv(5, user_value, &invocation), CBM_INDEX_WORKER_ARGV_INVALID); ASSERT_EQ(cbm_daemon_process_role(5, user_value), CBM_DAEMON_PROCESS_INVALID); @@ -1571,6 +1608,42 @@ TEST(index_supervisor_limit_after_worker_publication_claim_is_not_enforced) { PASS(); } +TEST(index_supervisor_task_temp_limit_terminates_as_resource_failure) { + char *root = th_mktempdir("cbm_supervisor_task_temp"); + ASSERT_NOT_NULL(root); + char db_path[INDEX_SUPERVISOR_TEST_PATH_CAP]; + (void)snprintf(db_path, sizeof(db_path), "%s/index.db", root); + cbm_index_resource_policy_t policy; + cbm_index_policy_init(&policy); + policy.max_task_temp_bytes = (cbm_index_limit_u64_t){.enabled = true, .value = 1}; + index_supervisor_task_temp_fake_t fake = {.bytes = 2, .succeeds = true}; + cbm_index_supervisor_set_task_temp_hook_for_testing(index_supervisor_task_temp_fake, &fake); + cbm_index_worker_handle_t *handle = NULL; + int start_rc = cbm_index_worker_start_with_storage_policy("{\"__cbm_test_worker\":\"clean\"}", + 0, &policy, db_path, false, NULL, + NULL, NULL, NULL, 0, &handle); + const cbm_index_worker_result_t *result = NULL; + bool terminal = handle && index_supervisor_test_poll_terminal( + handle, INDEX_SUPERVISOR_TEST_TERMINAL_MS, &result); + bool cancelled = result ? result->cancellation_requested : true; + cbm_index_resource_violation_t violation = + result ? result->resource_violation : (cbm_index_resource_violation_t){0}; + cbm_index_supervisor_reset_task_temp_hook_for_testing(); + if (handle) { + cbm_index_worker_destroy(handle); + } + th_cleanup(root); + + ASSERT_EQ(start_rc, 0); + ASSERT_TRUE(terminal); + ASSERT_FALSE(cancelled); + ASSERT_EQ(violation.resource, CBM_INDEX_RESOURCE_TASK_TEMP_BYTES); + ASSERT_EQ(violation.observed, 2); + ASSERT_EQ(violation.limit, 1); + ASSERT_TRUE(fake.calls > 0); + PASS(); +} + SUITE(index_supervisor) { RUN_TEST(index_supervisor_job_memory_limit_has_floor_headroom_and_no_overflow); RUN_TEST(index_supervisor_worker_argv_requires_exact_build_bound_grammar); @@ -1591,4 +1664,5 @@ SUITE(index_supervisor) { RUN_TEST(index_supervisor_three_failed_rss_probes_fail_closed); RUN_TEST(index_supervisor_limit_claims_publication_before_terminating); RUN_TEST(index_supervisor_limit_after_worker_publication_claim_is_not_enforced); + RUN_TEST(index_supervisor_task_temp_limit_terminates_as_resource_failure); } diff --git a/tests/test_main.c b/tests/test_main.c index e9ca43350f..44a0add005 100644 --- a/tests/test_main.c +++ b/tests/test_main.c @@ -510,7 +510,7 @@ static int tf_maybe_run_index_worker(int argc, char **argv) { free(worker_repo_path); cbm_index_set_worker_role_options(true, invocation.response_out, invocation.single_thread, invocation.marker_file, invocation.quarantine_file, - invocation.memory_budget_bytes); + invocation.memory_budget_bytes, invocation.stage_token); cbm_index_set_worker_publish_fence(invocation.publish_fence_file); cbm_mem_init_with_cap(0.5, invocation.memory_budget_bytes); cbm_log_init_for_process(false, true); diff --git a/tests/test_mcp.c b/tests/test_mcp.c index a75f45772c..5b76c7e745 100644 --- a/tests/test_mcp.c +++ b/tests/test_mcp.c @@ -17692,8 +17692,8 @@ static int doccand_rows(const char *cache) { sqlite3_finalize(st); st = NULL; n = 0; - if (has_table && sqlite3_prepare_v2(db, "SELECT count(*) FROM doc_link_candidates", -1, - &st, NULL) == SQLITE_OK) { + if (has_table && sqlite3_prepare_v2(db, "SELECT count(*) FROM doc_link_candidates", -1, &st, + NULL) == SQLITE_OK) { n = sqlite3_step(st) == SQLITE_ROW ? sqlite3_column_int(st, 0) : -1; sqlite3_finalize(st); } @@ -23864,6 +23864,17 @@ TEST(index_supervisor_unsafe_clean_is_never_fallback_or_recovery) { CBM_MCP_SUPERVISED_RESULT_UNSAFE_TERMINAL); result.supervision_failed = false; + result.tree_quiesced = true; + result.outcome = CBM_PROC_KILLED; + result.resource_violation = (cbm_index_resource_violation_t){ + .resource = CBM_INDEX_RESOURCE_TASK_TEMP_BYTES, .observed = 2048, .limit = 1024}; + ASSERT_EQ(cbm_mcp_supervised_result_disposition(0, &result), + CBM_MCP_SUPERVISED_RESULT_RESOURCE_FAILURE); + result.tree_quiesced = false; + ASSERT_EQ(cbm_mcp_supervised_result_disposition(0, &result), + CBM_MCP_SUPERVISED_RESULT_UNSAFE_TERMINAL); + result.resource_violation = (cbm_index_resource_violation_t){0}; + result.tree_quiesced = true; result.outcome = CBM_PROC_CRASH; result.response = NULL; ASSERT_EQ(cbm_mcp_supervised_result_disposition(0, &result), @@ -24292,6 +24303,26 @@ TEST(index_worker_resource_response_reports_preservation_for_the_request) { PASS(); } +TEST(index_supervisor_resource_probe_failure_response_is_structured) { + cbm_index_worker_result_t worker_result = { + .resource_violation = + { + .resource = CBM_INDEX_RESOURCE_CACHE_BYTES, + .limit = 1024, + .probe_failed = true, + }, + }; + char *response = cbm_mcp_index_worker_resource_response("{\"repo_path\":\"/tmp/missing\"}", + &worker_result, NULL); + ASSERT_NOT_NULL(response); + ASSERT_NOT_NULL(strstr(response, "\"code\":\"resource_probe_failed\"")); + ASSERT_NOT_NULL(strstr(response, "\"stage\":\"storage\"")); + ASSERT_NOT_NULL(strstr(response, "\"resource\":\"cache_bytes\"")); + ASSERT_NOT_NULL(strstr(response, "\"retryable\":true")); + free(response); + PASS(); +} + /* Child-side check: index a tiny fixture and verify it ran IN-PROCESS. * Distinct exit codes so the parent can report the exact failure mode. */ enum { @@ -25746,6 +25777,57 @@ TEST(mcp_ignore_worktrees_default_off_auto_index_reaches_count) { } #endif /* CBM_ENABLE_TEST_SEAMS */ +TEST(mcp_inprocess_autoindex_inherits_resource_policy) { + char *root = th_mktempdir("cbm-autoindex-policy-root"); + char *cache = th_mktempdir("cbm-autoindex-policy-cache"); + ASSERT_NOT_NULL(root); + ASSERT_NOT_NULL(cache); + ASSERT_EQ(th_write_file(TH_PATH(root, "first.c"), "int first(void) { return 1; }\n"), 0); + ASSERT_EQ(th_write_file(TH_PATH(root, "second.c"), "int second(void) { return 2; }\n"), 0); + + const char *saved_cache = getenv("CBM_CACHE_DIR"); + char *saved_cache_copy = saved_cache ? strdup(saved_cache) : NULL; + char saved_cwd[CBM_SZ_4K]; + bool cwd_saved = cbm_getcwd(saved_cwd, sizeof(saved_cwd)) != NULL; + bool environment_ready = + cwd_saved && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0 && cbm_chdir(root) == 0; + cbm_config_t *config = environment_ready ? cbm_config_open(cache) : NULL; + bool configured = config && cbm_config_set(config, CBM_CONFIG_AUTO_INDEX, "true") == 0 && + cbm_config_set(config, CBM_CONFIG_AUTO_WATCH, "false") == 0 && + cbm_config_set(config, CBM_INDEX_CONFIG_MAX_FILES, "1") == 0; + char *project = configured ? cbm_project_name_from_path(root) : NULL; + char db_path[CBM_SZ_4K] = {0}; + if (project) { + (void)snprintf(db_path, sizeof(db_path), "%s/%s.db", cache, project); + } + + cbm_mcp_server_t *server = configured && project ? cbm_mcp_server_new(NULL) : NULL; + if (server) { + cbm_mcp_server_set_config(server, config); + char *response = cbm_mcp_server_handle( + server, "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{}}"); + free(response); + cbm_mcp_server_free(server); /* joins the in-process auto-index thread */ + } + bool limit_preserved_no_index = project && !cbm_file_exists(db_path); + + cbm_config_close(config); + if (cwd_saved) { + (void)cbm_chdir(saved_cwd); + } + restore_cache_dir(saved_cache_copy); + free(saved_cache_copy); + free(project); + th_cleanup(root); + th_cleanup(cache); + + ASSERT_TRUE(environment_ready); + ASSERT_TRUE(configured); + ASSERT_NOT_NULL(server); + ASSERT_TRUE(limit_preserved_no_index); + PASS(); +} + /* ══════════════════════════════════════════════════════════════════ * #1466 / #713 — the auto_index_limit guard * @@ -27822,8 +27904,8 @@ static bool scan_fixture_list(const scan_fixture_t *fx, const char *extra) { /* Run the scan; returns the records it wrote (caller frees), CRs kept. */ static char *scan_fixture_run(const scan_fixture_t *fx, const char *pattern, bool use_regex, size_t output_limit, int *cause) { - *cause = cbm_mcp_search_scan_for_test(NULL, fx->list, pattern, use_regex, fx->out, - output_limit, UINT64_MAX); + *cause = cbm_mcp_search_scan_for_test(NULL, fx->list, pattern, use_regex, fx->out, output_limit, + UINT64_MAX); FILE *fp = cbm_fopen(fx->out, "rb"); if (!fp) { return NULL; @@ -28364,6 +28446,7 @@ SUITE(mcp) { RUN_TEST(index_repository_cli_name_override_issue823); RUN_TEST(index_repository_over_budget_reports_named_reason); RUN_TEST(index_supervisor_unsafe_clean_is_never_fallback_or_recovery); + RUN_TEST(index_supervisor_resource_probe_failure_response_is_structured); RUN_TEST(index_supervisor_gate_requires_marked_host_issue845); RUN_TEST(index_supervisor_start_failure_is_fail_closed_in_real_host); RUN_TEST(index_supervisor_clean_exit_without_response_response_issue1300); @@ -28468,6 +28551,7 @@ SUITE(mcp) { /* auto_watch gate (distilled from PR #625) */ RUN_TEST(mcp_auto_watch_default_registers_watcher_on_connect); RUN_TEST(mcp_auto_watch_false_skips_watcher_on_connect); + RUN_TEST(mcp_inprocess_autoindex_inherits_resource_policy); RUN_TEST(mcp_auto_watch_false_skips_supervised_autoindex_issue853); RUN_TEST(autoindex_skip_reports_numeric_limit_issue1466); RUN_TEST(autoindex_limit_guards_non_git_root_issue713); diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index e08a656bea..7b97baf612 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -449,6 +449,212 @@ TEST(pipeline_discovery_limit_returns_exact_violation_without_publishing) { PASS(); } +typedef struct { + const char *target_db; + cbm_pipeline_storage_checkpoint_t fail_at; + cbm_index_resource_t failed_resource; + cbm_index_storage_sample_t samples[3]; + bool sample_set[3]; + int calls[3]; + int unexpected_paths; +} pipeline_storage_probe_fake_t; + +static bool pipeline_storage_probe_fake(cbm_pipeline_storage_checkpoint_t checkpoint, + const char *final_db_path, const char *staging_db_path, + cbm_index_storage_sample_t *sample, + cbm_index_resource_t *failed_resource, void *context) { + (void)staging_db_path; + pipeline_storage_probe_fake_t *fake = (pipeline_storage_probe_fake_t *)context; + fake->calls[checkpoint]++; + if (fake->target_db && (!final_db_path || strcmp(fake->target_db, final_db_path) != 0)) { + fake->unexpected_paths++; + } + *sample = (cbm_index_storage_sample_t){ + .current_cache_bytes = 0, + .free_disk_bytes = UINT64_MAX, + }; + if (fake->sample_set[checkpoint]) { + *sample = fake->samples[checkpoint]; + } + if (fake->failed_resource != CBM_INDEX_RESOURCE_NONE && checkpoint == fake->fail_at && + fake->target_db && final_db_path && strcmp(fake->target_db, final_db_path) == 0) { + *failed_resource = fake->failed_resource; + return false; + } + return true; +} + +static int pipeline_stage_artifact_count(const char *directory) { + cbm_dir_t *dir = cbm_opendir(directory); + if (!dir) { + return -1; + } + int count = 0; + cbm_dirent_t *entry; + while ((entry = cbm_readdir(dir)) != NULL) { + if (strstr(entry->name, ".stage.")) { + count++; + } + } + cbm_closedir(dir); + return count; +} + +TEST(pipeline_disabled_storage_policy_performs_no_probes) { + char *repo = th_mktempdir("cbm_pipeline_storage_off"); + ASSERT_NOT_NULL(repo); + ASSERT_EQ(th_write_file(TH_PATH(repo, "main.c"), "int main(void) { return 0; }\n"), 0); + char db_path[CBM_SZ_4K]; + (void)snprintf(db_path, sizeof(db_path), "%s/index.db", repo); + cbm_pipeline_t *pipeline = cbm_pipeline_new(repo, db_path, CBM_MODE_FAST); + ASSERT_NOT_NULL(pipeline); + pipeline_storage_probe_fake_t fake = {.target_db = db_path, + .fail_at = CBM_PIPELINE_STORAGE_PREFLIGHT, + .failed_resource = CBM_INDEX_RESOURCE_FREE_DISK_BYTES}; + cbm_pipeline_set_storage_probe_for_testing(pipeline, pipeline_storage_probe_fake, &fake); + + ASSERT_EQ(cbm_pipeline_run(pipeline), 0); + ASSERT_EQ(fake.calls[CBM_PIPELINE_STORAGE_PREFLIGHT], 0); + ASSERT_EQ(fake.calls[CBM_PIPELINE_STORAGE_GROWTH], 0); + ASSERT_EQ(fake.calls[CBM_PIPELINE_STORAGE_PREPUBLISH], 0); + cbm_pipeline_free(pipeline); + th_cleanup(repo); + PASS(); +} + +TEST(pipeline_storage_probe_failure_preserves_old_db_and_cleans_stage) { + char *repo = th_mktempdir("cbm_pipeline_storage_probe"); + ASSERT_NOT_NULL(repo); + ASSERT_EQ(th_write_file(TH_PATH(repo, "main.c"), "int original(void) { return 1; }\n"), 0); + char db_path[CBM_SZ_4K]; + (void)snprintf(db_path, sizeof(db_path), "%s/index.db", repo); + cbm_pipeline_t *first = cbm_pipeline_new(repo, db_path, CBM_MODE_FAST); + ASSERT_NOT_NULL(first); + ASSERT_EQ(cbm_pipeline_run(first), 0); + const char *project = cbm_pipeline_project_name(first); + char *project_copy = project ? strdup(project) : NULL; + cbm_pipeline_free(first); + ASSERT_NOT_NULL(project_copy); + + cbm_store_t *before = cbm_store_open_path_query(db_path); + cbm_project_t before_project = {0}; + ASSERT_NOT_NULL(before); + ASSERT_EQ(cbm_store_get_project(before, project_copy, &before_project), CBM_STORE_OK); + char *indexed_at = before_project.indexed_at ? strdup(before_project.indexed_at) : NULL; + cbm_project_free_fields(&before_project); + cbm_store_close(before); + ASSERT_NOT_NULL(indexed_at); + ASSERT_EQ(th_write_file(TH_PATH(repo, "main.c"), "int changed(void) { return 2; }\n"), 0); + + cbm_pipeline_t *second = cbm_pipeline_new(repo, db_path, CBM_MODE_FAST); + ASSERT_NOT_NULL(second); + cbm_index_resource_policy_t policy; + cbm_index_policy_init(&policy); + policy.min_free_disk_bytes = (cbm_index_limit_u64_t){.enabled = true, .value = 1}; + cbm_pipeline_set_resource_policy(second, &policy); + pipeline_storage_probe_fake_t fake = {.target_db = db_path, + .fail_at = CBM_PIPELINE_STORAGE_PREPUBLISH, + .failed_resource = CBM_INDEX_RESOURCE_FREE_DISK_BYTES}; + cbm_pipeline_set_storage_probe_for_testing(second, pipeline_storage_probe_fake, &fake); + int rc = cbm_pipeline_run(second); + cbm_index_resource_violation_t violation = {0}; + cbm_pipeline_get_resource_violation(second, &violation); + cbm_pipeline_free(second); + + cbm_store_t *after = cbm_store_open_path_query(db_path); + cbm_project_t after_project = {0}; + bool preserved = after && + cbm_store_get_project(after, project_copy, &after_project) == CBM_STORE_OK && + after_project.indexed_at && strcmp(indexed_at, after_project.indexed_at) == 0; + cbm_project_free_fields(&after_project); + cbm_store_close(after); + int stage_count = pipeline_stage_artifact_count(repo); + free(indexed_at); + free(project_copy); + th_cleanup(repo); + + ASSERT_EQ(rc, CBM_PIPELINE_RESOURCE_LIMIT); + ASSERT_EQ(violation.resource, CBM_INDEX_RESOURCE_FREE_DISK_BYTES); + ASSERT_TRUE(violation.probe_failed); + ASSERT_TRUE(fake.calls[CBM_PIPELINE_STORAGE_PREPUBLISH] > 0); + ASSERT_TRUE(preserved); + ASSERT_EQ(stage_count, 0); + PASS(); +} + +TEST(pipeline_full_and_incremental_storage_limits_report_identical_failure) { + char *full_repo = th_mktempdir("cbm_pipeline_storage_full"); + char *incremental_repo = th_mktempdir("cbm_pipeline_storage_incremental"); + ASSERT_NOT_NULL(full_repo); + ASSERT_NOT_NULL(incremental_repo); + ASSERT_EQ(th_write_file(TH_PATH(full_repo, "main.c"), "int full(void) { return 1; }\n"), 0); + ASSERT_EQ( + th_write_file(TH_PATH(incremental_repo, "main.c"), "int incremental(void) { return 1; }\n"), + 0); + char full_db[CBM_SZ_4K]; + char incremental_db[CBM_SZ_4K]; + (void)snprintf(full_db, sizeof(full_db), "%s/index.db", full_repo); + (void)snprintf(incremental_db, sizeof(incremental_db), "%s/index.db", incremental_repo); + + cbm_pipeline_t *seed = cbm_pipeline_new(incremental_repo, incremental_db, CBM_MODE_FAST); + ASSERT_NOT_NULL(seed); + ASSERT_EQ(cbm_pipeline_run(seed), 0); + cbm_pipeline_free(seed); + ASSERT_EQ( + th_write_file(TH_PATH(incremental_repo, "main.c"), "int incremental(void) { return 2; }\n"), + 0); + + cbm_index_resource_policy_t policy; + cbm_index_policy_init(&policy); + policy.max_staging_bytes = (cbm_index_limit_u64_t){.enabled = true, .value = 100}; + pipeline_storage_probe_fake_t full_fake = {.target_db = full_db}; + full_fake.sample_set[CBM_PIPELINE_STORAGE_GROWTH] = true; + full_fake.samples[CBM_PIPELINE_STORAGE_GROWTH] = + (cbm_index_storage_sample_t){.free_disk_bytes = UINT64_MAX, .staging_bytes = 101}; + pipeline_storage_probe_fake_t incremental_fake = full_fake; + incremental_fake.target_db = incremental_db; + + cbm_pipeline_t *full = cbm_pipeline_new(full_repo, full_db, CBM_MODE_FAST); + cbm_pipeline_t *incremental = cbm_pipeline_new(incremental_repo, incremental_db, CBM_MODE_FAST); + ASSERT_NOT_NULL(full); + ASSERT_NOT_NULL(incremental); + cbm_pipeline_set_resource_policy(full, &policy); + cbm_pipeline_set_resource_policy(incremental, &policy); + cbm_pipeline_set_storage_probe_for_testing(full, pipeline_storage_probe_fake, &full_fake); + cbm_pipeline_set_storage_probe_for_testing(incremental, pipeline_storage_probe_fake, + &incremental_fake); + int full_rc = cbm_pipeline_run(full); + int incremental_rc = cbm_pipeline_run(incremental); + cbm_index_resource_violation_t full_violation = {0}; + cbm_index_resource_violation_t incremental_violation = {0}; + cbm_pipeline_get_resource_violation(full, &full_violation); + cbm_pipeline_get_resource_violation(incremental, &incremental_violation); + cbm_pipeline_free(full); + cbm_pipeline_free(incremental); + + ASSERT_EQ(full_rc, CBM_PIPELINE_RESOURCE_LIMIT); + ASSERT_EQ(incremental_rc, CBM_PIPELINE_RESOURCE_LIMIT); + ASSERT_EQ(full_violation.resource, CBM_INDEX_RESOURCE_STAGING_BYTES); + ASSERT_EQ(incremental_violation.resource, full_violation.resource); + ASSERT_EQ(incremental_violation.observed, full_violation.observed); + ASSERT_EQ(incremental_violation.limit, full_violation.limit); + ASSERT_STR_EQ(cbm_index_resource_unit(incremental_violation.resource), + cbm_index_resource_unit(full_violation.resource)); + ASSERT_EQ(full_fake.unexpected_paths, 0); + ASSERT_EQ(incremental_fake.unexpected_paths, 0); + ASSERT_EQ(full_fake.calls[CBM_PIPELINE_STORAGE_PREFLIGHT], 1); + ASSERT_EQ(full_fake.calls[CBM_PIPELINE_STORAGE_GROWTH], 1); + ASSERT_EQ(full_fake.calls[CBM_PIPELINE_STORAGE_PREPUBLISH], 0); + ASSERT_EQ(incremental_fake.calls[CBM_PIPELINE_STORAGE_PREFLIGHT], 1); + ASSERT_EQ(incremental_fake.calls[CBM_PIPELINE_STORAGE_GROWTH], 1); + ASSERT_EQ(incremental_fake.calls[CBM_PIPELINE_STORAGE_PREPUBLISH], 0); + ASSERT_EQ(pipeline_stage_artifact_count(full_repo), 0); + ASSERT_EQ(pipeline_stage_artifact_count(incremental_repo), 0); + th_cleanup(full_repo); + th_cleanup(incremental_repo); + PASS(); +} + /* ── Focused: file-backed store persistence ─────────────────────── */ TEST(store_file_persistence) { @@ -31898,6 +32104,9 @@ SUITE(pipeline) { RUN_TEST(pipeline_cancel_null); RUN_TEST(pipeline_run_null); RUN_TEST(pipeline_discovery_limit_returns_exact_violation_without_publishing); + RUN_TEST(pipeline_disabled_storage_policy_performs_no_probes); + RUN_TEST(pipeline_storage_probe_failure_preserves_old_db_and_cleans_stage); + RUN_TEST(pipeline_full_and_incremental_storage_limits_report_identical_failure); RUN_TEST(pipeline_late_source_limit_preserves_previous_generation); /* Extraction back-pressure */ RUN_TEST(pipeline_backpressure_futile_nap_disengages); diff --git a/tests/test_platform.c b/tests/test_platform.c index d83f90dab6..0bdef17799 100644 --- a/tests/test_platform.c +++ b/tests/test_platform.c @@ -2,6 +2,7 @@ * test_platform.c — RED phase tests for foundation/platform. */ #include "test_framework.h" +#include "test_helpers.h" #include "../src/foundation/compat.h" /* cbm_setenv / cbm_unsetenv (Windows-portable) */ #include "../src/foundation/compat_fs.h" #include "../src/foundation/compat_regex.h" @@ -690,6 +691,93 @@ TEST(platform_file_size) { PASS(); } +TEST(platform_storage_measurement_uses_logical_bytes_without_following_links) { + char *root = th_mktempdir("cbm_storage_measure"); + ASSERT_NOT_NULL(root); + ASSERT_TRUE(cbm_mkdir_p(TH_PATH(root, "nested"), 0700)); + ASSERT_EQ(th_write_file(TH_PATH(root, "first.bin"), "12345"), 0); + ASSERT_EQ(th_write_file(TH_PATH(root, "nested/second.bin"), "1234567"), 0); +#ifndef _WIN32 + ASSERT_EQ(symlink("/dev/null", TH_PATH(root, "external-link")), 0); +#endif + uint64_t bytes = 0; + uint64_t free_bytes = 0; + ASSERT_TRUE(cbm_directory_size_bytes(root, &bytes)); + ASSERT_EQ(bytes, 12); + ASSERT_TRUE(cbm_filesystem_free_bytes(root, &free_bytes)); + ASSERT_GT(free_bytes, 0); + th_cleanup(root); + PASS(); +} + +TEST(platform_db_artifact_measurement_saturates_and_counts_sidecars) { + char *root = th_mktempdir("cbm_storage_db_artifacts"); + ASSERT_NOT_NULL(root); + char db[CBM_SZ_4K]; + (void)snprintf(db, sizeof(db), "%s/project.db", root); + ASSERT_EQ(th_write_file(db, "12345"), 0); + ASSERT_EQ(th_write_file(TH_PATH(root, "project.db-wal"), "1234567"), 0); + ASSERT_EQ(th_write_file(TH_PATH(root, "project.db-shm"), "123"), 0); + uint64_t bytes = 0; + ASSERT_TRUE(cbm_db_artifact_bytes(db, &bytes)); + ASSERT_EQ(bytes, 15); + ASSERT_EQ(th_write_file(TH_PATH(root, "project.db.stage.task123.ABCDEF"), "12345678901"), 0); + ASSERT_EQ(th_write_file(TH_PATH(root, "project.db.stage.task123.ABCDEF-wal"), "1234567890123"), + 0); + ASSERT_EQ(th_write_file(TH_PATH(root, "project.db.stage.other456.ABCDEF"), "unrelated"), 0); + ASSERT_EQ(th_write_file(TH_PATH(root, "worker.log"), "12"), 0); + ASSERT_EQ(th_write_file(TH_PATH(root, "worker.response"), "123"), 0); + ASSERT_TRUE(cbm_index_task_temp_bytes(db, "task123", TH_PATH(root, "worker.log"), + TH_PATH(root, "worker.response"), &bytes)); + ASSERT_EQ(bytes, 29); + ASSERT_TRUE(cbm_index_staging_cleanup(db, "task123")); + ASSERT_FALSE(cbm_file_exists(TH_PATH(root, "project.db.stage.task123.ABCDEF"))); + ASSERT_FALSE(cbm_file_exists(TH_PATH(root, "project.db.stage.task123.ABCDEF-wal"))); + ASSERT_TRUE(cbm_file_exists(TH_PATH(root, "project.db.stage.other456.ABCDEF"))); + ASSERT_TRUE(cbm_file_exists(db)); + ASSERT_TRUE(cbm_file_exists(TH_PATH(root, "worker.log"))); + th_cleanup(root); + PASS(); +} + +#ifndef _WIN32 +TEST(platform_task_temp_measurement_supports_relative_database_paths) { + char *root = th_mktempdir("cbm_task_temp_relative"); + ASSERT_NOT_NULL(root); + char saved_cwd[CBM_SZ_4K]; + ASSERT_NOT_NULL(getcwd(saved_cwd, sizeof(saved_cwd))); + ASSERT_EQ(chdir(root), 0); + + int write_rc = th_write_file("project.db.stage.task123.ABCDEF", "12345"); + int wal_rc = th_write_file("project.db.stage.task123.ABCDEF-wal", "123456"); + int other_rc = th_write_file("project.db.stage.other456.ABCDEF", "unrelated"); + int log_rc = th_write_file("worker.log", "123"); + int response_rc = th_write_file("worker.response", "1234"); + uint64_t bytes = 0; + bool measured = + cbm_index_task_temp_bytes("project.db", "task123", "worker.log", "worker.response", &bytes); + bool cleaned = cbm_index_staging_cleanup("project.db", "task123"); + bool own_stage_removed = !cbm_file_exists("project.db.stage.task123.ABCDEF") && + !cbm_file_exists("project.db.stage.task123.ABCDEF-wal"); + bool other_stage_preserved = cbm_file_exists("project.db.stage.other456.ABCDEF"); + int restore_rc = chdir(saved_cwd); + th_cleanup(root); + + ASSERT_EQ(write_rc, 0); + ASSERT_EQ(wal_rc, 0); + ASSERT_EQ(other_rc, 0); + ASSERT_EQ(log_rc, 0); + ASSERT_EQ(response_rc, 0); + ASSERT_TRUE(measured); + ASSERT_EQ(bytes, 18); + ASSERT_TRUE(cleaned); + ASSERT_TRUE(own_stage_removed); + ASSERT_TRUE(other_stage_preserved); + ASSERT_EQ(restore_rc, 0); + PASS(); +} +#endif + TEST(platform_mmap) { /* mmap this test file and verify first bytes */ size_t sz = 0; @@ -1508,6 +1596,11 @@ SUITE(platform) { RUN_TEST(platform_file_exists); RUN_TEST(platform_is_dir); RUN_TEST(platform_file_size); + RUN_TEST(platform_storage_measurement_uses_logical_bytes_without_following_links); + RUN_TEST(platform_db_artifact_measurement_saturates_and_counts_sidecars); +#ifndef _WIN32 + RUN_TEST(platform_task_temp_measurement_supports_relative_database_paths); +#endif RUN_TEST(platform_mmap); RUN_TEST(platform_mmap_nonexistent); RUN_TEST(platform_path_helpers_use_per_thread_storage);