diff options
Diffstat (limited to 'src/storage')
| -rw-r--r-- | src/storage/README.md | 3 | ||||
| -rw-r--r-- | src/storage/codec.c | 76 | ||||
| -rw-r--r-- | src/storage/codec.h | 13 | ||||
| -rw-r--r-- | src/storage/store.c | 647 | ||||
| -rw-r--r-- | src/storage/store.h | 3 |
5 files changed, 398 insertions, 344 deletions
diff --git a/src/storage/README.md b/src/storage/README.md index 3c03dba..79874b0 100644 --- a/src/storage/README.md +++ b/src/storage/README.md @@ -2,7 +2,8 @@ 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`. +compaction. It depends only on the UTF-8, opaque-handle API in `src/sys`; OS +file and durability primitives do not leak into this layer. The binary format is private but versioned. Invalid complete records are never skipped; only an incomplete EOF record is truncated. Snapshot generation must diff --git a/src/storage/codec.c b/src/storage/codec.c new file mode 100644 index 0000000..af061ce --- /dev/null +++ b/src/storage/codec.c @@ -0,0 +1,76 @@ +#include "storage/codec.h" + +void +tiny_codec_put_u32(unsigned char *output, uint32_t value) +{ + if (output == NULL) + { + return; + } + output[0] = (unsigned char)value; + output[1] = (unsigned char)(value >> 8); + output[2] = (unsigned char)(value >> 16); + output[3] = (unsigned char)(value >> 24); +} + +void +tiny_codec_put_u64(unsigned char *output, uint64_t value) +{ + unsigned index; + if (output == NULL) + { + return; + } + for (index = 0; index < 8; ++index) + { + output[index] = (unsigned char)(value >> (index * 8)); + } +} + +uint32_t +tiny_codec_get_u32(const unsigned char *input) +{ + if (input == NULL) + { + return 0; + } + return (uint32_t)input[0] | ((uint32_t)input[1] << 8) | ((uint32_t)input[2] << 16) + | ((uint32_t)input[3] << 24); +} + +uint64_t +tiny_codec_get_u64(const unsigned char *input) +{ + uint64_t value = 0; + int index; + if (input == NULL) + { + return 0; + } + for (index = 7; index >= 0; --index) + { + value = (value << 8) | input[index]; + } + return value; +} + +uint32_t +tiny_codec_crc32(const unsigned char *data, size_t length) +{ + uint32_t crc = 0xffffffffU; + size_t index; + if (data == NULL && length != 0) + { + return 0; + } + for (index = 0; index < length; ++index) + { + unsigned bit; + crc ^= data[index]; + for (bit = 0; bit < 8; ++bit) + { + crc = (crc >> 1) ^ (0xedb88320U & (uint32_t)-(int32_t)(crc & 1)); + } + } + return ~crc; +} diff --git a/src/storage/codec.h b/src/storage/codec.h new file mode 100644 index 0000000..aafd174 --- /dev/null +++ b/src/storage/codec.h @@ -0,0 +1,13 @@ +#ifndef TINY_CONVMEM_STORAGE_CODEC_H +#define TINY_CONVMEM_STORAGE_CODEC_H + +#include <stddef.h> +#include <stdint.h> + +void tiny_codec_put_u32(unsigned char *output, uint32_t value); +void tiny_codec_put_u64(unsigned char *output, uint64_t value); +uint32_t tiny_codec_get_u32(const unsigned char *input); +uint64_t tiny_codec_get_u64(const unsigned char *input); +uint32_t tiny_codec_crc32(const unsigned char *data, size_t length); + +#endif diff --git a/src/storage/store.c b/src/storage/store.c index 9d26e0f..296f4d0 100644 --- a/src/storage/store.c +++ b/src/storage/store.c @@ -1,13 +1,13 @@ -#define WIN32_LEAN_AND_MEAN -#include <windows.h> -#include <bcrypt.h> - #include "storage/store.h" +#include "storage/codec.h" +#include "sys/system.h" #include <stdio.h> #include <stdlib.h> #include <string.h> +#include "base/allocator.h" + #define RECORD_HEADER_SIZE 24U #define RECORD_VERSION 1U #define RECORD_PUT 1U @@ -18,11 +18,11 @@ struct tiny_store { - HANDLE lock; - HANDLE log; - wchar_t *directory; - wchar_t *log_path; - wchar_t *snapshot_path; + tiny_sys_file_t lock; + tiny_sys_file_t log; + char *directory; + char *log_path; + char *snapshot_path; tiny_store_entry_t *entries; size_t entry_count; size_t entry_capacity; @@ -37,89 +37,19 @@ struct tiny_store }; 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); + snprintf(store->last_error, sizeof(store->last_error), "%s", message); } static bool -write_all(HANDLE file, const void *data, uint32_t length) +write_all(tiny_sys_file_t *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) + uint32_t written = 0; + if (!tiny_sys_file_write(file, position, length, &written) || written == 0) { return false; } @@ -130,14 +60,14 @@ write_all(HANDLE file, const void *data, uint32_t length) } static bool -read_all(HANDLE file, void *data, uint32_t length, uint32_t *actual) +read_all(tiny_sys_file_t *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)) + uint32_t count = 0; + if (!tiny_sys_file_read(file, position + total, length - total, &count)) { return false; } @@ -154,12 +84,7 @@ read_all(HANDLE file, void *data, uint32_t length, uint32_t *actual) 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; + return tiny_sys_now_milliseconds(); } static void @@ -171,11 +96,9 @@ make_ulid(tiny_store_t *store, uint64_t timestamp, char output[27]) 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) + if (!tiny_sys_random_bytes(store->last_random, sizeof(store->last_random))) { - uint64_t fallback = GetTickCount64() ^ timestamp; + uint64_t fallback = tiny_sys_monotonic_milliseconds() ^ timestamp; for (index = 0; index < sizeof(store->last_random); ++index) { store->last_random[index] = (unsigned char)(fallback >> ((index % 8) * 8)); @@ -274,19 +197,19 @@ serialize_put(const tiny_message_t *message, unsigned char **payload, uint32_t * return false; } offset = 0; - put_u64(data + offset, (uint64_t)message->timestamp_ms); + tiny_codec_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); + tiny_codec_put_u32(data + offset, (uint32_t)session_len); offset += 4; - put_u32(data + offset, (uint32_t)content_len); + tiny_codec_put_u32(data + offset, (uint32_t)content_len); offset += 4; - put_u32(data + offset, (uint32_t)tool_len); + tiny_codec_put_u32(data + offset, (uint32_t)tool_len); offset += 4; - put_u32(data + offset, (uint32_t)message->metadata.tag_count); + tiny_codec_put_u32(data + offset, (uint32_t)message->metadata.tag_count); offset += 4; memcpy(data + offset, message->session_id, session_len); offset += session_len; @@ -300,7 +223,7 @@ serialize_put(const tiny_message_t *message, unsigned char **payload, uint32_t * 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); + tiny_codec_put_u32(data + offset, (uint32_t)tag_len); offset += 4; memcpy(data + offset, message->metadata.tags[index], tag_len); offset += tag_len; @@ -323,6 +246,13 @@ copy_field(const unsigned char *data, size_t length) } static bool +deserialize_failed(tiny_message_t *message) +{ + tiny_message_free(message); + return false; +} + +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; @@ -332,20 +262,20 @@ deserialize_put(const unsigned char *data, uint32_t length, tiny_message_t *mess { return false; } - message->timestamp_ms = (int64_t)get_u64(data + offset); + message->timestamp_ms = (int64_t)tiny_codec_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); + session_len = tiny_codec_get_u32(data + offset); offset += 4; - content_len = get_u32(data + offset); + content_len = tiny_codec_get_u32(data + offset); offset += 4; - tool_len = get_u32(data + offset); + tool_len = tiny_codec_get_u32(data + offset); offset += 4; - tag_count = get_u32(data + offset); + tag_count = tiny_codec_get_u32(data + offset); offset += 4; if (!tiny_message_id_valid(message->message_id) || (message->role < TINY_ROLE_USER || message->role > TINY_ROLE_SYSTEM) @@ -369,33 +299,33 @@ deserialize_put(const unsigned char *data, uint32_t length, tiny_message_t *mess if (message->session_id == NULL || message->content == NULL || (tool_len && !message->metadata.tool_call_id)) { - goto fail; + return deserialize_failed(message); } if (tag_count > 0) { message->metadata.tags = calloc(tag_count, sizeof(char *)); if (message->metadata.tags == NULL) { - goto fail; + return deserialize_failed(message); } for (index = 0; index < tag_count; ++index) { uint32_t tag_len; if (offset + 4 > length) { - goto fail; + return deserialize_failed(message); } - tag_len = get_u32(data + offset); + tag_len = tiny_codec_get_u32(data + offset); offset += 4; if (tag_len == 0 || tag_len > TINY_TAG_MAX || offset + tag_len > length) { - goto fail; + return deserialize_failed(message); } message->metadata.tags[index] = copy_field(data + offset, tag_len); offset += tag_len; if (message->metadata.tags[index] == NULL) { - goto fail; + return deserialize_failed(message); } message->metadata.tag_count++; } @@ -403,16 +333,13 @@ deserialize_put(const unsigned char *data, uint32_t length, tiny_message_t *mess if (offset != length || !tiny_utf8_valid(message->session_id, session_len) || !tiny_utf8_valid(message->content, content_len)) { - goto fail; + return deserialize_failed(message); } return true; -fail: - tiny_message_free(message); - return false; } static tiny_store_status_t -append_record_to(HANDLE file, +append_record_to(tiny_sys_file_t *file, uint8_t type, uint64_t sequence, const unsigned char *payload, @@ -420,16 +347,14 @@ append_record_to(HANDLE file, 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))) + tiny_codec_put_u32(header + 8, payload_length); + tiny_codec_put_u64(header + 12, sequence); + tiny_codec_put_u32(header + 20, tiny_codec_crc32(payload, payload_length)); + if (!tiny_sys_file_seek(file, 0, TINY_SYS_SEEK_END) || !write_all(file, header, sizeof(header)) + || !write_all(file, payload, payload_length) || (flush && !tiny_sys_file_flush(file))) { return TINY_STORE_IO_ERROR; } @@ -469,38 +394,51 @@ clear_entries(tiny_store_t *store) } static tiny_store_status_t +recover_torn_tail(tiny_store_t *store, uint64_t boundary) +{ + if (boundary > INT64_MAX + || !tiny_sys_file_seek(&store->log, (int64_t)boundary, TINY_SYS_SEEK_BEGIN) + || !tiny_sys_file_truncate(&store->log) || !tiny_sys_file_flush(&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 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)) + uint64_t file_size, boundary = starting_offset; + if (starting_offset > INT64_MAX + || !tiny_sys_file_seek(&store->log, (int64_t)starting_offset, TINY_SYS_SEEK_BEGIN) + || !tiny_sys_file_size(&store->log, &file_size)) { return TINY_STORE_IO_ERROR; } - while (boundary.QuadPart < file_size.QuadPart) + while (boundary < file_size) { 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)) + if (!read_all(&store->log, header, sizeof(header), &actual)) { return TINY_STORE_IO_ERROR; } if (actual < sizeof(header)) { - goto torn_tail; + return recover_torn_tail(store, boundary); } 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); + payload_length = tiny_codec_get_u32(header + 8); + sequence = tiny_codec_get_u64(header + 12); + expected_crc = tiny_codec_get_u32(header + 20); if (payload_length > RECORD_PAYLOAD_MAX || (header[5] != RECORD_MARKER && sequence <= store->last_sequence)) { @@ -511,7 +449,7 @@ replay_log(tiny_store_t *store, uint64_t starting_offset) { return TINY_STORE_NO_MEMORY; } - if (!read_all(store->log, payload, payload_length, &actual)) + if (!read_all(&store->log, payload, payload_length, &actual)) { free(payload); return TINY_STORE_IO_ERROR; @@ -519,53 +457,62 @@ replay_log(tiny_store_t *store, uint64_t starting_offset) if (actual < payload_length) { free(payload); - goto torn_tail; + return recover_torn_tail(store, boundary); } - if (crc32_bytes(payload, payload_length) != expected_crc) + if (tiny_codec_crc32(payload, payload_length) != expected_crc) { free(payload); return TINY_STORE_CORRUPT; } - if (header[5] == RECORD_MARKER) + switch (header[5]) { - if (boundary.QuadPart != 0 || sequence != 0 - || payload_length != sizeof(store->generation)) + case RECORD_MARKER: + if (boundary != 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; + break; + + case 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); + } + break; + + case RECORD_DEL: + { + 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; + } + break; + + default: + free(payload); + return TINY_STORE_CORRUPT; } free(payload); if (status != TINY_STORE_OK) @@ -576,49 +523,43 @@ replay_log(tiny_store_t *store, uint64_t starting_offset) { store->last_sequence = sequence; } - boundary.QuadPart += RECORD_HEADER_SIZE + payload_length; + boundary += 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; + char *temporary = tiny_sys_path_join(store->directory, "snapshot.tmp"); + tiny_sys_file_t file; unsigned char data[SNAPSHOT_HEADER_SIZE] = { 0 }; - LARGE_INTEGER size; + uint64_t 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) + if (tiny_sys_file_open(&file, temporary, TINY_SYS_FILE_WRITE_NEW) != TINY_SYS_OK) { free(temporary); return TINY_STORE_IO_ERROR; } - GetFileSizeEx(store->log, &size); + if (!tiny_sys_file_size(&store->log, &size)) + { + tiny_sys_file_close(&file); + free(temporary); + return TINY_STORE_IO_ERROR; + } 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)); + tiny_codec_put_u64(data + 24, store->last_sequence); + tiny_codec_put_u64(data + 32, size); + tiny_codec_put_u64(data + 40, (uint64_t)store->entry_count); + tiny_codec_put_u32(data + 48, tiny_codec_crc32(data, 48)); + ok = write_all(&file, data, sizeof(data)); for (index = 0; ok && index < store->entry_count; ++index) { unsigned char prefix[16], *payload = NULL; @@ -628,22 +569,20 @@ write_snapshot(tiny_store_t *store) 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); + tiny_codec_put_u64(prefix, store->entries[index].sequence); + tiny_codec_put_u32(prefix + 8, length); + tiny_codec_put_u32(prefix + 12, tiny_codec_crc32(payload, length)); + ok = write_all(&file, prefix, sizeof(prefix)) && write_all(&file, payload, length); free(payload); } if (ok) { - ok = FlushFileBuffers(file) != 0; + ok = tiny_sys_file_flush(&file); } - CloseHandle(file); + tiny_sys_file_close(&file); if (ok) { - ok = MoveFileExW( - temporary, store->snapshot_path, MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH) - != 0; + ok = tiny_sys_path_replace(temporary, store->snapshot_path); } free(temporary); store->snapshot_valid = ok; @@ -655,13 +594,12 @@ 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) + if (!tiny_sys_file_seek(&store->log, 0, TINY_SYS_SEEK_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))) + || tiny_codec_get_u64(header + 12) != 0 || tiny_codec_get_u32(header + 8) != 16 + || !read_all(&store->log, payload, sizeof(payload), &actual) || actual != sizeof(payload) + || tiny_codec_get_u32(header + 20) != tiny_codec_crc32(payload, sizeof(payload))) { return false; } @@ -669,36 +607,43 @@ log_generation_matches(tiny_store_t *store, const unsigned char generation[16]) } static bool +snapshot_invalid(tiny_store_t *store, tiny_sys_file_t *file, bool clear_loaded_entries) +{ + if (clear_loaded_entries) + { + clear_entries(store); + store->has_generation = false; + } + set_error(store, "snapshot validation failed; the active log was replayed from the beginning"); + tiny_sys_file_close(file); + return false; +} + +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); + tiny_sys_file_t file; 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) + uint64_t log_size; + if (tiny_sys_file_open(&file, store->snapshot_path, TINY_SYS_FILE_READ) != TINY_SYS_OK) { return false; } - if (!read_all(file, header, sizeof(header), &actual) || actual != sizeof(header) + 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)) + || tiny_codec_get_u32(header + 48) != tiny_codec_crc32(header, 48) + || !tiny_sys_file_size(&store->log, &log_size)) { - goto invalid; + return snapshot_invalid(store, &file, false); } - 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)) + expected_sequence = tiny_codec_get_u64(header + 24); + offset = tiny_codec_get_u64(header + 32); + count = tiny_codec_get_u64(header + 40); + if (offset > log_size || count > SIZE_MAX || !log_generation_matches(store, header + 8)) { - goto invalid; + return snapshot_invalid(store, &file, false); } memcpy(store->generation, header + 8, 16); store->has_generation = true; @@ -708,99 +653,102 @@ load_snapshot(tiny_store_t *store, uint64_t *log_offset) uint32_t length, crc; tiny_message_t message; tiny_store_status_t status; - if (!read_all(file, prefix, sizeof(prefix), &actual) || actual != sizeof(prefix)) + if (!read_all(&file, prefix, sizeof(prefix), &actual) || actual != sizeof(prefix)) { - goto invalid_entries; + return snapshot_invalid(store, &file, true); } - length = get_u32(prefix + 8); - crc = get_u32(prefix + 12); + length = tiny_codec_get_u32(prefix + 8); + crc = tiny_codec_get_u32(prefix + 12); if (length > RECORD_PAYLOAD_MAX) { - goto invalid_entries; + return snapshot_invalid(store, &file, true); } payload = malloc(length ? length : 1); if (!payload) { - goto invalid_entries; + return snapshot_invalid(store, &file, true); } - if (!read_all(file, payload, length, &actual) || actual != length - || crc32_bytes(payload, length) != crc || !deserialize_put(payload, length, &message)) + if (!read_all(&file, payload, length, &actual) || actual != length + || tiny_codec_crc32(payload, length) != crc + || !deserialize_put(payload, length, &message)) { free(payload); - goto invalid_entries; + return snapshot_invalid(store, &file, true); } free(payload); - status = apply_put(store, &message, get_u64(prefix)); + status = apply_put(store, &message, tiny_codec_get_u64(prefix)); tiny_message_free(&message); if (status != TINY_STORE_OK) { - goto invalid_entries; + return snapshot_invalid(store, &file, true); } - store->last_sequence = get_u64(prefix); + store->last_sequence = tiny_codec_get_u64(prefix); } if (store->last_sequence != expected_sequence) { - goto invalid_entries; + return snapshot_invalid(store, &file, true); } - CloseHandle(file); + tiny_sys_file_close(&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; +} + +static tiny_store_status_t +store_open_failed(tiny_store_t *store, char *lock_path, tiny_store_status_t status) +{ + free(lock_path); + tiny_store_close(store); + return status; } tiny_store_status_t -tiny_store_open(tiny_store_t **output, const wchar_t *directory) +tiny_store_open(tiny_store_t **output, const char *directory) { - tiny_store_t *store = calloc(1, sizeof(*store)); - wchar_t *lock_path = NULL; + tiny_store_t *store; + char *lock_path = NULL; + tiny_sys_status_t sys_status; tiny_store_status_t status; + if (output == NULL) + { + return TINY_STORE_IO_ERROR; + } + *output = NULL; + if (directory == NULL || directory[0] == '\0') + { + return TINY_STORE_IO_ERROR; + } + store = calloc(1, sizeof(*store)); 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"); + tiny_sys_file_init(&store->lock); + tiny_sys_file_init(&store->log); + store->directory = copy_field((const unsigned char *)directory, strlen(directory)); + store->log_path = tiny_sys_path_join(directory, "active.log"); + store->snapshot_path = tiny_sys_path_join(directory, "snapshot.bin"); + lock_path = tiny_sys_path_join(directory, "store.lock"); if (!store->directory || !store->log_path || !store->snapshot_path || !lock_path) { - status = TINY_STORE_NO_MEMORY; - goto fail; + return store_open_failed(store, lock_path, TINY_STORE_NO_MEMORY); } - if (!CreateDirectoryW(directory, NULL) && GetLastError() != ERROR_ALREADY_EXISTS) + if (tiny_sys_directory_create(directory) != TINY_SYS_OK) { - status = TINY_STORE_IO_ERROR; - goto fail; + return store_open_failed(store, lock_path, TINY_STORE_IO_ERROR); } - store->lock = CreateFileW( - lock_path, GENERIC_READ | GENERIC_WRITE, 0, NULL, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL); + sys_status = tiny_sys_file_open(&store->lock, lock_path, TINY_SYS_FILE_LOCK); free(lock_path); lock_path = NULL; - if (store->lock == INVALID_HANDLE_VALUE) + if (sys_status != TINY_SYS_OK) { - status = GetLastError() == ERROR_SHARING_VIOLATION ? TINY_STORE_BUSY : TINY_STORE_IO_ERROR; - goto fail; + status = sys_status == TINY_SYS_BUSY ? TINY_STORE_BUSY : TINY_STORE_IO_ERROR; + return store_open_failed(store, lock_path, status); } - 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) + if (tiny_sys_file_open(&store->log, store->log_path, TINY_SYS_FILE_READ_WRITE_CREATE) + != TINY_SYS_OK) { - status = TINY_STORE_IO_ERROR; - goto fail; + return store_open_failed(store, lock_path, TINY_STORE_IO_ERROR); } { uint64_t replay_offset = 0; @@ -809,14 +757,10 @@ tiny_store_open(tiny_store_t **output, const wchar_t *directory) } if (status != TINY_STORE_OK) { - goto fail; + return store_open_failed(store, lock_path, status); } *output = store; return TINY_STORE_OK; -fail: - free(lock_path); - tiny_store_close(store); - return status; } void @@ -827,14 +771,8 @@ tiny_store_close(tiny_store_t *store) { return; } - if (store->log != INVALID_HANDLE_VALUE) - { - CloseHandle(store->log); - } - if (store->lock != INVALID_HANDLE_VALUE) - { - CloseHandle(store->lock); - } + tiny_sys_file_close(&store->log); + tiny_sys_file_close(&store->lock); for (index = 0; index < store->entry_count; ++index) { tiny_message_free(&store->entries[index].message); @@ -873,6 +811,15 @@ tiny_store_write(tiny_store_t *store, const tiny_message_t *input, tiny_message_ unsigned char *payload = NULL; uint32_t length = 0; tiny_store_status_t status; + if (written != NULL) + { + tiny_message_init(written); + } + if (store == NULL || input == NULL || written == NULL || input->session_id == NULL + || input->content == NULL) + { + return TINY_STORE_IO_ERROR; + } tiny_message_init(&message); tiny_message_init(&indexed); if (tiny_message_copy(&message, input) != 0) @@ -890,7 +837,7 @@ tiny_store_write(tiny_store_t *store, const tiny_message_t *input, tiny_message_ return TINY_STORE_NO_MEMORY; } status - = append_record_to(store->log, RECORD_PUT, store->last_sequence + 1, payload, length, true); + = append_record_to(&store->log, RECORD_PUT, store->last_sequence + 1, payload, length, true); free(payload); if (status != TINY_STORE_OK) { @@ -911,18 +858,23 @@ tiny_store_write(tiny_store_t *store, const tiny_message_t *input, tiny_message_ 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); + tiny_store_entry_t *entry; unsigned char payload[34]; tiny_store_status_t status; + if (store == NULL || message_id == NULL || !tiny_message_id_valid(message_id)) + { + return TINY_STORE_NOT_FOUND; + } + entry = find_entry(store, message_id); 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()); + tiny_codec_put_u64(payload + 26, now_milliseconds()); status = append_record_to( - store->log, RECORD_DEL, store->last_sequence + 1, payload, sizeof(payload), true); + &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"); @@ -936,35 +888,45 @@ tiny_store_delete(tiny_store_t *store, const char *message_id, const char *sessi 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; + char *temporary; + tiny_sys_file_t file; + uint64_t before = 0, after = 0; size_t read_index, write_index = 0; uint64_t removed = 0, max_sequence = 0; tiny_store_status_t status = TINY_STORE_OK; + unsigned char previous_generation[sizeof(store->generation)]; + bool previous_has_generation; + bool log_replaced = false; + if (report != NULL) + { + memset(report, 0, sizeof(*report)); + } + if (store == NULL || report == NULL) + { + return TINY_STORE_IO_ERROR; + } + memcpy(previous_generation, store->generation, sizeof(previous_generation)); + previous_has_generation = store->has_generation; + temporary = tiny_sys_path_join(store->directory, "active.tmp"); 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) + if (!tiny_sys_file_size(&store->log, &before) + || tiny_sys_file_open(&file, temporary, TINY_SYS_FILE_WRITE_NEW) != TINY_SYS_OK) { free(temporary); return TINY_STORE_IO_ERROR; } - if (BCryptGenRandom( - NULL, store->generation, sizeof(store->generation), BCRYPT_USE_SYSTEM_PREFERRED_RNG) - != 0) + if (!tiny_sys_random_bytes(store->generation, sizeof(store->generation))) { - CloseHandle(file); + tiny_sys_file_close(&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); + &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) @@ -977,7 +939,7 @@ tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report) break; } status = append_record_to( - file, RECORD_PUT, store->entries[read_index].sequence, payload, length, false); + &file, RECORD_PUT, store->entries[read_index].sequence, payload, length, false); free(payload); if (status != TINY_STORE_OK) { @@ -993,28 +955,24 @@ tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report) removed++; } } - if (status == TINY_STORE_OK && !FlushFileBuffers(file)) + if (status == TINY_STORE_OK && !tiny_sys_file_flush(&file)) { status = TINY_STORE_IO_ERROR; } - CloseHandle(file); + tiny_sys_file_close(&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)) + tiny_sys_file_close(&store->log); + if (!tiny_sys_path_replace(temporary, store->log_path)) { 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) + else + { + log_replaced = true; + } + if (tiny_sys_file_open(&store->log, store->log_path, TINY_SYS_FILE_READ_WRITE_CREATE) + != TINY_SYS_OK) { status = TINY_STORE_IO_ERROR; } @@ -1038,15 +996,20 @@ tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report) } store->entry_count = write_index; store->last_sequence = max_sequence; - GetFileSizeEx(store->log, &after); + (void)tiny_sys_file_size(&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; + report->bytes_before = before; + report->bytes_after = after; status = write_snapshot(store); } if (status != TINY_STORE_OK) { + if (!log_replaced) + { + memcpy(store->generation, previous_generation, sizeof(store->generation)); + store->has_generation = previous_has_generation; + } set_error(store, "compaction did not complete; the previous valid log is retained when replacement " "had not begun"); @@ -1058,28 +1021,30 @@ 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) { - LARGE_INTEGER size; + uint64_t size; size_t index; + if (health == NULL) + { + return; + } memset(health, 0, sizeof(*health)); - if (GetFileSizeEx(store->log, &size)) + if (store == NULL) + { + return; + } + if (tiny_sys_file_size(&store->log, &size)) { - health->log_size = (uint64_t)size.QuadPart; + health->log_size = size; } { - HANDLE f = CreateFileW(store->snapshot_path, - GENERIC_READ, - FILE_SHARE_READ, - NULL, - OPEN_EXISTING, - FILE_ATTRIBUTE_NORMAL, - NULL); - if (f != INVALID_HANDLE_VALUE) + tiny_sys_file_t file; + if (tiny_sys_file_open(&file, store->snapshot_path, TINY_SYS_FILE_READ) == TINY_SYS_OK) { - if (GetFileSizeEx(f, &size)) + if (tiny_sys_file_size(&file, &size)) { - health->snapshot_size = (uint64_t)size.QuadPart; + health->snapshot_size = size; } - CloseHandle(f); + tiny_sys_file_close(&file); } } for (index = 0; index < store->entry_count; ++index) @@ -1096,16 +1061,16 @@ tiny_store_health(const tiny_store_t *store, tiny_store_health_t *health) 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); + snprintf(health->last_error, sizeof(health->last_error), "%s", store->last_error); } size_t tiny_store_entry_count(const tiny_store_t *store) { - return store->entry_count; + return store == NULL ? 0 : 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; + return store != NULL && index < store->entry_count ? &store->entries[index] : NULL; } diff --git a/src/storage/store.h b/src/storage/store.h index cc14a80..619f86f 100644 --- a/src/storage/store.h +++ b/src/storage/store.h @@ -4,7 +4,6 @@ #include <stdbool.h> #include <stddef.h> #include <stdint.h> -#include <wchar.h> #include "domain/message.h" @@ -47,7 +46,7 @@ typedef struct tiny_store_health char last_error[160]; } tiny_store_health_t; -tiny_store_status_t tiny_store_open(tiny_store_t **store, const wchar_t *directory); +tiny_store_status_t tiny_store_open(tiny_store_t **store, const char *directory); void tiny_store_close(tiny_store_t *store); const char *tiny_store_status_name(tiny_store_status_t status); |