diff options
| -rw-r--r-- | README.md | 2 | ||||
| -rw-r--r-- | docs/Design.md | 3 | ||||
| -rw-r--r-- | docs/Review.md | 11 | ||||
| -rw-r--r-- | scripts/benchmark_p1_cp.py | 47 | ||||
| -rw-r--r-- | src/app/main.c | 42 | ||||
| -rw-r--r-- | src/cli/cli.h | 2 | ||||
| -rw-r--r-- | src/cli/import.c | 436 | ||||
| -rw-r--r-- | src/storage/store.c | 215 | ||||
| -rw-r--r-- | src/storage/store.h | 13 | ||||
| -rw-r--r-- | tests/CMakeLists.txt | 57 | ||||
| -rw-r--r-- | tests/import_export_test.py | 52 | ||||
| -rw-r--r-- | tests/import_test.c | 241 |
12 files changed, 901 insertions, 220 deletions
@@ -100,6 +100,8 @@ filesystemを使用します。削除やcompactionはfilesystem、SSD、既存ba ## 容量とメモリ予算 importは1行`1 MiB + 4096 bytes`、100,000 record、decode後合計4 GiBが上限です。 +2-pass importは本文全件を保持せず、compact ID indexと最大1行だけをメモリに保持します。 +全件検証とstaging logのflushが成功した場合だけ`active.log`を置換します。 `get_system_health.memory_estimate`はstore entry、本文、metadata、search、history、relation indexの 確保量概算を返します。allocator overhead等は含まないため、実working setには余裕を持たせて ください。現行v1はlive corpusをメモリに保持し、retentionやdisk-backed bodyは実装しません。 diff --git a/docs/Design.md b/docs/Design.md index 3485a73..152f6ae 100644 --- a/docs/Design.md +++ b/docs/Design.md @@ -157,6 +157,9 @@ importはJSONLを行単位で読み、JSON stringをUTF-8へ完全decodeして 整数は小数・指数表記を許可せず、timestamp/updated_atは0～INT64_MAX、revisionは 1～INT32_MAX、importanceは0～100とする。1行はcontent上限+4096 bytes、100,000 record、 decode後合計4 GiBが上限である。file exportはtemporary fileのflush/close成功後だけ原子的に置換する。 +importは通常storeをロードせずlockだけを取得する。第1 passはcompact ID indexで全件と参照を検証し、 +第2 passは選択recordをstaging logへ直接書く。stagingのflush成功後だけactive logを原子的に置換し、 +失敗時は旧logを維持する。 ## 8. I/O 予算 diff --git a/docs/Review.md b/docs/Review.md index b4d351a..ed9f60c 100644 --- a/docs/Review.md +++ b/docs/Review.md @@ -12,7 +12,7 @@ bounded line streaming、atomic export、memory estimate、セキュリティ運 - 比較基準: commit `1efecaef1f867cb218366e2b4ab4b901fdd2835f` 以降 - 対象: domain、storage、MCP tools、CLI、system abstraction、tests、docs - 実行: Windows VS2026 Debug と ASan の build/CTest -- 対応後結果: Debug、Release、ASanはいずれも17/17 pass +- 対応後結果: Debug、Release、ASanはいずれも20/20 pass 復旧したprotocol workflowは各構成で検索1,000回、MCP lifecycle、rebuild、restartまで成功した。 @@ -21,8 +21,8 @@ Release import/export benchmark（2026-07-04、同一開発機）の結果は次 | records | import | export | round-trip import | |---:|---:|---:|---:| | 10,000 | 0.277 s | 0.165 s | 0.284 s | -| 50,000 | 4.668 s | 4.226 s | 4.794 s | -| 100,000 | 19.725 s | 18.465 s | 19.644 s | +| 50,000 | 0.854 s / 12,500,992 bytes peak | 4.327 s | 0.875 s / 11,616,256 bytes peak | +| 100,000 | 1.605 s / 16,285,696 bytes peak | 19.107 s | 1.774 s / 16,203,776 bytes peak | 既存のMCP workload計測では10,000件が18,960,384 bytes、50,000件が51,240,960 bytesの peak working setだった。100,000件はCLI上限とround-tripを確認し、実プロセスの継続監視には @@ -79,6 +79,11 @@ commit `2861118` で `tests/mcp_protocol_test.py` が削除されたが、`tests ### R3 — Medium: import size guard が約1 TiBで全体を一括確保する +対応済み。2-pass readerは最大1行とcompact ID indexだけを保持し、第2 passで選択recordを +`active.import.tmp`へ直接serializeする。flush後のatomic replaceまでは旧logを変更しない。 +100,001件、decoded上限±1、allocation全点、open/read/write/flush/replace失敗、dangling参照、 +truncated/長行を専用testで検証した。 + `IMPORT_MAX_LINE * 1024 * 1024` は約1 TiBとなる。実際には `ftell` と address space が先に制限するが、入力ファイル全体を `malloc` し、さらに parse 済み Message を保持する。 影響: diff --git a/scripts/benchmark_p1_cp.py b/scripts/benchmark_p1_cp.py index f456fbb..c6a9d0a 100644 --- a/scripts/benchmark_p1_cp.py +++ b/scripts/benchmark_p1_cp.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse +import ctypes import json import shutil import subprocess @@ -12,6 +13,21 @@ from pathlib import Path FIXTURE_PREFIX = "p1-cp-benchmark-" +class ProcessMemoryCounters(ctypes.Structure): + _fields_ = [ + ("cb", ctypes.c_ulong), + ("PageFaultCount", ctypes.c_ulong), + ("PeakWorkingSetSize", ctypes.c_size_t), + ("WorkingSetSize", ctypes.c_size_t), + ("QuotaPeakPagedPoolUsage", ctypes.c_size_t), + ("QuotaPagedPoolUsage", ctypes.c_size_t), + ("QuotaPeakNonPagedPoolUsage", ctypes.c_size_t), + ("QuotaNonPagedPoolUsage", ctypes.c_size_t), + ("PagefileUsage", ctypes.c_size_t), + ("PeakPagefileUsage", ctypes.c_size_t), + ] + + def make_ulid(timestamp_ms: int, random_bits: int) -> str: alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" bytes_data = bytearray(16) @@ -59,18 +75,28 @@ def write_fixture(path: Path, count: int) -> None: file.write(json.dumps(message, separators=(",", ":")) + "\n") -def run_import(executable: Path, data_dir: Path, input_path: Path) -> float: +def run_import(executable: Path, data_dir: Path, input_path: Path) -> tuple[float, int]: shutil.rmtree(data_dir, ignore_errors=True) start = time.perf_counter() - result = subprocess.run( + process = subprocess.Popen( [str(executable), "--data-dir", str(data_dir), "--import", str(input_path)], - capture_output=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, - check=True, ) + stdout, stderr = process.communicate() elapsed = time.perf_counter() - start - print(result.stdout.strip()) - return elapsed + if process.returncode != 0: + raise RuntimeError(stderr) + counters = ProcessMemoryCounters() + counters.cb = ctypes.sizeof(counters) + peak = 0 + if ctypes.windll.psapi.GetProcessMemoryInfo( # type: ignore[attr-defined] + int(process._handle), ctypes.byref(counters), counters.cb # type: ignore[attr-defined] + ): + peak = int(counters.PeakWorkingSetSize) + print(stdout.strip()) + return elapsed, peak def run_export(executable: Path, data_dir: Path, output_path: Path) -> float: @@ -105,15 +131,16 @@ def main() -> None: print(f"fixture size: {fixture_size:,} bytes") import_dir = work_dir / f"import-{count}" - import_ms = run_import(executable, import_dir, fixture) + import_ms, import_peak = run_import(executable, import_dir, fixture) print(f"import: {import_ms:.3f}s ({count / import_ms:,.0f} msg/s)") + print(f"import peak working set: {import_peak:,} bytes") export_path = work_dir / f"exported-{count}.jsonl" export_ms = run_export(executable, import_dir, export_path) print(f"export: {export_ms:.3f}s ({count / export_ms:,.0f} msg/s)") roundtrip_dir = work_dir / f"roundtrip-{count}" - roundtrip_ms = run_import(executable, roundtrip_dir, export_path) + roundtrip_ms, roundtrip_peak = run_import(executable, roundtrip_dir, export_path) print(f"round-trip import: {roundtrip_ms:.3f}s ({count / roundtrip_ms:,.0f} msg/s)") export_size = export_path.stat().st_size @@ -124,8 +151,10 @@ def main() -> None: "count": count, "fixture_bytes": fixture_size, "import_ms": round(import_ms * 1000, 3), + "import_peak_working_set_bytes": import_peak, "export_ms": round(export_ms * 1000, 3), "roundtrip_ms": round(roundtrip_ms * 1000, 3), + "roundtrip_peak_working_set_bytes": roundtrip_peak, "export_bytes": export_size, } ) @@ -135,4 +164,4 @@ def main() -> None: if __name__ == "__main__": - main()
\ No newline at end of file + main() diff --git a/src/app/main.c b/src/app/main.c index 415e999..157c8b2 100644 --- a/src/app/main.c +++ b/src/app/main.c @@ -61,26 +61,16 @@ run_server(tiny_store_t *store) return result; } -static int -run_cli_command(tiny_store_t *store, const tiny_sys_arguments_t *arguments) +static void +read_cli_options(tiny_cli_options_t *options, const tiny_sys_arguments_t *arguments) { - tiny_cli_options_t options = { 0 }; - options.data_dir = arguments->values[2]; - options.import_path = find_argument(arguments, "--import"); - options.export_path = find_argument(arguments, "--export"); - options.session_id = find_argument(arguments, "--session"); - options.policy_name = find_argument(arguments, "--policy"); - options.include_deleted = flag_present(arguments, "--include-deleted"); - - if (options.import_path) - { - return tiny_cli_import(store, &options); - } - if (options.export_path) - { - return tiny_cli_export(store, &options); - } - return 2; + memset(options, 0, sizeof(*options)); + options->data_dir = arguments->values[2]; + options->import_path = find_argument(arguments, "--import"); + options->export_path = find_argument(arguments, "--export"); + options->session_id = find_argument(arguments, "--session"); + options->policy_name = find_argument(arguments, "--policy"); + options->include_deleted = flag_present(arguments, "--include-deleted"); } static int @@ -91,6 +81,7 @@ run(const tiny_sys_arguments_t *arguments) tiny_store_status_t status; int result; int server_mode = 0; + tiny_cli_options_t options; if (arguments->count == 2 && strcmp(arguments->values[1], "--help") == 0) { usage(stdout); @@ -116,15 +107,24 @@ run(const tiny_sys_arguments_t *arguments) fprintf(stderr, "unable to resolve --data-dir\n"); return 2; } + read_cli_options(&options, arguments); + options.data_dir = absolute; + if (options.import_path) + { + result = tiny_cli_import(&options); + free(absolute); + return result; + } status = tiny_store_open(&store, absolute); - free(absolute); if (status != TINY_STORE_OK) { fprintf(stderr, "unable to open conversation store: %s\n", tiny_store_status_name(status)); + free(absolute); return status == TINY_STORE_BUSY ? 3 : 1; } - result = server_mode ? run_server(store) : run_cli_command(store, arguments); + result = server_mode ? run_server(store) : tiny_cli_export(store, &options); tiny_store_close(store); + free(absolute); return result; } diff --git a/src/cli/cli.h b/src/cli/cli.h index 1b173b4..6515a58 100644 --- a/src/cli/cli.h +++ b/src/cli/cli.h @@ -15,7 +15,7 @@ typedef struct tiny_cli_options int include_deleted; } tiny_cli_options_t; -int tiny_cli_import(tiny_store_t *store, const tiny_cli_options_t *options); +int tiny_cli_import(const tiny_cli_options_t *options); int tiny_cli_export(tiny_store_t *store, const tiny_cli_options_t *options); #endif diff --git a/src/cli/import.c b/src/cli/import.c index f4688ab..ee4d1cf 100644 --- a/src/cli/import.c +++ b/src/cli/import.c @@ -3,6 +3,7 @@ #include "domain/message.h" #include "mcp/json.h" #include "storage/store.h" +#include "sys/system.h" #include "mjson.h" @@ -11,55 +12,162 @@ #include <string.h> #include <limits.h> -#define IMPORT_MAX_LINE TINY_CLI_LINE_MAX -#define IMPORT_MAX_RECORDS 100000U -#define IMPORT_MAX_DECODED_BYTES (UINT64_C(4) * 1024U * 1024U * 1024U) +#include "base/allocator.h" -typedef struct import_record_list +#ifndef TINY_IMPORT_MAX_LINE +#define TINY_IMPORT_MAX_LINE TINY_CLI_LINE_MAX +#endif +#ifndef TINY_IMPORT_MAX_RECORDS +#define TINY_IMPORT_MAX_RECORDS 100000U +#endif +#ifndef TINY_IMPORT_MAX_DECODED_BYTES +#define TINY_IMPORT_MAX_DECODED_BYTES (UINT64_C(4) * 1024U * 1024U * 1024U) +#endif +#define IMPORT_READ_BUFFER 65536U + +typedef struct import_compact_entry +{ + char message_id[TINY_MESSAGE_ID_LENGTH + 1]; + size_t selected_record; + bool deleted; +} import_compact_entry_t; + +typedef struct import_compact_index { - tiny_import_record_t *items; + import_compact_entry_t *entries; size_t count; size_t capacity; -} import_record_list_t; + size_t *slots; + size_t slot_capacity; +} import_compact_index_t; + +typedef struct import_reader +{ + tiny_sys_file_t file; + unsigned char input[IMPORT_READ_BUFFER]; + size_t input_offset; + size_t input_length; + char *line; + size_t line_length; + bool eof; +} import_reader_t; static void -import_record_list_free(import_record_list_t *list) +import_compact_index_free(import_compact_index_t *index) { - size_t index; - if (!list) + if (!index) { return; } - for (index = 0; index < list->count; ++index) + free(index->slots); + free(index->entries); + memset(index, 0, sizeof(*index)); +} + +static uint64_t +import_id_hash(const char *value) +{ + uint64_t hash = UINT64_C(1469598103934665603); + while (*value) { - tiny_message_free(&list->items[index].message); + hash ^= (unsigned char)*value++; + hash *= UINT64_C(1099511628211); } - free(list->items); - list->items = NULL; - list->count = 0; - list->capacity = 0; + return hash; } static bool -import_record_list_push(import_record_list_t *list) +import_compact_index_rehash(import_compact_index_t *index, size_t capacity) { - tiny_import_record_t *resized; - size_t capacity; - if (list->count < list->capacity) + size_t *slots; + size_t position; + if (capacity > SIZE_MAX / sizeof(*slots) || !(slots = malloc(capacity * sizeof(*slots)))) { - return true; + return false; } - capacity = list->capacity == 0 ? 32 : list->capacity * 2; - resized = realloc(list->items, capacity * sizeof(*resized)); - if (!resized) + for (position = 0; position < capacity; ++position) { - return false; + slots[position] = SIZE_MAX; } - list->items = resized; - list->capacity = capacity; + for (position = 0; position < index->count; ++position) + { + size_t slot = (size_t)import_id_hash(index->entries[position].message_id) & (capacity - 1U); + while (slots[slot] != SIZE_MAX) + { + slot = (slot + 1U) & (capacity - 1U); + } + slots[slot] = position; + } + free(index->slots); + index->slots = slots; + index->slot_capacity = capacity; return true; } +static import_compact_entry_t * +import_compact_index_find(const import_compact_index_t *index, const char *message_id) +{ + size_t slot; + if (!index->slot_capacity) + { + return NULL; + } + slot = (size_t)import_id_hash(message_id) & (index->slot_capacity - 1U); + while (index->slots[slot] != SIZE_MAX) + { + import_compact_entry_t *entry = &index->entries[index->slots[slot]]; + if (strcmp(entry->message_id, message_id) == 0) + { + return entry; + } + slot = (slot + 1U) & (index->slot_capacity - 1U); + } + return NULL; +} + +static import_compact_entry_t * +import_compact_index_add(import_compact_index_t *index, const char *message_id) +{ + import_compact_entry_t *resized; + size_t capacity; + if (index->count == index->capacity) + { + capacity = index->capacity ? index->capacity * 2U : 32U; + if (capacity < index->capacity || capacity > SIZE_MAX / sizeof(*resized) + || !(resized = realloc(index->entries, capacity * sizeof(*resized)))) + { + return NULL; + } + index->entries = resized; + index->capacity = capacity; + } + if (!index->slot_capacity || (index->count + 1U) * 10U > index->slot_capacity * 7U) + { + capacity = index->slot_capacity ? index->slot_capacity * 2U : 64U; + if (capacity < index->slot_capacity || !import_compact_index_rehash(index, capacity)) + { + return NULL; + } + } + memset(&index->entries[index->count], 0, sizeof(index->entries[index->count])); + memcpy(index->entries[index->count].message_id, message_id, TINY_MESSAGE_ID_LENGTH + 1U); + { + size_t slot = (size_t)import_id_hash(message_id) & (index->slot_capacity - 1U); + while (index->slots[slot] != SIZE_MAX) + { + slot = (slot + 1U) & (index->slot_capacity - 1U); + } + index->slots[slot] = index->count; + } + index->count++; + if (index->count > TINY_IMPORT_MAX_RECORDS) + { + index->count--; + return NULL; + } + return &index->entries[index->count - 1U]; +} + static bool import_get_string(const char *json, int len, const char *path, char *out, size_t out_size) { @@ -518,7 +626,8 @@ import_manifest_valid(const char *line, int len) { char type_name[64]; uint64_t version = 0; - if (!import_get_string(line, len, "$.type", type_name, sizeof(type_name)) + if (mjson_find(line, len, "$", NULL, NULL) != MJSON_TOK_OBJECT + || !import_get_string(line, len, "$.type", type_name, sizeof(type_name)) || strcmp(type_name, "tiny-convmem-manifest") != 0 || !import_get_u64(line, len, "$.format_version", &version)) { @@ -532,7 +641,8 @@ import_parse_del_record(const char *line, int len, tiny_import_record_t *record) { char type_name[32]; tiny_message_init(&record->message); - if (!import_get_string(line, len, "$.type", type_name, sizeof(type_name)) + if (mjson_find(line, len, "$", NULL, NULL) != MJSON_TOK_OBJECT + || !import_get_string(line, len, "$.type", type_name, sizeof(type_name)) || strcmp(type_name, "DEL") != 0 || !import_get_string( line, len, "$.message_id", record->message.message_id, sizeof(record->message.message_id)) @@ -553,7 +663,8 @@ import_parse_message_record(const char *line, int len, tiny_import_record_t *rec size_t content_length = 0; tiny_message_init(&record->message); record->deleted = false; - if (!import_get_string( + if (mjson_find(line, len, "$", NULL, NULL) != MJSON_TOK_OBJECT + || !import_get_string( line, len, "$.message_id", record->message.message_id, sizeof(record->message.message_id)) || !import_get_string(line, len, "$.role", role_name, sizeof(role_name)) || !import_get_allocated_string( @@ -615,35 +726,129 @@ import_record_decoded_bytes(const tiny_import_record_t *record) } static bool -import_parse_stream(FILE *file, import_record_list_t *records) +import_reader_open(import_reader_t *reader, const char *path) { - char *line = malloc((size_t)IMPORT_MAX_LINE + 2U); - bool manifest_seen = false; - uint64_t decoded_total = 0; - if (!line) + memset(reader, 0, sizeof(*reader)); + tiny_sys_file_init(&reader->file); + reader->line = malloc((size_t)TINY_IMPORT_MAX_LINE + 1U); + return reader->line && tiny_sys_file_open(&reader->file, path, TINY_SYS_FILE_READ) == TINY_SYS_OK; +} + +static void +import_reader_close(import_reader_t *reader) +{ + tiny_sys_file_close(&reader->file); + free(reader->line); + memset(reader, 0, sizeof(*reader)); +} + +static bool +import_reader_rewind(import_reader_t *reader) +{ + reader->input_offset = 0; + reader->input_length = 0; + reader->line_length = 0; + reader->eof = false; + return tiny_sys_file_seek(&reader->file, 0, TINY_SYS_SEEK_BEGIN); +} + +static int +import_reader_next(import_reader_t *reader, const char **line, size_t *length) +{ + reader->line_length = 0; + while (true) { - return false; + unsigned char value; + if (reader->input_offset == reader->input_length) + { + uint32_t actual = 0; + if (reader->eof) + { + if (reader->line_length == 0) + { + return 0; + } + break; + } + if (!tiny_sys_file_read(&reader->file, reader->input, sizeof(reader->input), &actual)) + { + return -1; + } + reader->input_offset = 0; + reader->input_length = actual; + reader->eof = actual == 0; + continue; + } + value = reader->input[reader->input_offset++]; + if (value == '\n') + { + break; + } + if (value == 0 || reader->line_length >= TINY_IMPORT_MAX_LINE) + { + return -1; + } + reader->line[reader->line_length++] = (char)value; } - while (fgets(line, (int)IMPORT_MAX_LINE + 2, file) != NULL) + if (reader->line_length > 0 && reader->line[reader->line_length - 1U] == '\r') { - size_t length = strlen(line); - tiny_import_record_t record; - uint64_t decoded; - if (length == 0 || length > IMPORT_MAX_LINE - || (length == IMPORT_MAX_LINE + 1U && line[length - 1] != '\n') - || (line[length - 1] != '\n' && !feof(file))) + reader->line_length--; + } + reader->line[reader->line_length] = '\0'; + *line = reader->line; + *length = reader->line_length; + return 1; +} + +static bool +import_record_references_valid(const tiny_import_record_t *record, + const import_compact_index_t *index) +{ + size_t position; + if (record->deleted) + { + return true; + } + for (position = 0; position < record->message.metadata.relation_count; ++position) + { + import_compact_entry_t *target = import_compact_index_find( + index, record->message.metadata.relations[position].message_id); + if (!target || target->deleted) { - free(line); return false; } - if (length > 0 && line[length - 1] == '\n') - { - line[--length] = '\0'; - } - if (length > 0 && line[length - 1] == '\r') + } + for (position = 0; position < record->message.metadata.source_count; ++position) + { + import_compact_entry_t *target + = import_compact_index_find(index, record->message.metadata.source_message_ids[position]); + if (!target || target->deleted) { - line[--length] = '\0'; + return false; } + } + return true; +} + +static bool +import_validate_pass(import_reader_t *reader, + tiny_import_policy_t policy, + import_compact_index_t *index, + size_t *accepted, + size_t *rejected) +{ + const char *line; + size_t length, record_number = 0; + uint64_t decoded_total = 0; + bool manifest_seen = false; + int next; + *accepted = 0; + *rejected = 0; + while ((next = import_reader_next(reader, &line, &length)) > 0) + { + tiny_import_record_t record; + import_compact_entry_t *entry; + uint64_t decoded; if (length == 0) { continue; @@ -652,53 +857,111 @@ import_parse_stream(FILE *file, import_record_list_t *records) { if (!import_manifest_valid(line, (int)length)) { - free(line); return false; } manifest_seen = true; continue; } - if (records->count >= IMPORT_MAX_RECORDS) - { - free(line); - return false; - } - if (!import_parse_del_record(line, (int)length, &record) - && !import_parse_message_record(line, (int)length, &record)) + if (record_number >= TINY_IMPORT_MAX_RECORDS + || (!import_parse_del_record(line, (int)length, &record) + && !import_parse_message_record(line, (int)length, &record))) { - free(line); return false; } decoded = import_record_decoded_bytes(&record); - if (decoded > IMPORT_MAX_DECODED_BYTES - decoded_total || !import_record_list_push(records)) + if (decoded > TINY_IMPORT_MAX_DECODED_BYTES - decoded_total) { tiny_message_free(&record.message); - free(line); return false; } decoded_total += decoded; - records->items[records->count++] = record; + entry = import_compact_index_find(index, record.message.message_id); + if (!entry) + { + entry = import_compact_index_add(index, record.message.message_id); + if (!entry) + { + tiny_message_free(&record.message); + return false; + } + } + else if (policy == TINY_IMPORT_REJECT && !record.deleted && !entry->deleted) + { + (*rejected)++; + tiny_message_free(&record.message); + record_number++; + continue; + } + entry->selected_record = record_number; + entry->deleted = record.deleted; + (*accepted)++; + tiny_message_free(&record.message); + record_number++; } - if (ferror(file)) + return next == 0 && manifest_seen; +} + +static tiny_store_status_t +import_staging_pass(import_reader_t *reader, + const import_compact_index_t *index, + tiny_import_transaction_t *transaction) +{ + const char *line; + size_t length, record_number = 0; + bool manifest_seen = false; + int next; + while ((next = import_reader_next(reader, &line, &length)) > 0) { - free(line); - return false; + tiny_import_record_t record; + import_compact_entry_t *entry; + tiny_store_status_t status; + if (length == 0) + { + continue; + } + if (!manifest_seen) + { + manifest_seen = true; + continue; + } + if (!import_parse_del_record(line, (int)length, &record) + && !import_parse_message_record(line, (int)length, &record)) + { + return TINY_STORE_CORRUPT; + } + entry = import_compact_index_find(index, record.message.message_id); + if (entry && entry->selected_record == record_number && !record.deleted) + { + if (!import_record_references_valid(&record, index)) + { + tiny_message_free(&record.message); + return TINY_STORE_CORRUPT; + } + status = tiny_store_import_append(transaction, &record); + if (status != TINY_STORE_OK) + { + tiny_message_free(&record.message); + return status; + } + } + tiny_message_free(&record.message); + record_number++; } - free(line); - return manifest_seen; + return next == 0 ? TINY_STORE_OK : TINY_STORE_IO_ERROR; } int -tiny_cli_import(tiny_store_t *store, const tiny_cli_options_t *options) +tiny_cli_import(const tiny_cli_options_t *options) { - import_record_list_t records = { NULL, 0, 0 }; + import_compact_index_t index = { 0 }; + import_reader_t reader; + tiny_import_transaction_t *transaction = NULL; size_t imported = 0, rejected = 0; - tiny_store_status_t status; + tiny_store_status_t status = TINY_STORE_CORRUPT; tiny_import_policy_t policy = TINY_IMPORT_OVERWRITE; - FILE *file = NULL; int result = 1; - if (!options || !options->import_path) + if (!options || !options->data_dir || !options->import_path) { fprintf(stderr, "import: path is required\n"); return 1; @@ -721,29 +984,30 @@ tiny_cli_import(tiny_store_t *store, const tiny_cli_options_t *options) } } - if (fopen_s(&file, options->import_path, "rb") != 0 || !file) + if (!import_reader_open(&reader, options->import_path)) { fprintf(stderr, "import: unable to open '%s'\n", options->import_path); + import_reader_close(&reader); return 1; } - - if (!import_parse_stream(file, &records)) + if (!import_validate_pass(&reader, policy, &index, &imported, &rejected) + || !import_reader_rewind(&reader)) { fprintf(stderr, "import: invalid JSONL in '%s'\n", options->import_path); - fclose(file); - import_record_list_free(&records); - return 1; } - if (fclose(file) != 0) + else { - fprintf(stderr, "import: unable to close '%s'\n", options->import_path); - import_record_list_free(&records); - return 1; + status = tiny_store_import_begin(&transaction, options->data_dir); + if (status == TINY_STORE_OK) + { + status = import_staging_pass(&reader, &index, transaction); + } + if (status == TINY_STORE_OK) + { + status = tiny_store_import_commit(transaction); + } } - status = tiny_store_import(store, records.items, records.count, policy, &imported, &rejected); - import_record_list_free(&records); - if (status == TINY_STORE_OK) { printf("{\"imported\":%zu,\"rejected\":%zu}\n", imported, rejected); @@ -762,6 +1026,8 @@ tiny_cli_import(tiny_store_t *store, const tiny_cli_options_t *options) { fprintf(stderr, "import: failed\n"); } - + tiny_store_import_abort(transaction); + import_reader_close(&reader); + import_compact_index_free(&index); return result; } diff --git a/src/storage/store.c b/src/storage/store.c index e7ade28..d49f176 100644 --- a/src/storage/store.c +++ b/src/storage/store.c @@ -42,6 +42,17 @@ struct tiny_store char last_error[160]; }; +struct tiny_import_transaction +{ + tiny_sys_file_t lock; + tiny_sys_file_t staging; + char *staging_path; + char *log_path; + char *snapshot_path; + uint64_t sequence; + bool committed; +}; + static void set_error(tiny_store_t *store, const char *message) { @@ -2012,115 +2023,129 @@ tiny_store_health(const tiny_store_t *store, tiny_store_health_t *health) snprintf(health->last_error, sizeof(health->last_error), "%s", store->last_error); } +void +tiny_store_import_abort(tiny_import_transaction_t *transaction) +{ + if (!transaction) + { + return; + } + tiny_sys_file_close(&transaction->staging); + tiny_sys_file_close(&transaction->lock); + if (!transaction->committed && transaction->staging_path) + { + (void)tiny_sys_file_remove(transaction->staging_path); + } + free(transaction->snapshot_path); + free(transaction->log_path); + free(transaction->staging_path); + free(transaction); +} + tiny_store_status_t -tiny_store_import(tiny_store_t *store, - const tiny_import_record_t *records, - size_t record_count, - tiny_import_policy_t policy, - size_t *imported_count, - size_t *rejected_count) +tiny_store_import_begin(tiny_import_transaction_t **output, const char *directory) { - tiny_store_entry_t *saved_entries; - size_t saved_count, saved_capacity, index; - tiny_rebuild_report_t report; - tiny_store_status_t status = TINY_STORE_OK; - if (!store || !records || !imported_count || !rejected_count) + tiny_import_transaction_t *transaction; + char *lock_path; + unsigned char generation[16]; + tiny_sys_status_t status; + if (!output || !directory || !directory[0]) { return TINY_STORE_IO_ERROR; } - *imported_count = 0; - *rejected_count = 0; - saved_entries = store->entries; - saved_count = store->entry_count; - saved_capacity = store->entry_capacity; - store->entries = NULL; - store->entry_count = 0; - store->entry_capacity = 0; - for (index = 0; index < record_count && status == TINY_STORE_OK; ++index) + *output = NULL; + transaction = calloc(1, sizeof(*transaction)); + if (!transaction) { - const tiny_import_record_t *record = &records[index]; - tiny_store_entry_t *existing = find_entry(store, record->message.message_id); - if (existing != NULL && policy == TINY_IMPORT_REJECT && !record->deleted - && !existing->deleted) - { - (*rejected_count)++; - continue; - } - if (existing != NULL) - { - tiny_message_t replacement; - tiny_message_init(&replacement); - if (tiny_message_copy(&replacement, &record->message) != 0) - { - status = TINY_STORE_NO_MEMORY; - break; - } - tiny_message_free(&existing->message); - existing->deleted = record->deleted; - existing->sequence = (uint64_t)record->message.timestamp_ms; - existing->message = replacement; - (*imported_count)++; - } - else - { - tiny_message_t copied; - tiny_message_init(&copied); - if (!ensure_entries(store) || tiny_message_copy(&copied, &record->message) != 0) - { - tiny_message_free(&copied); - status = TINY_STORE_NO_MEMORY; - break; - } - memset( - &store->entries[store->entry_count], 0, sizeof(store->entries[store->entry_count])); - store->entries[store->entry_count].deleted = record->deleted; - store->entries[store->entry_count].sequence = (uint64_t)record->message.timestamp_ms; - store->entries[store->entry_count].message = copied; - store->entry_count++; - (*imported_count)++; - } + return TINY_STORE_NO_MEMORY; } - if (status == TINY_STORE_OK) + tiny_sys_file_init(&transaction->lock); + tiny_sys_file_init(&transaction->staging); + transaction->staging_path = tiny_sys_path_join(directory, "active.import.tmp"); + transaction->log_path = tiny_sys_path_join(directory, "active.log"); + transaction->snapshot_path = tiny_sys_path_join(directory, "snapshot.bin"); + lock_path = tiny_sys_path_join(directory, "store.lock"); + if (!transaction->staging_path || !transaction->log_path || !transaction->snapshot_path + || !lock_path) { - store->update_records = 0; - store->dead_bytes = 0; - store->snapshot_valid = false; - store->recovered_tail = false; - status = tiny_store_rebuild(store, &report); - if (status != TINY_STORE_OK) - { - for (index = 0; index < store->entry_count; ++index) - { - tiny_message_free(&store->entries[index].message); - } - free(store->entries); - store->entries = saved_entries; - store->entry_count = saved_count; - store->entry_capacity = saved_capacity; - } - else - { - for (index = 0; index < saved_count; ++index) - { - tiny_message_free(&saved_entries[index].message); - } - free(saved_entries); - } + free(lock_path); + tiny_store_import_abort(transaction); + return TINY_STORE_NO_MEMORY; } - else + if (tiny_sys_directory_create(directory) != TINY_SYS_OK) { - for (index = 0; index < store->entry_count; ++index) - { - tiny_message_free(&store->entries[index].message); - } - free(store->entries); - store->entries = saved_entries; - store->entry_count = saved_count; - store->entry_capacity = saved_capacity; + free(lock_path); + tiny_store_import_abort(transaction); + return TINY_STORE_IO_ERROR; + } + status = tiny_sys_file_open(&transaction->lock, lock_path, TINY_SYS_FILE_LOCK); + free(lock_path); + if (status != TINY_SYS_OK) + { + tiny_store_import_abort(transaction); + return status == TINY_SYS_BUSY ? TINY_STORE_BUSY : TINY_STORE_IO_ERROR; + } + if (tiny_sys_file_open(&transaction->staging, transaction->staging_path, TINY_SYS_FILE_WRITE_NEW) + != TINY_SYS_OK + || !tiny_sys_random_bytes(generation, sizeof(generation)) + || append_record_to( + &transaction->staging, RECORD_MARKER, 0, generation, sizeof(generation), false) + != TINY_STORE_OK) + { + tiny_store_import_abort(transaction); + return TINY_STORE_IO_ERROR; + } + *output = transaction; + return TINY_STORE_OK; +} + +tiny_store_status_t +tiny_store_import_append(tiny_import_transaction_t *transaction, const tiny_import_record_t *record) +{ + unsigned char *payload = NULL; + uint32_t length = 0; + tiny_store_status_t status; + if (!transaction || !record || transaction->committed) + { + return TINY_STORE_IO_ERROR; + } + transaction->sequence++; + if (record->deleted) + { + return append_record_to(&transaction->staging, + RECORD_DEL, + transaction->sequence, + (const unsigned char *)record->message.message_id, + TINY_MESSAGE_ID_LENGTH, + false); } + if (!serialize_put(&record->message, &payload, &length)) + { + return TINY_STORE_NO_MEMORY; + } + status = append_record_to( + &transaction->staging, RECORD_PUT, transaction->sequence, payload, length, false); + free(payload); return status; } +tiny_store_status_t +tiny_store_import_commit(tiny_import_transaction_t *transaction) +{ + if (!transaction || transaction->committed || !tiny_sys_file_flush(&transaction->staging)) + { + return TINY_STORE_IO_ERROR; + } + tiny_sys_file_close(&transaction->staging); + if (!tiny_sys_path_replace(transaction->staging_path, transaction->log_path)) + { + return TINY_STORE_IO_ERROR; + } + transaction->committed = true; + (void)tiny_sys_file_remove(transaction->snapshot_path); + return TINY_STORE_OK; +} + size_t tiny_store_entry_count(const tiny_store_t *store) { diff --git a/src/storage/store.h b/src/storage/store.h index 9efb02a..78c8cbe 100644 --- a/src/storage/store.h +++ b/src/storage/store.h @@ -26,6 +26,7 @@ typedef struct tiny_store_entry } tiny_store_entry_t; typedef struct tiny_store tiny_store_t; +typedef struct tiny_import_transaction tiny_import_transaction_t; typedef struct tiny_rebuild_report { @@ -94,12 +95,12 @@ typedef struct tiny_import_record } tiny_import_record_t; tiny_store_status_t tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report); -tiny_store_status_t tiny_store_import(tiny_store_t *store, - const tiny_import_record_t *records, - size_t record_count, - tiny_import_policy_t policy, - size_t *imported_count, - size_t *rejected_count); +tiny_store_status_t tiny_store_import_begin(tiny_import_transaction_t **transaction, + const char *directory); +tiny_store_status_t tiny_store_import_append(tiny_import_transaction_t *transaction, + const tiny_import_record_t *record); +tiny_store_status_t tiny_store_import_commit(tiny_import_transaction_t *transaction); +void tiny_store_import_abort(tiny_import_transaction_t *transaction); void tiny_store_health(const tiny_store_t *store, tiny_store_health_t *health); size_t tiny_store_entry_count(const tiny_store_t *store); diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index f2c51ce..94f188a 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -114,6 +114,60 @@ if(MSVC) endif() add_test(NAME tiny-convmem-mcp.storage COMMAND tiny-convmem-storage-test) +add_executable(tiny-convmem-import-test + import_test.c + "${PROJECT_SOURCE_DIR}/src/cli/import.c" + "${PROJECT_SOURCE_DIR}/src/domain/message.c" + "${PROJECT_SOURCE_DIR}/src/storage/codec.c" + "${PROJECT_SOURCE_DIR}/src/storage/store.c" +) +target_include_directories(tiny-convmem-import-test PRIVATE + "${PROJECT_SOURCE_DIR}/src" + "${PROJECT_SOURCE_DIR}/third_party/mcpc/src" +) +target_compile_features(tiny-convmem-import-test PRIVATE c_std_17) +target_link_libraries(tiny-convmem-import-test PRIVATE mcpc::mcpc tiny-convmem-sys) +if(MSVC) + target_compile_options(tiny-convmem-import-test PRIVATE /utf-8 /W4) +endif() +add_test(NAME tiny-convmem-mcp.import-faults COMMAND tiny-convmem-import-test) + +foreach(import_boundary records decoded) + add_executable(tiny-convmem-import-${import_boundary}-test + import_test.c + "${PROJECT_SOURCE_DIR}/src/cli/import.c" + "${PROJECT_SOURCE_DIR}/src/domain/message.c" + "${PROJECT_SOURCE_DIR}/src/storage/codec.c" + "${PROJECT_SOURCE_DIR}/src/storage/store.c" + ) + target_include_directories(tiny-convmem-import-${import_boundary}-test PRIVATE + "${PROJECT_SOURCE_DIR}/src" + "${PROJECT_SOURCE_DIR}/third_party/mcpc/src" + ) + target_compile_features(tiny-convmem-import-${import_boundary}-test PRIVATE c_std_17) + target_link_libraries(tiny-convmem-import-${import_boundary}-test PRIVATE mcpc::mcpc tiny-convmem-sys) + target_compile_definitions(tiny-convmem-import-${import_boundary}-test PRIVATE + TINY_IMPORT_TEST_BOUNDARIES=1 + ) + if(import_boundary STREQUAL "records") + target_compile_definitions(tiny-convmem-import-${import_boundary}-test PRIVATE + TINY_IMPORT_MAX_RECORDS=3 + TINY_IMPORT_MAX_DECODED_BYTES=1000 + ) + else() + target_compile_definitions(tiny-convmem-import-${import_boundary}-test PRIVATE + TINY_IMPORT_TEST_DECODED=1 + TINY_IMPORT_MAX_RECORDS=10 + TINY_IMPORT_MAX_DECODED_BYTES=100 + ) + endif() + if(MSVC) + target_compile_options(tiny-convmem-import-${import_boundary}-test PRIVATE /utf-8 /W4) + endif() + add_test(NAME tiny-convmem-mcp.import-${import_boundary}-limits + COMMAND tiny-convmem-import-${import_boundary}-test) +endforeach() + add_test( NAME tiny-convmem-mcp.lifecycle COMMAND "${CMAKE_COMMAND}" @@ -170,6 +224,9 @@ if(TINY_CONVMEM_ENABLE_ASAN AND MSVC) tiny-convmem-mcp.json tiny-convmem-mcp.sys tiny-convmem-mcp.storage + tiny-convmem-mcp.import-faults + tiny-convmem-mcp.import-records-limits + tiny-convmem-mcp.import-decoded-limits tiny-convmem-mcp.lifecycle tiny-convmem-mcp.cli tiny-convmem-mcp.requires-data-dir diff --git a/tests/import_export_test.py b/tests/import_export_test.py index 0ba5c5b..9e7fa02 100644 --- a/tests/import_export_test.py +++ b/tests/import_export_test.py @@ -11,6 +11,16 @@ from pathlib import Path MESSAGE_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV" +def make_ulid(index: int) -> str: + alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + value = index + output = ["0"] * 26 + for position in range(25, -1, -1): + output[position] = alphabet[value & 31] + value >>= 5 + return "".join(output) + + def record(content: str = "valid", **changes: object) -> dict[str, object]: value: dict[str, object] = { "message_id": MESSAGE_ID, @@ -103,6 +113,48 @@ def main() -> None: ) run(server, root / "long-data", "--import", str(long_line), success=False) + truncated = root / "truncated.jsonl" + write_jsonl(truncated, record()) + truncated.write_bytes(truncated.read_bytes()[:-8]) + run(server, root / "truncated-data", "--import", str(truncated), success=False) + + crlf = root / "crlf.jsonl" + crlf.write_bytes(source.read_bytes().replace(b"\n", b"\r\n", 1).replace(b"\r\n", b"\r\n\r\n", 1)) + run(server, root / "crlf-data", "--import", str(crlf), success=True) + + dangling = root / "dangling.jsonl" + dangling_record = record() + dangling_record["metadata"] = dict(dangling_record["metadata"]) # type: ignore[arg-type] + dangling_record["metadata"]["relations"] = [ # type: ignore[index] + {"type": "derived_from", "message_id": "01ARZ3NDEKTSV4RRFFQ69G5FAW"} + ] + write_jsonl(dangling, dangling_record) + run(server, root / "dangling-data", "--import", str(dangling), success=False) + + duplicate = root / "duplicate.jsonl" + duplicate.write_text( + json.dumps({"type": "tiny-convmem-manifest", "format_version": 2}) + + "\n" + + json.dumps(record("first"), separators=(",", ":")) + + "\n" + + json.dumps(record("last"), separators=(",", ":")) + + "\n", + encoding="utf-8", + ) + duplicate_data = root / "duplicate-data" + run(server, duplicate_data, "--import", str(duplicate), success=True) + duplicate_export = root / "duplicate-export.jsonl" + run(server, duplicate_data, "--export", str(duplicate_export), success=True) + assert json.loads(duplicate_export.read_text(encoding="utf-8").splitlines()[1])["content"] == "last" + + over_count = root / "over-count.jsonl" + with over_count.open("w", encoding="utf-8", newline="\n") as file: + file.write(json.dumps({"type": "tiny-convmem-manifest", "format_version": 2}) + "\n") + for index in range(100_001): + file.write(json.dumps({"type": "DEL", "message_id": make_ulid(index)}) + "\n") + run(server, root / "over-count-data", "--import", str(over_count), success=False) + assert not (root / "over-count-data" / "active.import.tmp").exists() + atomic_target = root / "atomic.jsonl" atomic_target.write_text("old backup", encoding="utf-8") run(server, first, "--export", str(atomic_target), success=True) diff --git a/tests/import_test.c b/tests/import_test.c new file mode 100644 index 0000000..87632d9 --- /dev/null +++ b/tests/import_test.c @@ -0,0 +1,241 @@ +#include "cli/cli.h" +#include "domain/message.h" +#include "storage/store.h" +#include "sys/system.h" + +#include <stdio.h> +#include <stdlib.h> +#include <string.h> + +#include "base/allocator.h" + +static int failures; +#define CHECK(value) \ + do \ + { \ + if (!(value)) \ + { \ + fprintf(stderr, "CHECK failed at line %d: %s\n", __LINE__, #value); \ + failures++; \ + } \ + } \ + while (0) + +static bool +write_fixture(const char *path) +{ + FILE *file = NULL; + if (fopen_s(&file, path, "wb") != 0 || !file) + { + return false; + } + fprintf(file, "{\"type\":\"tiny-convmem-manifest\",\"format_version\":2}\n"); + fprintf(file, + "{\"message_id\":\"01ARZ3NDEKTSV4RRFFQ69G5FAV\",\"session_id\":\"import\"," + "\"timestamp\":1,\"updated_at\":1,\"revision\":1,\"role\":\"user\"," + "\"content\":\"replacement\",\"metadata\":{\"memory_tier\":\"short\"," + "\"memory_kind\":\"conversation\",\"importance\":1}}\n"); + return fclose(file) == 0; +} + +static bool +write_limit_fixture(const char *path, size_t count, size_t content_length) +{ + static const char alphabet[] = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"; + FILE *file = NULL; + size_t record; + if (fopen_s(&file, path, "wb") != 0 || !file) + { + return false; + } + fprintf(file, "{\"type\":\"tiny-convmem-manifest\",\"format_version\":2}\n"); + for (record = 0; record < count; ++record) + { + size_t position; + fprintf(file, + "{\"message_id\":\"01ARZ3NDEKTSV4RRFFQ69G5FA%c\",\"session_id\":\"s\"," + "\"timestamp\":1,\"updated_at\":1,\"revision\":1,\"role\":\"user\"," + "\"content\":\"", + alphabet[record % 32U]); + for (position = 0; position < content_length; ++position) + { + fputc('x', file); + } + fprintf(file, + "\",\"metadata\":{\"memory_tier\":\"short\"," + "\"memory_kind\":\"conversation\",\"importance\":1}}\n"); + } + return fclose(file) == 0; +} + +static bool +seed_old_store(const char *directory) +{ + tiny_store_t *store = NULL; + tiny_message_t input, written; + bool ok; + tiny_message_init(&input); + tiny_message_init(&written); + input.session_id = malloc(sizeof("old")); + input.content = malloc(sizeof("old value")); + if (input.session_id) + { + memcpy(input.session_id, "old", sizeof("old")); + } + if (input.content) + { + memcpy(input.content, "old value", sizeof("old value")); + } + input.role = TINY_ROLE_USER; + ok = input.session_id && input.content && tiny_store_open(&store, directory) == TINY_STORE_OK + && tiny_store_write(store, &input, &written) == TINY_STORE_OK; + tiny_store_close(store); + tiny_message_free(&input); + tiny_message_free(&written); + return ok; +} + +static bool +old_store_intact(const char *directory) +{ + tiny_store_t *store = NULL; + bool intact = tiny_store_open(&store, directory) == TINY_STORE_OK + && tiny_store_entry_count(store) == 1 + && strcmp(tiny_store_entry(store, 0)->message.content, "old value") == 0; + tiny_store_close(store); + return intact; +} + +static bool +temporary_absent(const char *directory) +{ + char *path = tiny_sys_path_join(directory, "active.import.tmp"); + tiny_sys_file_t file; + bool absent; + tiny_sys_file_init(&file); + absent = path && tiny_sys_file_open(&file, path, TINY_SYS_FILE_READ) == TINY_SYS_NOT_FOUND; + tiny_sys_file_close(&file); + free(path); + return absent; +} + +static int +run_boundaries(void) +{ + char *root = tiny_sys_temp_directory_create("tcb"); + char *below = root ? tiny_sys_path_join(root, "below.jsonl") : NULL; + char *exact = root ? tiny_sys_path_join(root, "exact.jsonl") : NULL; + char *above = root ? tiny_sys_path_join(root, "above.jsonl") : NULL; + char *below_data = root ? tiny_sys_path_join(root, "below-data") : NULL; + char *exact_data = root ? tiny_sys_path_join(root, "exact-data") : NULL; + char *above_data = root ? tiny_sys_path_join(root, "above-data") : NULL; + tiny_cli_options_t options = { 0 }; +#ifdef TINY_IMPORT_TEST_DECODED + const size_t below_count = 1, exact_count = 1, above_count = 1; + const size_t below_content = 72, exact_content = 73, above_content = 74; +#else + const size_t below_count = 2, exact_count = 3, above_count = 4; + const size_t below_content = 1, exact_content = 1, above_content = 1; +#endif + CHECK(root && below && exact && above && below_data && exact_data && above_data); + CHECK(write_limit_fixture(below, below_count, below_content)); + CHECK(write_limit_fixture(exact, exact_count, exact_content)); + CHECK(write_limit_fixture(above, above_count, above_content)); + options.import_path = below; + options.data_dir = below_data; + CHECK(tiny_cli_import(&options) == 0); + options.import_path = exact; + options.data_dir = exact_data; + CHECK(tiny_cli_import(&options) == 0); + options.import_path = above; + options.data_dir = above_data; + CHECK(tiny_cli_import(&options) != 0); + CHECK(temporary_absent(above_data)); + free(above_data); + free(exact_data); + free(below_data); + free(above); + free(exact); + free(below); + free(root); + return failures == 0 ? 0 : 1; +} + +static int +run_faults(void) +{ + static const tiny_sys_test_operation_t operations[] = { TINY_SYS_TEST_OPEN, + TINY_SYS_TEST_READ, + TINY_SYS_TEST_WRITE, + TINY_SYS_TEST_FLUSH, + TINY_SYS_TEST_REPLACE }; + char *root = tiny_sys_temp_directory_create("tci"); + char *fixture = root ? tiny_sys_path_join(root, "fixture.jsonl") : NULL; + size_t operation; + CHECK(root && fixture && write_fixture(fixture)); + for (operation = 0; operation < sizeof(operations) / sizeof(operations[0]); ++operation) + { + size_t call; + size_t calls = operations[operation] == TINY_SYS_TEST_OPEN ? 3U + : operations[operation] == TINY_SYS_TEST_READ ? 4U + : operations[operation] == TINY_SYS_TEST_WRITE ? 4U + : 1U; + for (call = 1; call <= calls; ++call) + { + char name[32]; + char *directory; + tiny_cli_options_t options = { 0 }; + snprintf(name, sizeof(name), "io-%zu-%zu", operation, call); + directory = tiny_sys_path_join(root, name); + CHECK(directory && seed_old_store(directory)); + options.data_dir = directory; + options.import_path = fixture; + tiny_sys_test_reset(); + CHECK(tiny_sys_test_fail_after(operations[operation], call)); + CHECK(tiny_cli_import(&options) != 0); + tiny_sys_test_reset(); + CHECK(old_store_intact(directory)); + CHECK(temporary_absent(directory)); + free(directory); + } + } + { + char *directory = tiny_sys_path_join(root, "allocation"); + tiny_cli_options_t options = { 0 }; + size_t baseline; + size_t allocation; + bool succeeded = false; + CHECK(directory && seed_old_store(directory)); + baseline = tiny_allocator_outstanding(); + options.data_dir = directory; + options.import_path = fixture; + for (allocation = 0; allocation < 512; ++allocation) + { + tiny_allocator_fail_after(allocation); + if (tiny_cli_import(&options) == 0) + { + succeeded = true; + tiny_allocator_reset(); + break; + } + tiny_allocator_reset(); + CHECK(tiny_allocator_outstanding() == baseline); + CHECK(temporary_absent(directory)); + } + CHECK(succeeded); + free(directory); + } + free(fixture); + free(root); + return failures == 0 ? 0 : 1; +} + +int +main(void) +{ +#ifdef TINY_IMPORT_TEST_BOUNDARIES + return run_boundaries(); +#else + return run_faults(); +#endif +} |