Bound import memory with staged streaming

1 files changed, 241 insertions, 0 deletions

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