diff options
Diffstat (limited to 'src/storage/store.c')
| -rw-r--r-- | src/storage/store.c | 1111 |
1 files changed, 1111 insertions, 0 deletions
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; +} |