diff options
Diffstat (limited to 'src/storage')
| -rw-r--r-- | src/storage/store.c | 856 | ||||
| -rw-r--r-- | src/storage/store.h | 25 |
2 files changed, 873 insertions, 8 deletions
diff --git a/src/storage/store.c b/src/storage/store.c index 296f4d0..d798d86 100644 --- a/src/storage/store.c +++ b/src/storage/store.c @@ -9,10 +9,14 @@ #include "base/allocator.h" #define RECORD_HEADER_SIZE 24U -#define RECORD_VERSION 1U +#define RECORD_VERSION 2U +#define RECORD_VERSION_LEGACY 1U #define RECORD_PUT 1U #define RECORD_DEL 2U #define RECORD_MARKER 3U +#define RECORD_UPDATE 4U +#define UPDATE_CONTENT 1U +#define UPDATE_METADATA 2U #define RECORD_PAYLOAD_MAX (TINY_CONTENT_MAX + 16384U) #define SNAPSHOT_HEADER_SIZE 52U @@ -27,6 +31,8 @@ struct tiny_store size_t entry_count; size_t entry_capacity; uint64_t last_sequence; + uint64_t update_records; + uint64_t dead_bytes; uint64_t last_ulid_ms; unsigned char last_random[10]; unsigned char generation[16]; @@ -180,6 +186,7 @@ serialize_put(const tiny_message_t *message, unsigned char **payload, uint32_t * { 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 summary_len = message->metadata.summary ? strlen(message->metadata.summary) : 0; size_t total = 8 + 26 + 2 + 16 + session_len + content_len + tool_len; size_t index, offset; unsigned char *data; @@ -187,6 +194,8 @@ serialize_put(const tiny_message_t *message, unsigned char **payload, uint32_t * { total += 4 + strlen(message->metadata.tags[index]); } + total += 32 + summary_len + message->metadata.relation_count * 27 + + message->metadata.source_count * 26; if (total > UINT32_MAX || total > RECORD_PAYLOAD_MAX) { return false; @@ -228,6 +237,36 @@ serialize_put(const tiny_message_t *message, unsigned char **payload, uint32_t * memcpy(data + offset, message->metadata.tags[index], tag_len); offset += tag_len; } + memcpy(data + offset, "M2", 2); + offset += 2; + tiny_codec_put_u64(data + offset, message->revision); + offset += 8; + tiny_codec_put_u64(data + offset, (uint64_t)message->updated_at_ms); + offset += 8; + data[offset++] = (unsigned char)message->metadata.memory_kind; + data[offset++] = message->metadata.importance; + tiny_codec_put_u32(data + offset, (uint32_t)summary_len); + offset += 4; + tiny_codec_put_u32(data + offset, (uint32_t)message->metadata.relation_count); + offset += 4; + tiny_codec_put_u32(data + offset, (uint32_t)message->metadata.source_count); + offset += 4; + if (summary_len > 0) + { + memcpy(data + offset, message->metadata.summary, summary_len); + offset += summary_len; + } + for (index = 0; index < message->metadata.relation_count; ++index) + { + data[offset++] = (unsigned char)message->metadata.relations[index].type; + memcpy(data + offset, message->metadata.relations[index].message_id, 26); + offset += 26; + } + for (index = 0; index < message->metadata.source_count; ++index) + { + memcpy(data + offset, message->metadata.source_message_ids[index], 26); + offset += 26; + } *payload = data; *length = (uint32_t)total; return true; @@ -330,12 +369,95 @@ deserialize_put(const unsigned char *data, uint32_t length, tiny_message_t *mess message->metadata.tag_count++; } } - if (offset != length || !tiny_utf8_valid(message->session_id, session_len) + if (!tiny_utf8_valid(message->session_id, session_len) || !tiny_utf8_valid(message->content, content_len)) { return deserialize_failed(message); } - return true; + if (offset == length) + { + message->updated_at_ms = message->timestamp_ms; + return true; + } + { + uint32_t summary_len, relation_count, source_count; + if (length - offset < 32 || memcmp(data + offset, "M2", 2) != 0) + { + return deserialize_failed(message); + } + offset += 2; + message->revision = tiny_codec_get_u64(data + offset); + offset += 8; + message->updated_at_ms = (int64_t)tiny_codec_get_u64(data + offset); + offset += 8; + message->metadata.memory_kind = (tiny_memory_kind_t)data[offset++]; + message->metadata.importance = data[offset++]; + summary_len = tiny_codec_get_u32(data + offset); + offset += 4; + relation_count = tiny_codec_get_u32(data + offset); + offset += 4; + source_count = tiny_codec_get_u32(data + offset); + offset += 4; + if (message->revision == 0 || message->metadata.memory_kind < TINY_KIND_CONVERSATION + || message->metadata.memory_kind > TINY_KIND_SUMMARY || summary_len > TINY_SUMMARY_MAX + || relation_count > TINY_RELATION_COUNT_MAX || source_count > TINY_SOURCE_COUNT_MAX + || offset + summary_len + (size_t)relation_count * 27 + (size_t)source_count * 26 != length) + { + return deserialize_failed(message); + } + if (summary_len > 0) + { + message->metadata.summary = copy_field(data + offset, summary_len); + if (!message->metadata.summary || !tiny_utf8_valid(message->metadata.summary, summary_len)) + { + return deserialize_failed(message); + } + offset += summary_len; + } + if (relation_count > 0) + { + message->metadata.relations = calloc(relation_count, sizeof(tiny_relation_t)); + if (!message->metadata.relations) + { + return deserialize_failed(message); + } + for (index = 0; index < relation_count; ++index) + { + tiny_relation_t *relation = &message->metadata.relations[index]; + relation->type = (tiny_relation_type_t)data[offset++]; + memcpy(relation->message_id, data + offset, 26); + relation->message_id[26] = '\0'; + offset += 26; + if (relation->type < TINY_RELATION_DEPENDS_ON + || relation->type > TINY_RELATION_RELATED_TO + || !tiny_message_id_valid(relation->message_id)) + { + return deserialize_failed(message); + } + message->metadata.relation_count++; + } + } + if (source_count > 0) + { + message->metadata.source_message_ids = calloc(source_count, sizeof(char *)); + if (!message->metadata.source_message_ids) + { + return deserialize_failed(message); + } + for (index = 0; index < source_count; ++index) + { + message->metadata.source_message_ids[index] = copy_field(data + offset, 26); + offset += 26; + if (!message->metadata.source_message_ids[index] + || !tiny_message_id_valid(message->metadata.source_message_ids[index])) + { + return deserialize_failed(message); + } + message->metadata.source_count++; + } + } + } + return offset == length; } static tiny_store_status_t @@ -381,6 +503,545 @@ apply_put(tiny_store_t *store, tiny_message_t *message, uint64_t sequence) return TINY_STORE_OK; } +static bool +string_equal(const char *left, const char *right) +{ + return left == right || (left && right && strcmp(left, right) == 0); +} + +static bool +metadata_patch_changes(const tiny_metadata_t *current, + const tiny_metadata_t *patch, + tiny_metadata_mask_t mask) +{ + tiny_metadata_t merged = *current; + if (mask & TINY_METADATA_TOOL_CALL_ID) + { + merged.tool_call_id = patch->tool_call_id; + } + if (mask & TINY_METADATA_TAGS) + { + merged.tags = patch->tags; + merged.tag_count = patch->tag_count; + } + if (mask & TINY_METADATA_TIER) + { + merged.memory_tier = patch->memory_tier; + } + if (mask & TINY_METADATA_KIND) + { + merged.memory_kind = patch->memory_kind; + } + if (mask & TINY_METADATA_IMPORTANCE) + { + merged.importance = patch->importance; + } + if (mask & TINY_METADATA_SUMMARY) + { + merged.summary = patch->summary; + } + if (mask & TINY_METADATA_RELATIONS) + { + merged.relations = patch->relations; + merged.relation_count = patch->relation_count; + } + if (mask & TINY_METADATA_SOURCES) + { + merged.source_message_ids = patch->source_message_ids; + merged.source_count = patch->source_count; + } + if (!string_equal(current->tool_call_id, merged.tool_call_id) + || current->tag_count != merged.tag_count || current->memory_tier != merged.memory_tier + || current->memory_kind != merged.memory_kind || current->importance != merged.importance + || !string_equal(current->summary, merged.summary) + || current->relation_count != merged.relation_count + || current->source_count != merged.source_count) + { + return true; + } + if (current->tag_count) + { + size_t index; + for (index = 0; index < current->tag_count; ++index) + { + if (strcmp(current->tags[index], merged.tags[index]) != 0) + { + return true; + } + } + } + if (current->relation_count + && memcmp(current->relations, + merged.relations, + current->relation_count * sizeof(*current->relations)) + != 0) + { + return true; + } + if (current->source_count) + { + size_t index; + for (index = 0; index < current->source_count; ++index) + { + if (strcmp(current->source_message_ids[index], merged.source_message_ids[index]) != 0) + { + return true; + } + } + } + return false; +} + +static bool +metadata_targets_valid(tiny_store_t *store, const char *self_id, const tiny_metadata_t *metadata) +{ + size_t index; + if (!tiny_metadata_valid(metadata)) + { + return false; + } + for (index = 0; index < metadata->relation_count; ++index) + { + tiny_store_entry_t *target; + if (self_id && strcmp(self_id, metadata->relations[index].message_id) == 0) + { + return false; + } + target = find_entry(store, metadata->relations[index].message_id); + if (!target || target->deleted) + { + return false; + } + } + for (index = 0; index < metadata->source_count; ++index) + { + tiny_store_entry_t *target; + if (self_id && strcmp(self_id, metadata->source_message_ids[index]) == 0) + { + return false; + } + target = find_entry(store, metadata->source_message_ids[index]); + if (!target || target->deleted) + { + return false; + } + } + return true; +} + +static tiny_store_status_t +prepare_update(tiny_store_t *store, + tiny_store_entry_t *entry, + const tiny_message_patch_t *patch, + uint64_t revision, + int64_t updated_at, + tiny_message_t *prepared) +{ + tiny_message_t merged = entry->message; + tiny_metadata_t metadata = entry->message.metadata; + if (patch->content) + { + merged.content = (char *)patch->content; + } + if (patch->metadata_mask & TINY_METADATA_TOOL_CALL_ID) + { + metadata.tool_call_id = patch->metadata->tool_call_id; + } + if (patch->metadata_mask & TINY_METADATA_TAGS) + { + metadata.tags = patch->metadata->tags; + metadata.tag_count = patch->metadata->tag_count; + } + if (patch->metadata_mask & TINY_METADATA_TIER) + { + metadata.memory_tier = patch->metadata->memory_tier; + } + if (patch->metadata_mask & TINY_METADATA_KIND) + { + metadata.memory_kind = patch->metadata->memory_kind; + } + if (patch->metadata_mask & TINY_METADATA_IMPORTANCE) + { + metadata.importance = patch->metadata->importance; + } + if (patch->metadata_mask & TINY_METADATA_SUMMARY) + { + metadata.summary = patch->metadata->summary; + } + if (patch->metadata_mask & TINY_METADATA_RELATIONS) + { + metadata.relations = patch->metadata->relations; + metadata.relation_count = patch->metadata->relation_count; + } + if (patch->metadata_mask & TINY_METADATA_SOURCES) + { + metadata.source_message_ids = patch->metadata->source_message_ids; + metadata.source_count = patch->metadata->source_count; + } + merged.metadata = metadata; + if (!merged.content || !merged.content[0] || strlen(merged.content) > TINY_CONTENT_MAX + || !tiny_utf8_valid(merged.content, strlen(merged.content)) + || !metadata_targets_valid(store, merged.message_id, &metadata)) + { + return TINY_STORE_CONFLICT; + } + if (tiny_message_copy(prepared, &merged) != 0) + { + return TINY_STORE_NO_MEMORY; + } + prepared->revision = revision; + prepared->updated_at_ms = updated_at; + return TINY_STORE_OK; +} + +static bool +delta_put_string(unsigned char *data, size_t capacity, size_t *offset, const char *value) +{ + size_t length = value ? strlen(value) : 0; + if (length > UINT32_MAX || *offset > capacity || capacity - *offset < 4 + length) + { + return false; + } + tiny_codec_put_u32(data + *offset, (uint32_t)length); + *offset += 4; + if (length) + { + memcpy(data + *offset, value, length); + *offset += length; + } + return true; +} + +static bool +serialize_update(const tiny_store_entry_t *entry, + const tiny_message_patch_t *patch, + uint64_t revision, + int64_t updated_at, + unsigned char **payload, + uint32_t *payload_length) +{ + size_t capacity = 64 + (patch->content ? strlen(patch->content) : 0), offset = 0, index; + unsigned char *data; + tiny_metadata_mask_t mask = patch->metadata_mask; + if (mask & TINY_METADATA_TOOL_CALL_ID) + { + capacity += 4 + (patch->metadata->tool_call_id ? strlen(patch->metadata->tool_call_id) : 0); + } + if (mask & TINY_METADATA_TAGS) + { + capacity += 4; + for (index = 0; index < patch->metadata->tag_count; ++index) + { + capacity += 4 + strlen(patch->metadata->tags[index]); + } + } + if (mask & TINY_METADATA_SUMMARY) + { + capacity += 4 + (patch->metadata->summary ? strlen(patch->metadata->summary) : 0); + } + if (mask & TINY_METADATA_RELATIONS) + { + capacity += 4 + patch->metadata->relation_count * 27; + } + if (mask & TINY_METADATA_SOURCES) + { + capacity += 4 + patch->metadata->source_count * 26; + } + capacity += (mask & TINY_METADATA_TIER) ? 1 : 0; + capacity += (mask & TINY_METADATA_KIND) ? 1 : 0; + capacity += (mask & TINY_METADATA_IMPORTANCE) ? 1 : 0; + if (capacity > RECORD_PAYLOAD_MAX || capacity > UINT32_MAX || !(data = malloc(capacity))) + { + return false; + } + memcpy(data, entry->message.message_id, 26); + offset = 26; + tiny_codec_put_u64(data + offset, revision); + offset += 8; + tiny_codec_put_u64(data + offset, (uint64_t)updated_at); + offset += 8; + data[offset++] = patch->content ? UPDATE_CONTENT : 0; + data[offset - 1] |= mask ? UPDATE_METADATA : 0; + tiny_codec_put_u32(data + offset, mask); + offset += 4; + if (patch->content && !delta_put_string(data, capacity, &offset, patch->content)) + { + free(data); + return false; + } + if ((mask & TINY_METADATA_TOOL_CALL_ID) + && !delta_put_string(data, capacity, &offset, patch->metadata->tool_call_id)) + { + free(data); + return false; + } + if (mask & TINY_METADATA_TAGS) + { + tiny_codec_put_u32(data + offset, (uint32_t)patch->metadata->tag_count); + offset += 4; + for (index = 0; index < patch->metadata->tag_count; ++index) + { + if (!delta_put_string(data, capacity, &offset, patch->metadata->tags[index])) + { + free(data); + return false; + } + } + } + if (mask & TINY_METADATA_TIER) + { + data[offset++] = (unsigned char)patch->metadata->memory_tier; + } + if (mask & TINY_METADATA_KIND) + { + data[offset++] = (unsigned char)patch->metadata->memory_kind; + } + if (mask & TINY_METADATA_IMPORTANCE) + { + data[offset++] = patch->metadata->importance; + } + if ((mask & TINY_METADATA_SUMMARY) + && !delta_put_string(data, capacity, &offset, patch->metadata->summary)) + { + free(data); + return false; + } + if (mask & TINY_METADATA_RELATIONS) + { + tiny_codec_put_u32(data + offset, (uint32_t)patch->metadata->relation_count); + offset += 4; + for (index = 0; index < patch->metadata->relation_count; ++index) + { + data[offset++] = (unsigned char)patch->metadata->relations[index].type; + memcpy(data + offset, patch->metadata->relations[index].message_id, 26); + offset += 26; + } + } + if (mask & TINY_METADATA_SOURCES) + { + tiny_codec_put_u32(data + offset, (uint32_t)patch->metadata->source_count); + offset += 4; + for (index = 0; index < patch->metadata->source_count; ++index) + { + memcpy(data + offset, patch->metadata->source_message_ids[index], 26); + offset += 26; + } + } + *payload = data; + *payload_length = (uint32_t)offset; + return true; +} + +static char * +delta_get_string(const unsigned char *data, size_t length, size_t *offset, size_t maximum) +{ + uint32_t field_length; + char *value; + if (*offset > length || length - *offset < 4) + { + return NULL; + } + field_length = tiny_codec_get_u32(data + *offset); + *offset += 4; + if (field_length > maximum || *offset > length || length - *offset < field_length) + { + return NULL; + } + value = copy_field(data + *offset, field_length); + *offset += field_length; + return value; +} + +static tiny_store_status_t +prepare_update_payload(tiny_store_t *store, + const unsigned char *payload, + uint32_t length, + tiny_store_entry_t **target, + tiny_message_t *prepared) +{ + tiny_message_t holder; + tiny_message_patch_t patch; + size_t offset = 0, index; + uint64_t revision; + int64_t updated_at; + uint8_t flags; + uint32_t mask; + char id[27]; + tiny_store_status_t status = TINY_STORE_CORRUPT; + tiny_message_init(&holder); + memset(&patch, 0, sizeof(patch)); +#define UPDATE_PAYLOAD_BAD() \ + do \ + { \ + tiny_message_free(&holder); \ + return TINY_STORE_CORRUPT; \ + } \ + while (0) + if (length < 47) + { + UPDATE_PAYLOAD_BAD(); + } + memcpy(id, payload, 26); + id[26] = '\0'; + offset = 26; + revision = tiny_codec_get_u64(payload + offset); + offset += 8; + updated_at = (int64_t)tiny_codec_get_u64(payload + offset); + offset += 8; + flags = payload[offset++]; + mask = tiny_codec_get_u32(payload + offset); + offset += 4; + *target = find_entry(store, id); + if (!*target || (*target)->deleted || revision != (*target)->message.revision + 1 + || revision > INT32_MAX || !updated_at || (mask & ~TINY_METADATA_ALL) + || flags != ((flags & UPDATE_CONTENT ? UPDATE_CONTENT : 0) | (mask ? UPDATE_METADATA : 0))) + { + UPDATE_PAYLOAD_BAD(); + } + if (flags & UPDATE_CONTENT) + { + holder.content = delta_get_string(payload, length, &offset, TINY_CONTENT_MAX); + if (!holder.content) + { + UPDATE_PAYLOAD_BAD(); + } + patch.content = holder.content; + } + patch.metadata = &holder.metadata; + patch.metadata_mask = mask; + if (mask & TINY_METADATA_TOOL_CALL_ID) + { + holder.metadata.tool_call_id + = delta_get_string(payload, length, &offset, TINY_TOOL_CALL_ID_MAX); + if (!holder.metadata.tool_call_id) + { + UPDATE_PAYLOAD_BAD(); + } + if (!holder.metadata.tool_call_id[0]) + { + free(holder.metadata.tool_call_id); + holder.metadata.tool_call_id = NULL; + } + } + if (mask & TINY_METADATA_TAGS) + { + uint32_t count; + if (offset > length || length - offset < 4) + { + UPDATE_PAYLOAD_BAD(); + } + count = tiny_codec_get_u32(payload + offset); + offset += 4; + if (count > TINY_TAG_COUNT_MAX + || (count && !(holder.metadata.tags = calloc(count, sizeof(char *))))) + { + UPDATE_PAYLOAD_BAD(); + } + for (index = 0; index < count; ++index) + { + holder.metadata.tags[index] = delta_get_string(payload, length, &offset, TINY_TAG_MAX); + if (!holder.metadata.tags[index] || !holder.metadata.tags[index][0]) + { + UPDATE_PAYLOAD_BAD(); + } + holder.metadata.tag_count++; + } + } + if (mask & TINY_METADATA_TIER) + { + if (offset >= length) + { + UPDATE_PAYLOAD_BAD(); + } + holder.metadata.memory_tier = (tiny_memory_tier_t)payload[offset++]; + } + if (mask & TINY_METADATA_KIND) + { + if (offset >= length) + { + UPDATE_PAYLOAD_BAD(); + } + holder.metadata.memory_kind = (tiny_memory_kind_t)payload[offset++]; + } + if (mask & TINY_METADATA_IMPORTANCE) + { + if (offset >= length) + { + UPDATE_PAYLOAD_BAD(); + } + holder.metadata.importance = payload[offset++]; + } + if (mask & TINY_METADATA_SUMMARY) + { + holder.metadata.summary = delta_get_string(payload, length, &offset, TINY_SUMMARY_MAX); + if (!holder.metadata.summary) + { + UPDATE_PAYLOAD_BAD(); + } + if (!holder.metadata.summary[0]) + { + free(holder.metadata.summary); + holder.metadata.summary = NULL; + } + } + if (mask & TINY_METADATA_RELATIONS) + { + uint32_t count; + if (offset > length || length - offset < 4) + { + UPDATE_PAYLOAD_BAD(); + } + count = tiny_codec_get_u32(payload + offset); + offset += 4; + if (count > TINY_RELATION_COUNT_MAX || offset > length || length - offset < (size_t)count * 27 + || (count && !(holder.metadata.relations = calloc(count, sizeof(tiny_relation_t))))) + { + UPDATE_PAYLOAD_BAD(); + } + for (index = 0; index < count; ++index) + { + holder.metadata.relations[index].type = (tiny_relation_type_t)payload[offset++]; + memcpy(holder.metadata.relations[index].message_id, payload + offset, 26); + holder.metadata.relations[index].message_id[26] = '\0'; + offset += 26; + holder.metadata.relation_count++; + } + } + if (mask & TINY_METADATA_SOURCES) + { + uint32_t count; + if (offset > length || length - offset < 4) + { + UPDATE_PAYLOAD_BAD(); + } + count = tiny_codec_get_u32(payload + offset); + offset += 4; + if (count > TINY_SOURCE_COUNT_MAX || offset > length || length - offset < (size_t)count * 26 + || (count && !(holder.metadata.source_message_ids = calloc(count, sizeof(char *))))) + { + UPDATE_PAYLOAD_BAD(); + } + for (index = 0; index < count; ++index) + { + holder.metadata.source_message_ids[index] = copy_field(payload + offset, 26); + if (!holder.metadata.source_message_ids[index]) + { + UPDATE_PAYLOAD_BAD(); + } + offset += 26; + holder.metadata.source_count++; + } + } + if (offset != length) + { + UPDATE_PAYLOAD_BAD(); + } + status = prepare_update(store, *target, &patch, revision, updated_at, prepared); + tiny_message_free(&holder); +#undef UPDATE_PAYLOAD_BAD + return status; +} + static void clear_entries(tiny_store_t *store) { @@ -431,8 +1092,10 @@ replay_log(tiny_store_t *store, uint64_t starting_offset) { 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)) + if (memcmp(header, "TCMR", 4) != 0 + || (header[4] != RECORD_VERSION && header[4] != RECORD_VERSION_LEGACY) + || (header[5] != RECORD_PUT && header[5] != RECORD_DEL && header[5] != RECORD_MARKER + && header[5] != RECORD_UPDATE)) { return TINY_STORE_CORRUPT; } @@ -507,9 +1170,28 @@ replay_log(tiny_store_t *store, uint64_t starting_offset) return TINY_STORE_CORRUPT; } entry->deleted = true; + store->dead_bytes += RECORD_HEADER_SIZE + payload_length; } break; + case RECORD_UPDATE: + { + tiny_store_entry_t *target = NULL; + tiny_message_t prepared; + tiny_message_init(&prepared); + status = prepare_update_payload(store, payload, payload_length, &target, &prepared); + if (status == TINY_STORE_OK) + { + tiny_message_free(&target->message); + target->message = prepared; + tiny_message_init(&prepared); + } + tiny_message_free(&prepared); + } + store->update_records++; + store->dead_bytes += RECORD_HEADER_SIZE + payload_length; + break; + default: free(payload); return TINY_STORE_CORRUPT; @@ -596,8 +1278,10 @@ log_generation_matches(tiny_store_t *store, const unsigned char generation[16]) uint32_t actual = 0; 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 - || tiny_codec_get_u64(header + 12) != 0 || tiny_codec_get_u32(header + 8) != 16 + || memcmp(header, "TCMR", 4) != 0 + || (header[4] != RECORD_VERSION && header[4] != RECORD_VERSION_LEGACY) + || header[5] != RECORD_MARKER || 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))) { @@ -799,6 +1483,8 @@ tiny_store_status_name(tiny_store_status_t status) return "IO_ERROR"; case TINY_STORE_CORRUPT: return "CORRUPT_STORE"; + case TINY_STORE_CONFLICT: + return "CONFLICT"; default: return "INTERNAL_ERROR"; } @@ -820,6 +1506,12 @@ tiny_store_write(tiny_store_t *store, const tiny_message_t *input, tiny_message_ { return TINY_STORE_IO_ERROR; } + if (!input->content[0] || strlen(input->content) > TINY_CONTENT_MAX + || !tiny_utf8_valid(input->content, strlen(input->content)) + || !metadata_targets_valid(store, NULL, &input->metadata)) + { + return TINY_STORE_CONFLICT; + } tiny_message_init(&message); tiny_message_init(&indexed); if (tiny_message_copy(&message, input) != 0) @@ -827,6 +1519,8 @@ tiny_store_write(tiny_store_t *store, const tiny_message_t *input, tiny_message_ return TINY_STORE_NO_MEMORY; } message.timestamp_ms = (int64_t)now_milliseconds(); + message.updated_at_ms = message.timestamp_ms; + message.revision = 1; 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) @@ -856,6 +1550,127 @@ tiny_store_write(tiny_store_t *store, const tiny_message_t *input, tiny_message_ } tiny_store_status_t +tiny_store_update(tiny_store_t *store, + const char *message_id, + const tiny_message_patch_t *patch, + tiny_message_t *updated, + bool *changed) +{ + tiny_store_entry_t *entry; + tiny_message_t prepared, response; + tiny_message_patch_t effective; + unsigned char *payload = NULL; + uint32_t payload_length = 0; + uint64_t log_size_before = 0; + uint64_t revision; + int64_t updated_at; + tiny_store_status_t status; + if (!store || !message_id || !patch || !updated || !changed || !tiny_message_id_valid(message_id)) + { + return TINY_STORE_NOT_FOUND; + } + tiny_message_init(updated); + tiny_message_init(&prepared); + tiny_message_init(&response); + *changed = false; + entry = find_entry(store, message_id); + if (!entry || entry->deleted) + { + return TINY_STORE_NOT_FOUND; + } + if (patch->has_expected_revision && patch->expected_revision != entry->message.revision) + { + return TINY_STORE_CONFLICT; + } + if (entry->message.revision >= INT32_MAX) + { + return TINY_STORE_CONFLICT; + } + effective = *patch; + if ((effective.metadata_mask & ~TINY_METADATA_ALL) != 0 + || (effective.metadata_mask + && (!effective.metadata || !tiny_metadata_valid(effective.metadata)))) + { + return TINY_STORE_CONFLICT; + } + if (effective.content && strcmp(effective.content, entry->message.content) == 0) + { + effective.content = NULL; + } + if (!effective.metadata || effective.metadata_mask == 0 + || !metadata_patch_changes( + &entry->message.metadata, effective.metadata, effective.metadata_mask)) + { + effective.metadata_mask = 0; + } + if (!effective.content && effective.metadata_mask == 0) + { + return tiny_message_copy(updated, &entry->message) == 0 ? TINY_STORE_OK + : TINY_STORE_NO_MEMORY; + } + revision = entry->message.revision + 1; + updated_at = (int64_t)now_milliseconds(); + status = prepare_update(store, entry, &effective, revision, updated_at, &prepared); + if (status != TINY_STORE_OK) + { + tiny_message_free(&prepared); + return status; + } + if (tiny_message_copy(&response, &prepared) != 0) + { + status = TINY_STORE_NO_MEMORY; + tiny_message_free(&prepared); + return status; + } + if (!serialize_update(entry, &effective, revision, updated_at, &payload, &payload_length)) + { + status = TINY_STORE_NO_MEMORY; + tiny_message_free(&prepared); + tiny_message_free(&response); + return status; + } + if (!tiny_sys_file_size(&store->log, &log_size_before)) + { + free(payload); + tiny_message_free(&prepared); + tiny_message_free(&response); + return TINY_STORE_IO_ERROR; + } + status = append_record_to( + &store->log, RECORD_UPDATE, store->last_sequence + 1, payload, payload_length, true); + if (status != TINY_STORE_OK) + { + bool rolled_back = log_size_before <= INT64_MAX + && tiny_sys_file_seek( + &store->log, (int64_t)log_size_before, TINY_SYS_SEEK_BEGIN) + && tiny_sys_file_truncate(&store->log) + && tiny_sys_file_flush(&store->log); + set_error(store, "failed to append and flush an UPDATE record"); + if (!rolled_back) + { + set_error(store, "UPDATE failed and its incomplete record could not be rolled back"); + } + free(payload); + tiny_message_free(&prepared); + tiny_message_free(&response); + return status; + } + tiny_message_free(&entry->message); + entry->message = prepared; + tiny_message_init(&prepared); + *updated = response; + tiny_message_init(&response); + store->last_sequence++; + store->update_records++; + store->dead_bytes += RECORD_HEADER_SIZE + payload_length; + *changed = true; + free(payload); + tiny_message_free(&prepared); + tiny_message_free(&response); + 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; @@ -882,6 +1697,7 @@ tiny_store_delete(tiny_store_t *store, const char *message_id, const char *sessi } store->last_sequence++; entry->deleted = true; + store->dead_bytes += RECORD_HEADER_SIZE + sizeof(payload); return TINY_STORE_OK; } @@ -905,6 +1721,15 @@ tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report) { return TINY_STORE_IO_ERROR; } + if (store->snapshot_valid && store->update_records == 0 && store->dead_bytes == 0) + { + uint64_t size = 0; + (void)tiny_sys_file_size(&store->log, &size); + report->messages_indexed = store->entry_count; + report->bytes_before = size; + report->bytes_after = size; + return TINY_STORE_OK; + } memcpy(previous_generation, store->generation, sizeof(previous_generation)); previous_has_generation = store->has_generation; temporary = tiny_sys_path_join(store->directory, "active.tmp"); @@ -1002,6 +1827,11 @@ tiny_store_rebuild(tiny_store_t *store, tiny_rebuild_report_t *report) report->bytes_before = before; report->bytes_after = after; status = write_snapshot(store); + if (status == TINY_STORE_OK) + { + store->update_records = 0; + store->dead_bytes = 0; + } } if (status != TINY_STORE_OK) { @@ -1032,6 +1862,7 @@ tiny_store_health(const tiny_store_t *store, tiny_store_health_t *health) { return; } + health->format_version = RECORD_VERSION; if (tiny_sys_file_size(&store->log, &size)) { health->log_size = size; @@ -1059,6 +1890,17 @@ tiny_store_health(const tiny_store_t *store, tiny_store_health_t *health) } } health->last_sequence = store->last_sequence; + health->update_records = store->update_records; + health->dead_bytes = store->dead_bytes; + { + tiny_sys_io_counters_t counters; + tiny_sys_io_counters_get(&counters); + health->write_bytes = counters.write_bytes; + health->write_calls = counters.write_calls; + health->flush_calls = counters.flush_calls; + health->truncate_calls = counters.truncate_calls; + health->replace_calls = counters.replace_calls; + } health->recovered_tail = store->recovered_tail; health->snapshot_valid = store->snapshot_valid; snprintf(health->last_error, sizeof(health->last_error), "%s", store->last_error); diff --git a/src/storage/store.h b/src/storage/store.h index 619f86f..3ddd24d 100644 --- a/src/storage/store.h +++ b/src/storage/store.h @@ -14,7 +14,8 @@ typedef enum tiny_store_status TINY_STORE_BUSY, TINY_STORE_IO_ERROR, TINY_STORE_CORRUPT, - TINY_STORE_NO_MEMORY + TINY_STORE_NO_MEMORY, + TINY_STORE_CONFLICT } tiny_store_status_t; typedef struct tiny_store_entry @@ -36,16 +37,33 @@ typedef struct tiny_rebuild_report typedef struct tiny_store_health { + uint32_t format_version; uint64_t log_size; uint64_t snapshot_size; uint64_t active_messages; uint64_t deleted_messages; uint64_t last_sequence; + uint64_t update_records; + uint64_t dead_bytes; + uint64_t write_bytes; + uint64_t write_calls; + uint64_t flush_calls; + uint64_t truncate_calls; + uint64_t replace_calls; bool recovered_tail; bool snapshot_valid; char last_error[160]; } tiny_store_health_t; +typedef struct tiny_message_patch +{ + const char *content; + const tiny_metadata_t *metadata; + tiny_metadata_mask_t metadata_mask; + uint64_t expected_revision; + bool has_expected_revision; +} tiny_message_patch_t; + 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); @@ -54,6 +72,11 @@ 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_update(tiny_store_t *store, + const char *message_id, + const tiny_message_patch_t *patch, + tiny_message_t *updated, + bool *changed); 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); |