summaryrefslogtreecommitdiff
path: root/src/storage
diff options
context:
space:
mode:
Diffstat (limited to 'src/storage')
-rw-r--r--src/storage/README.md9
-rw-r--r--src/storage/store.c1111
-rw-r--r--src/storage/store.h64
3 files changed, 1181 insertions, 3 deletions
diff --git a/src/storage/README.md b/src/storage/README.md
index ceb8965..3c03dba 100644
--- a/src/storage/README.md
+++ b/src/storage/README.md
@@ -1,6 +1,9 @@
# Storage layer
-This directory is reserved for persistence adapters. The storage engine and its
-schema are intentionally undecided; SQLite or another dependency will be added
-to `vcpkg.json` only after the storage contract is designed.
+The storage adapter owns `active.log`, `snapshot.bin`, the data-directory lock,
+in-memory live/deleted state, synchronous PUT/DEL writes, startup replay, and
+compaction. It uses Win32 64-bit file APIs and `FlushFileBuffers`.
+The binary format is private but versioned. Invalid complete records are never
+skipped; only an incomplete EOF record is truncated. Snapshot generation must
+match the active log before snapshot state is trusted.
diff --git a/src/storage/store.c b/src/storage/store.c
new file mode 100644
index 0000000..9d26e0f
--- /dev/null
+++ b/src/storage/store.c
@@ -0,0 +1,1111 @@
+#define WIN32_LEAN_AND_MEAN
+#include <windows.h>
+#include <bcrypt.h>
+
+#include "storage/store.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#define RECORD_HEADER_SIZE 24U
+#define RECORD_VERSION 1U
+#define RECORD_PUT 1U
+#define RECORD_DEL 2U
+#define RECORD_MARKER 3U
+#define RECORD_PAYLOAD_MAX (TINY_CONTENT_MAX + 16384U)
+#define SNAPSHOT_HEADER_SIZE 52U
+
+struct tiny_store
+{
+ HANDLE lock;
+ HANDLE log;
+ wchar_t *directory;
+ wchar_t *log_path;
+ wchar_t *snapshot_path;
+ tiny_store_entry_t *entries;
+ size_t entry_count;
+ size_t entry_capacity;
+ uint64_t last_sequence;
+ uint64_t last_ulid_ms;
+ unsigned char last_random[10];
+ unsigned char generation[16];
+ bool has_generation;
+ bool recovered_tail;
+ bool snapshot_valid;
+ char last_error[160];
+};
+
+static void
+put_u32(unsigned char *p, uint32_t v)
+{
+ p[0] = (unsigned char)v;
+ p[1] = (unsigned char)(v >> 8);
+ p[2] = (unsigned char)(v >> 16);
+ p[3] = (unsigned char)(v >> 24);
+}
+static void
+put_u64(unsigned char *p, uint64_t v)
+{
+ unsigned i;
+ for (i = 0; i < 8; ++i)
+ {
+ p[i] = (unsigned char)(v >> (i * 8));
+ }
+}
+static uint32_t
+get_u32(const unsigned char *p)
+{
+ return (uint32_t)p[0] | ((uint32_t)p[1] << 8) | ((uint32_t)p[2] << 16) | ((uint32_t)p[3] << 24);
+}
+static uint64_t
+get_u64(const unsigned char *p)
+{
+ uint64_t v = 0;
+ int i;
+ for (i = 7; i >= 0; --i)
+ {
+ v = (v << 8) | p[i];
+ }
+ return v;
+}
+
+static uint32_t
+crc32_bytes(const unsigned char *data, size_t length)
+{
+ uint32_t crc = 0xffffffffU;
+ size_t i;
+ for (i = 0; i < length; ++i)
+ {
+ unsigned bit;
+ crc ^= data[i];
+ for (bit = 0; bit < 8; ++bit)
+ {
+ crc = (crc >> 1) ^ (0xedb88320U & (uint32_t)-(int32_t)(crc & 1));
+ }
+ }
+ return ~crc;
+}
+
+static wchar_t *
+join_path(const wchar_t *directory, const wchar_t *name)
+{
+ const size_t left = wcslen(directory), right = wcslen(name);
+ const bool separator = left > 0 && directory[left - 1] != L'\\' && directory[left - 1] != L'/';
+ wchar_t *result = malloc((left + right + (separator ? 2 : 1)) * sizeof(wchar_t));
+ if (result == NULL)
+ {
+ return NULL;
+ }
+ wcscpy_s(result, left + right + (separator ? 2 : 1), directory);
+ if (separator)
+ {
+ wcscat_s(result, left + right + 2, L"\\");
+ }
+ wcscat_s(result, left + right + (separator ? 2 : 1), name);
+ return result;
+}
+
+static void
+set_error(tiny_store_t *store, const char *message)
+{
+ strncpy_s(store->last_error, sizeof(store->last_error), message, _TRUNCATE);
+}
+
+static bool
+write_all(HANDLE file, const void *data, uint32_t length)
+{
+ const unsigned char *position = data;
+ while (length > 0)
+ {
+ DWORD written = 0;
+ if (!WriteFile(file, position, length, &written, NULL) || written == 0)
+ {
+ return false;
+ }
+ position += written;
+ length -= written;
+ }
+ return true;
+}
+
+static bool
+read_all(HANDLE file, void *data, uint32_t length, uint32_t *actual)
+{
+ unsigned char *position = data;
+ uint32_t total = 0;
+ while (total < length)
+ {
+ DWORD count = 0;
+ if (!ReadFile(file, position + total, length - total, &count, NULL))
+ {
+ return false;
+ }
+ if (count == 0)
+ {
+ break;
+ }
+ total += count;
+ }
+ *actual = total;
+ return true;
+}
+
+static uint64_t
+now_milliseconds(void)
+{
+ FILETIME ft;
+ ULARGE_INTEGER ticks;
+ GetSystemTimeAsFileTime(&ft);
+ ticks.LowPart = ft.dwLowDateTime;
+ ticks.HighPart = ft.dwHighDateTime;
+ return ticks.QuadPart / 10000ULL - 11644473600000ULL;
+}
+
+static void
+make_ulid(tiny_store_t *store, uint64_t timestamp, char output[27])
+{
+ static const char alphabet[] = "0123456789ABCDEFGHJKMNPQRSTVWXYZ";
+ unsigned char bytes[16];
+ unsigned index;
+ if (timestamp > store->last_ulid_ms)
+ {
+ store->last_ulid_ms = timestamp;
+ if (BCryptGenRandom(
+ NULL, store->last_random, sizeof(store->last_random), BCRYPT_USE_SYSTEM_PREFERRED_RNG)
+ != 0)
+ {
+ uint64_t fallback = GetTickCount64() ^ timestamp;
+ for (index = 0; index < sizeof(store->last_random); ++index)
+ {
+ store->last_random[index] = (unsigned char)(fallback >> ((index % 8) * 8));
+ }
+ }
+ }
+ else
+ {
+ timestamp = store->last_ulid_ms;
+ for (index = 10; index-- > 0;)
+ {
+ if (++store->last_random[index] != 0)
+ {
+ break;
+ }
+ }
+ }
+ bytes[0] = (unsigned char)(timestamp >> 40);
+ bytes[1] = (unsigned char)(timestamp >> 32);
+ bytes[2] = (unsigned char)(timestamp >> 24);
+ bytes[3] = (unsigned char)(timestamp >> 16);
+ bytes[4] = (unsigned char)(timestamp >> 8);
+ bytes[5] = (unsigned char)timestamp;
+ memcpy(bytes + 6, store->last_random, 10);
+ for (index = 0; index < 26; ++index)
+ {
+ const unsigned bit = index * 5;
+ unsigned value = 0, part;
+ for (part = 0; part < 5; ++part)
+ {
+ const int source_bit = (int)bit + (int)part - 2;
+ value <<= 1;
+ if (source_bit >= 0)
+ {
+ value |= (bytes[source_bit / 8] >> (7 - source_bit % 8)) & 1U;
+ }
+ }
+ output[index] = alphabet[value];
+ }
+ output[26] = '\0';
+}
+
+static bool
+ensure_entries(tiny_store_t *store)
+{
+ tiny_store_entry_t *resized;
+ size_t capacity;
+ if (store->entry_count < store->entry_capacity)
+ {
+ return true;
+ }
+ capacity = store->entry_capacity == 0 ? 32 : store->entry_capacity * 2;
+ resized = realloc(store->entries, capacity * sizeof(*resized));
+ if (resized == NULL)
+ {
+ return false;
+ }
+ store->entries = resized;
+ store->entry_capacity = capacity;
+ return true;
+}
+
+static tiny_store_entry_t *
+find_entry(tiny_store_t *store, const char *message_id)
+{
+ size_t index;
+ for (index = 0; index < store->entry_count; ++index)
+ {
+ if (strcmp(store->entries[index].message.message_id, message_id) == 0)
+ {
+ return &store->entries[index];
+ }
+ }
+ return NULL;
+}
+
+static bool
+serialize_put(const tiny_message_t *message, unsigned char **payload, uint32_t *length)
+{
+ size_t session_len = strlen(message->session_id), content_len = strlen(message->content);
+ size_t tool_len = message->metadata.tool_call_id ? strlen(message->metadata.tool_call_id) : 0;
+ size_t total = 8 + 26 + 2 + 16 + session_len + content_len + tool_len;
+ size_t index, offset;
+ unsigned char *data;
+ for (index = 0; index < message->metadata.tag_count; ++index)
+ {
+ total += 4 + strlen(message->metadata.tags[index]);
+ }
+ if (total > UINT32_MAX || total > RECORD_PAYLOAD_MAX)
+ {
+ return false;
+ }
+ data = malloc(total);
+ if (data == NULL)
+ {
+ return false;
+ }
+ offset = 0;
+ put_u64(data + offset, (uint64_t)message->timestamp_ms);
+ offset += 8;
+ memcpy(data + offset, message->message_id, 26);
+ offset += 26;
+ data[offset++] = (unsigned char)message->role;
+ data[offset++] = (unsigned char)message->metadata.memory_tier;
+ put_u32(data + offset, (uint32_t)session_len);
+ offset += 4;
+ put_u32(data + offset, (uint32_t)content_len);
+ offset += 4;
+ put_u32(data + offset, (uint32_t)tool_len);
+ offset += 4;
+ put_u32(data + offset, (uint32_t)message->metadata.tag_count);
+ offset += 4;
+ memcpy(data + offset, message->session_id, session_len);
+ offset += session_len;
+ memcpy(data + offset, message->content, content_len);
+ offset += content_len;
+ if (tool_len > 0)
+ {
+ memcpy(data + offset, message->metadata.tool_call_id, tool_len);
+ offset += tool_len;
+ }
+ for (index = 0; index < message->metadata.tag_count; ++index)
+ {
+ const size_t tag_len = strlen(message->metadata.tags[index]);
+ put_u32(data + offset, (uint32_t)tag_len);
+ offset += 4;
+ memcpy(data + offset, message->metadata.tags[index], tag_len);
+ offset += tag_len;
+ }
+ *payload = data;
+ *length = (uint32_t)total;
+ return true;
+}
+
+static char *
+copy_field(const unsigned char *data, size_t length)
+{
+ char *value = malloc(length + 1);
+ if (value != NULL)
+ {
+ memcpy(value, data, length);
+ value[length] = '\0';
+ }
+ return value;
+}
+
+static bool
+deserialize_put(const unsigned char *data, uint32_t length, tiny_message_t *message)
+{
+ uint32_t session_len, content_len, tool_len, tag_count;
+ size_t offset = 0, index;
+ tiny_message_init(message);
+ if (length < 8 + 26 + 2 + 16)
+ {
+ return false;
+ }
+ message->timestamp_ms = (int64_t)get_u64(data + offset);
+ offset += 8;
+ memcpy(message->message_id, data + offset, 26);
+ message->message_id[26] = '\0';
+ offset += 26;
+ message->role = (tiny_role_t)data[offset++];
+ message->metadata.memory_tier = (tiny_memory_tier_t)data[offset++];
+ session_len = get_u32(data + offset);
+ offset += 4;
+ content_len = get_u32(data + offset);
+ offset += 4;
+ tool_len = get_u32(data + offset);
+ offset += 4;
+ tag_count = get_u32(data + offset);
+ offset += 4;
+ if (!tiny_message_id_valid(message->message_id)
+ || (message->role < TINY_ROLE_USER || message->role > TINY_ROLE_SYSTEM)
+ || (message->metadata.memory_tier < TINY_TIER_SHORT
+ || message->metadata.memory_tier > TINY_TIER_DISCARD)
+ || session_len == 0 || session_len > TINY_SESSION_ID_MAX || content_len == 0
+ || content_len > TINY_CONTENT_MAX || tool_len > TINY_TOOL_CALL_ID_MAX
+ || tag_count > TINY_TAG_COUNT_MAX || offset + session_len + content_len + tool_len > length)
+ {
+ return false;
+ }
+ message->session_id = copy_field(data + offset, session_len);
+ offset += session_len;
+ message->content = copy_field(data + offset, content_len);
+ offset += content_len;
+ if (tool_len > 0)
+ {
+ message->metadata.tool_call_id = copy_field(data + offset, tool_len);
+ offset += tool_len;
+ }
+ if (message->session_id == NULL || message->content == NULL
+ || (tool_len && !message->metadata.tool_call_id))
+ {
+ goto fail;
+ }
+ if (tag_count > 0)
+ {
+ message->metadata.tags = calloc(tag_count, sizeof(char *));
+ if (message->metadata.tags == NULL)
+ {
+ goto fail;
+ }
+ for (index = 0; index < tag_count; ++index)
+ {
+ uint32_t tag_len;
+ if (offset + 4 > length)
+ {
+ goto fail;
+ }
+ tag_len = get_u32(data + offset);
+ offset += 4;
+ if (tag_len == 0 || tag_len > TINY_TAG_MAX || offset + tag_len > length)
+ {
+ goto fail;
+ }
+ message->metadata.tags[index] = copy_field(data + offset, tag_len);
+ offset += tag_len;
+ if (message->metadata.tags[index] == NULL)
+ {
+ goto fail;
+ }
+ message->metadata.tag_count++;
+ }
+ }
+ if (offset != length || !tiny_utf8_valid(message->session_id, session_len)
+ || !tiny_utf8_valid(message->content, content_len))
+ {
+ goto fail;
+ }
+ return true;
+fail:
+ tiny_message_free(message);
+ return false;
+}
+
+static tiny_store_status_t
+append_record_to(HANDLE file,
+ uint8_t type,
+ uint64_t sequence,
+ const unsigned char *payload,
+ uint32_t payload_length,
+ bool flush)
+{
+ unsigned char header[RECORD_HEADER_SIZE] = { 0 };
+ LARGE_INTEGER end;
+ memcpy(header, "TCMR", 4);
+ header[4] = RECORD_VERSION;
+ header[5] = type;
+ put_u32(header + 8, payload_length);
+ put_u64(header + 12, sequence);
+ put_u32(header + 20, crc32_bytes(payload, payload_length));
+ end.QuadPart = 0;
+ if (!SetFilePointerEx(file, end, NULL, FILE_END) || !write_all(file, header, sizeof(header))
+ || !write_all(file, payload, payload_length) || (flush && !FlushFileBuffers(file)))
+ {
+ return TINY_STORE_IO_ERROR;
+ }
+ return TINY_STORE_OK;
+}
+
+static tiny_store_status_t
+apply_put(tiny_store_t *store, tiny_message_t *message, uint64_t sequence)
+{
+ tiny_store_entry_t *entry = find_entry(store, message->message_id);
+ if (entry != NULL)
+ {
+ return TINY_STORE_CORRUPT;
+ }
+ if (!ensure_entries(store))
+ {
+ return TINY_STORE_NO_MEMORY;
+ }
+ entry = &store->entries[store->entry_count++];
+ memset(entry, 0, sizeof(*entry));
+ entry->message = *message;
+ tiny_message_init(message);
+ entry->sequence = sequence;
+ return TINY_STORE_OK;
+}
+
+static void
+clear_entries(tiny_store_t *store)
+{
+ size_t index;
+ for (index = 0; index < store->entry_count; ++index)
+ {
+ tiny_message_free(&store->entries[index].message);
+ }
+ store->entry_count = 0;
+ store->last_sequence = 0;
+}
+
+static tiny_store_status_t
+replay_log(tiny_store_t *store, uint64_t starting_offset)
+{
+ LARGE_INTEGER start, file_size, boundary;
+ start.QuadPart = (LONGLONG)starting_offset;
+ boundary = start;
+ if (!SetFilePointerEx(store->log, start, NULL, FILE_BEGIN)
+ || !GetFileSizeEx(store->log, &file_size))
+ {
+ return TINY_STORE_IO_ERROR;
+ }
+ while (boundary.QuadPart < file_size.QuadPart)
+ {
+ unsigned char header[RECORD_HEADER_SIZE], *payload = NULL;
+ uint32_t actual = 0, payload_length, expected_crc;
+ uint64_t sequence;
+ tiny_store_status_t status = TINY_STORE_OK;
+ if (!read_all(store->log, header, sizeof(header), &actual))
+ {
+ return TINY_STORE_IO_ERROR;
+ }
+ if (actual < sizeof(header))
+ {
+ goto torn_tail;
+ }
+ if (memcmp(header, "TCMR", 4) != 0 || header[4] != RECORD_VERSION
+ || (header[5] != RECORD_PUT && header[5] != RECORD_DEL && header[5] != RECORD_MARKER))
+ {
+ return TINY_STORE_CORRUPT;
+ }
+ payload_length = get_u32(header + 8);
+ sequence = get_u64(header + 12);
+ expected_crc = get_u32(header + 20);
+ if (payload_length > RECORD_PAYLOAD_MAX
+ || (header[5] != RECORD_MARKER && sequence <= store->last_sequence))
+ {
+ return TINY_STORE_CORRUPT;
+ }
+ payload = malloc(payload_length == 0 ? 1 : payload_length);
+ if (payload == NULL)
+ {
+ return TINY_STORE_NO_MEMORY;
+ }
+ if (!read_all(store->log, payload, payload_length, &actual))
+ {
+ free(payload);
+ return TINY_STORE_IO_ERROR;
+ }
+ if (actual < payload_length)
+ {
+ free(payload);
+ goto torn_tail;
+ }
+ if (crc32_bytes(payload, payload_length) != expected_crc)
+ {
+ free(payload);
+ return TINY_STORE_CORRUPT;
+ }
+ if (header[5] == RECORD_MARKER)
+ {
+ if (boundary.QuadPart != 0 || sequence != 0
+ || payload_length != sizeof(store->generation))
+ {
+ free(payload);
+ return TINY_STORE_CORRUPT;
+ }
+ memcpy(store->generation, payload, sizeof(store->generation));
+ store->has_generation = true;
+ }
+ else if (header[5] == RECORD_PUT)
+ {
+ tiny_message_t message;
+ if (!deserialize_put(payload, payload_length, &message))
+ {
+ free(payload);
+ return TINY_STORE_CORRUPT;
+ }
+ status = apply_put(store, &message, sequence);
+ tiny_message_free(&message);
+ }
+ else
+ {
+ char id[27];
+ tiny_store_entry_t *entry;
+ if (payload_length != 34)
+ {
+ free(payload);
+ return TINY_STORE_CORRUPT;
+ }
+ memcpy(id, payload, 26);
+ id[26] = '\0';
+ entry = find_entry(store, id);
+ if (entry == NULL || entry->deleted || !tiny_message_id_valid(id))
+ {
+ free(payload);
+ return TINY_STORE_CORRUPT;
+ }
+ entry->deleted = true;
+ }
+ free(payload);
+ if (status != TINY_STORE_OK)
+ {
+ return status;
+ }
+ if (header[5] != RECORD_MARKER)
+ {
+ store->last_sequence = sequence;
+ }
+ boundary.QuadPart += RECORD_HEADER_SIZE + payload_length;
+ }
+ return TINY_STORE_OK;
+
+torn_tail:
+ if (!SetFilePointerEx(store->log, boundary, NULL, FILE_BEGIN) || !SetEndOfFile(store->log)
+ || !FlushFileBuffers(store->log))
+ {
+ return TINY_STORE_IO_ERROR;
+ }
+ store->recovered_tail = true;
+ set_error(store, "an incomplete trailing record was truncated during recovery");
+ return TINY_STORE_OK;
+}
+
+static tiny_store_status_t
+write_snapshot(tiny_store_t *store)
+{
+ wchar_t *temporary = join_path(store->directory, L"snapshot.tmp");
+ HANDLE file;
+ unsigned char data[SNAPSHOT_HEADER_SIZE] = { 0 };
+ LARGE_INTEGER size;
+ bool ok;
+ size_t index;
+ if (temporary == NULL)
+ {
+ return TINY_STORE_NO_MEMORY;
+ }
+ file = CreateFileW(temporary, GENERIC_WRITE, 0, NULL, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
+ if (file == INVALID_HANDLE_VALUE)
+ {
+ free(temporary);
+ return TINY_STORE_IO_ERROR;
+ }
+ GetFileSizeEx(store->log, &size);
+ memcpy(data, "TCMS", 4);
+ data[4] = 1;
+ memcpy(data + 8, store->generation, 16);
+ put_u64(data + 24, store->last_sequence);
+ put_u64(data + 32, (uint64_t)size.QuadPart);
+ put_u64(data + 40, (uint64_t)store->entry_count);
+ put_u32(data + 48, crc32_bytes(data, 48));
+ ok = write_all(file, data, sizeof(data));
+ for (index = 0; ok && index < store->entry_count; ++index)
+ {
+ unsigned char prefix[16], *payload = NULL;
+ uint32_t length = 0;
+ if (!serialize_put(&store->entries[index].message, &payload, &length))
+ {
+ ok = false;
+ break;
+ }
+ put_u64(prefix, store->entries[index].sequence);
+ put_u32(prefix + 8, length);
+ put_u32(prefix + 12, crc32_bytes(payload, length));
+ ok = write_all(file, prefix, sizeof(prefix)) && write_all(file, payload, length);
+ free(payload);
+ }
+ if (ok)
+ {
+ ok = FlushFileBuffers(file) != 0;
+ }
+ CloseHandle(file);
+ if (ok)
+ {
+ ok = MoveFileExW(
+ temporary, store->snapshot_path, MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH)
+ != 0;
+ }
+ free(temporary);
+ store->snapshot_valid = ok;
+ return ok ? TINY_STORE_OK : TINY_STORE_IO_ERROR;
+}
+
+static bool
+log_generation_matches(tiny_store_t *store, const unsigned char generation[16])
+{
+ unsigned char header[RECORD_HEADER_SIZE], payload[16];
+ uint32_t actual = 0;
+ LARGE_INTEGER zero = { 0 };
+ if (!SetFilePointerEx(store->log, zero, NULL, FILE_BEGIN)
+ || !read_all(store->log, header, sizeof(header), &actual) || actual != sizeof(header)
+ || memcmp(header, "TCMR", 4) != 0 || header[4] != RECORD_VERSION || header[5] != RECORD_MARKER
+ || get_u64(header + 12) != 0 || get_u32(header + 8) != 16
+ || !read_all(store->log, payload, sizeof(payload), &actual) || actual != sizeof(payload)
+ || get_u32(header + 20) != crc32_bytes(payload, sizeof(payload)))
+ {
+ return false;
+ }
+ return memcmp(payload, generation, 16) == 0;
+}
+
+static bool
+load_snapshot(tiny_store_t *store, uint64_t *log_offset)
+{
+ HANDLE file = CreateFileW(store->snapshot_path,
+ GENERIC_READ,
+ FILE_SHARE_READ,
+ NULL,
+ OPEN_EXISTING,
+ FILE_ATTRIBUTE_NORMAL,
+ NULL);
+ unsigned char header[SNAPSHOT_HEADER_SIZE];
+ uint32_t actual = 0;
+ uint64_t count, index, expected_sequence, offset;
+ LARGE_INTEGER log_size;
+ if (file == INVALID_HANDLE_VALUE)
+ {
+ return false;
+ }
+ if (!read_all(file, header, sizeof(header), &actual) || actual != sizeof(header)
+ || memcmp(header, "TCMS", 4) != 0 || header[4] != 1
+ || get_u32(header + 48) != crc32_bytes(header, 48) || !GetFileSizeEx(store->log, &log_size))
+ {
+ goto invalid;
+ }
+ expected_sequence = get_u64(header + 24);
+ offset = get_u64(header + 32);
+ count = get_u64(header + 40);
+ if (offset > (uint64_t)log_size.QuadPart || count > SIZE_MAX
+ || !log_generation_matches(store, header + 8))
+ {
+ goto invalid;
+ }
+ memcpy(store->generation, header + 8, 16);
+ store->has_generation = true;
+ for (index = 0; index < count; ++index)
+ {
+ unsigned char prefix[16], *payload;
+ uint32_t length, crc;
+ tiny_message_t message;
+ tiny_store_status_t status;
+ if (!read_all(file, prefix, sizeof(prefix), &actual) || actual != sizeof(prefix))
+ {
+ goto invalid_entries;
+ }
+ length = get_u32(prefix + 8);
+ crc = get_u32(prefix + 12);
+ if (length > RECORD_PAYLOAD_MAX)
+ {
+ goto invalid_entries;
+ }
+ payload = malloc(length ? length : 1);
+ if (!payload)
+ {
+ goto invalid_entries;
+ }
+ if (!read_all(file, payload, length, &actual) || actual != length
+ || crc32_bytes(payload, length) != crc || !deserialize_put(payload, length, &message))
+ {
+ free(payload);
+ goto invalid_entries;
+ }
+ free(payload);
+ status = apply_put(store, &message, get_u64(prefix));
+ tiny_message_free(&message);
+ if (status != TINY_STORE_OK)
+ {
+ goto invalid_entries;
+ }
+ store->last_sequence = get_u64(prefix);
+ }
+ if (store->last_sequence != expected_sequence)
+ {
+ goto invalid_entries;
+ }
+ CloseHandle(file);
+ *log_offset = offset;
+ store->snapshot_valid = true;
+ return true;
+invalid_entries:
+ clear_entries(store);
+ store->has_generation = false;
+invalid:
+ set_error(store, "snapshot validation failed; the active log was replayed from the beginning");
+ CloseHandle(file);
+ return false;
+}
+
+tiny_store_status_t
+tiny_store_open(tiny_store_t **output, const wchar_t *directory)
+{
+ tiny_store_t *store = calloc(1, sizeof(*store));
+ wchar_t *lock_path = NULL;
+ tiny_store_status_t status;
+ if (store == NULL)
+ {
+ return TINY_STORE_NO_MEMORY;
+ }
+ store->lock = INVALID_HANDLE_VALUE;
+ store->log = INVALID_HANDLE_VALUE;
+ store->directory = _wcsdup(directory);
+ store->log_path = join_path(directory, L"active.log");
+ store->snapshot_path = join_path(directory, L"snapshot.bin");
+ lock_path = join_path(directory, L"store.lock");
+ if (!store->directory || !store->log_path || !store->snapshot_path || !lock_path)
+ {
+ status = TINY_STORE_NO_MEMORY;
+ goto fail;
+ }
+ if (!CreateDirectoryW(directory, NULL) && GetLastError() != ERROR_ALREADY_EXISTS)
+ {
+ status = TINY_STORE_IO_ERROR;
+ goto fail;
+ }
+ store->lock = CreateFileW(
+ lock_path, GENERIC_READ | GENERIC_WRITE, 0, NULL, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
+ free(lock_path);
+ lock_path = NULL;
+ if (store->lock == INVALID_HANDLE_VALUE)
+ {
+ status = GetLastError() == ERROR_SHARING_VIOLATION ? TINY_STORE_BUSY : TINY_STORE_IO_ERROR;
+ goto fail;
+ }
+ store->log = CreateFileW(store->log_path,
+ GENERIC_READ | GENERIC_WRITE,
+ FILE_SHARE_READ,
+ NULL,
+ OPEN_ALWAYS,
+ FILE_ATTRIBUTE_NORMAL,
+ NULL);
+ if (store->log == INVALID_HANDLE_VALUE)
+ {
+ status = TINY_STORE_IO_ERROR;
+ goto fail;
+ }
+ {
+ uint64_t replay_offset = 0;
+ (void)load_snapshot(store, &replay_offset);
+ status = replay_log(store, replay_offset);
+ }
+ if (status != TINY_STORE_OK)
+ {
+ goto fail;
+ }
+ *output = store;
+ return TINY_STORE_OK;
+fail:
+ free(lock_path);
+ tiny_store_close(store);
+ return status;
+}
+
+void
+tiny_store_close(tiny_store_t *store)
+{
+ size_t index;
+ if (!store)
+ {
+ return;
+ }
+ if (store->log != INVALID_HANDLE_VALUE)
+ {
+ CloseHandle(store->log);
+ }
+ if (store->lock != INVALID_HANDLE_VALUE)
+ {
+ CloseHandle(store->lock);
+ }
+ for (index = 0; index < store->entry_count; ++index)
+ {
+ tiny_message_free(&store->entries[index].message);
+ }
+ free(store->entries);
+ free(store->directory);
+ free(store->log_path);
+ free(store->snapshot_path);
+ free(store);
+}
+
+const char *
+tiny_store_status_name(tiny_store_status_t status)
+{
+ switch (status)
+ {
+ case TINY_STORE_OK:
+ return "OK";
+ case TINY_STORE_NOT_FOUND:
+ return "NOT_FOUND";
+ case TINY_STORE_BUSY:
+ return "BUSY";
+ case TINY_STORE_IO_ERROR:
+ return "IO_ERROR";
+ case TINY_STORE_CORRUPT:
+ return "CORRUPT_STORE";
+ default:
+ return "INTERNAL_ERROR";
+ }
+}
+
+tiny_store_status_t
+tiny_store_write(tiny_store_t *store, const tiny_message_t *input, tiny_message_t *written)
+{
+ tiny_message_t message, indexed;
+ unsigned char *payload = NULL;
+ uint32_t length = 0;
+ tiny_store_status_t status;
+ tiny_message_init(&message);
+ tiny_message_init(&indexed);
+ if (tiny_message_copy(&message, input) != 0)
+ {
+ return TINY_STORE_NO_MEMORY;
+ }
+ message.timestamp_ms = (int64_t)now_milliseconds();
+ make_ulid(store, (uint64_t)message.timestamp_ms, message.message_id);
+ if (!serialize_put(&message, &payload, &length) || !ensure_entries(store)
+ || tiny_message_copy(&indexed, &message) != 0)
+ {
+ free(payload);
+ tiny_message_free(&message);
+ tiny_message_free(&indexed);
+ return TINY_STORE_NO_MEMORY;
+ }
+ status
+ = append_record_to(store->log, RECORD_PUT, store->last_sequence + 1, payload, length, true);
+ free(payload);
+ if (status != TINY_STORE_OK)
+ {
+ tiny_message_free(&message);
+ tiny_message_free(&indexed);
+ set_error(store, "failed to append and flush a PUT record");
+ return status;
+ }
+ store->last_sequence++;
+ memset(&store->entries[store->entry_count], 0, sizeof(store->entries[store->entry_count]));
+ store->entries[store->entry_count].message = indexed;
+ store->entries[store->entry_count].sequence = store->last_sequence;
+ store->entry_count++;
+ *written = message;
+ return TINY_STORE_OK;
+}
+
+tiny_store_status_t
+tiny_store_delete(tiny_store_t *store, const char *message_id, const char *session_id)
+{
+ tiny_store_entry_t *entry = find_entry(store, message_id);
+ unsigned char payload[34];
+ tiny_store_status_t status;
+ if (!entry || entry->deleted
+ || (session_id && strcmp(entry->message.session_id, session_id) != 0))
+ {
+ return TINY_STORE_NOT_FOUND;
+ }
+ memcpy(payload, message_id, 26);
+ put_u64(payload + 26, now_milliseconds());
+ status = append_record_to(
+ store->log, RECORD_DEL, store->last_sequence + 1, payload, sizeof(payload), true);
+ if (status != TINY_STORE_OK)
+ {
+ set_error(store, "failed to append and flush a DEL record");
+ return status;
+ }
+ store->last_sequence++;
+ entry->deleted = true;
+ return TINY_STORE_OK;
+}
+
+tiny_store_status_t
+tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report)
+{
+ wchar_t *temporary = join_path(store->directory, L"active.tmp");
+ HANDLE file;
+ LARGE_INTEGER before, after;
+ size_t read_index, write_index = 0;
+ uint64_t removed = 0, max_sequence = 0;
+ tiny_store_status_t status = TINY_STORE_OK;
+ if (!temporary)
+ {
+ return TINY_STORE_NO_MEMORY;
+ }
+ GetFileSizeEx(store->log, &before);
+ file = CreateFileW(
+ temporary, GENERIC_WRITE | GENERIC_READ, 0, NULL, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
+ if (file == INVALID_HANDLE_VALUE)
+ {
+ free(temporary);
+ return TINY_STORE_IO_ERROR;
+ }
+ if (BCryptGenRandom(
+ NULL, store->generation, sizeof(store->generation), BCRYPT_USE_SYSTEM_PREFERRED_RNG)
+ != 0)
+ {
+ CloseHandle(file);
+ free(temporary);
+ return TINY_STORE_IO_ERROR;
+ }
+ store->has_generation = true;
+ status = append_record_to(
+ file, RECORD_MARKER, 0, store->generation, sizeof(store->generation), false);
+ for (read_index = 0; read_index < store->entry_count; ++read_index)
+ {
+ if (status == TINY_STORE_OK && !store->entries[read_index].deleted)
+ {
+ unsigned char *payload = NULL;
+ uint32_t length = 0;
+ if (!serialize_put(&store->entries[read_index].message, &payload, &length))
+ {
+ status = TINY_STORE_NO_MEMORY;
+ break;
+ }
+ status = append_record_to(
+ file, RECORD_PUT, store->entries[read_index].sequence, payload, length, false);
+ free(payload);
+ if (status != TINY_STORE_OK)
+ {
+ break;
+ }
+ if (store->entries[read_index].sequence > max_sequence)
+ {
+ max_sequence = store->entries[read_index].sequence;
+ }
+ }
+ else
+ {
+ removed++;
+ }
+ }
+ if (status == TINY_STORE_OK && !FlushFileBuffers(file))
+ {
+ status = TINY_STORE_IO_ERROR;
+ }
+ CloseHandle(file);
+ if (status == TINY_STORE_OK)
+ {
+ CloseHandle(store->log);
+ store->log = INVALID_HANDLE_VALUE;
+ if (!MoveFileExW(
+ temporary, store->log_path, MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH))
+ {
+ status = TINY_STORE_IO_ERROR;
+ }
+ store->log = CreateFileW(store->log_path,
+ GENERIC_READ | GENERIC_WRITE,
+ FILE_SHARE_READ,
+ NULL,
+ OPEN_ALWAYS,
+ FILE_ATTRIBUTE_NORMAL,
+ NULL);
+ if (store->log == INVALID_HANDLE_VALUE)
+ {
+ status = TINY_STORE_IO_ERROR;
+ }
+ }
+ if (status == TINY_STORE_OK)
+ {
+ for (read_index = 0; read_index < store->entry_count; ++read_index)
+ {
+ if (store->entries[read_index].deleted)
+ {
+ tiny_message_free(&store->entries[read_index].message);
+ }
+ else
+ {
+ if (write_index != read_index)
+ {
+ store->entries[write_index] = store->entries[read_index];
+ }
+ write_index++;
+ }
+ }
+ store->entry_count = write_index;
+ store->last_sequence = max_sequence;
+ GetFileSizeEx(store->log, &after);
+ report->messages_indexed = write_index;
+ report->records_removed = removed;
+ report->bytes_before = (uint64_t)before.QuadPart;
+ report->bytes_after = (uint64_t)after.QuadPart;
+ status = write_snapshot(store);
+ }
+ if (status != TINY_STORE_OK)
+ {
+ set_error(store,
+ "compaction did not complete; the previous valid log is retained when replacement "
+ "had not begun");
+ }
+ free(temporary);
+ return status;
+}
+
+void
+tiny_store_health(const tiny_store_t *store, tiny_store_health_t *health)
+{
+ LARGE_INTEGER size;
+ size_t index;
+ memset(health, 0, sizeof(*health));
+ if (GetFileSizeEx(store->log, &size))
+ {
+ health->log_size = (uint64_t)size.QuadPart;
+ }
+ {
+ HANDLE f = CreateFileW(store->snapshot_path,
+ GENERIC_READ,
+ FILE_SHARE_READ,
+ NULL,
+ OPEN_EXISTING,
+ FILE_ATTRIBUTE_NORMAL,
+ NULL);
+ if (f != INVALID_HANDLE_VALUE)
+ {
+ if (GetFileSizeEx(f, &size))
+ {
+ health->snapshot_size = (uint64_t)size.QuadPart;
+ }
+ CloseHandle(f);
+ }
+ }
+ for (index = 0; index < store->entry_count; ++index)
+ {
+ if (store->entries[index].deleted)
+ {
+ health->deleted_messages++;
+ }
+ else
+ {
+ health->active_messages++;
+ }
+ }
+ health->last_sequence = store->last_sequence;
+ health->recovered_tail = store->recovered_tail;
+ health->snapshot_valid = store->snapshot_valid;
+ strncpy_s(health->last_error, sizeof(health->last_error), store->last_error, _TRUNCATE);
+}
+
+size_t
+tiny_store_entry_count(const tiny_store_t *store)
+{
+ return store->entry_count;
+}
+const tiny_store_entry_t *
+tiny_store_entry(const tiny_store_t *store, size_t index)
+{
+ return index < store->entry_count ? &store->entries[index] : NULL;
+}
diff --git a/src/storage/store.h b/src/storage/store.h
new file mode 100644
index 0000000..cc14a80
--- /dev/null
+++ b/src/storage/store.h
@@ -0,0 +1,64 @@
+#ifndef TINY_CONVMEM_STORAGE_STORE_H
+#define TINY_CONVMEM_STORAGE_STORE_H
+
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdint.h>
+#include <wchar.h>
+
+#include "domain/message.h"
+
+typedef enum tiny_store_status
+{
+ TINY_STORE_OK = 0,
+ TINY_STORE_NOT_FOUND,
+ TINY_STORE_BUSY,
+ TINY_STORE_IO_ERROR,
+ TINY_STORE_CORRUPT,
+ TINY_STORE_NO_MEMORY
+} tiny_store_status_t;
+
+typedef struct tiny_store_entry
+{
+ tiny_message_t message;
+ bool deleted;
+ uint64_t sequence;
+} tiny_store_entry_t;
+
+typedef struct tiny_store tiny_store_t;
+
+typedef struct tiny_rebuild_report
+{
+ uint64_t messages_indexed;
+ uint64_t records_removed;
+ uint64_t bytes_before;
+ uint64_t bytes_after;
+} tiny_rebuild_report_t;
+
+typedef struct tiny_store_health
+{
+ uint64_t log_size;
+ uint64_t snapshot_size;
+ uint64_t active_messages;
+ uint64_t deleted_messages;
+ uint64_t last_sequence;
+ bool recovered_tail;
+ bool snapshot_valid;
+ char last_error[160];
+} tiny_store_health_t;
+
+tiny_store_status_t tiny_store_open(tiny_store_t **store, const wchar_t *directory);
+void tiny_store_close(tiny_store_t *store);
+const char *tiny_store_status_name(tiny_store_status_t status);
+
+tiny_store_status_t
+tiny_store_write(tiny_store_t *store, const tiny_message_t *input, tiny_message_t *written);
+tiny_store_status_t
+tiny_store_delete(tiny_store_t *store, const char *message_id, const char *session_id);
+tiny_store_status_t tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report);
+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);
+const tiny_store_entry_t *tiny_store_entry(const tiny_store_t *store, size_t index);
+
+#endif
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-19 17:16:09 +0000