途中経過コミット

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);