Bound import memory with staged streaming

3 files changed, 350 insertions, 0 deletions

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
+}