From d83580bd68f30c1e96e70f80667cec22145ee9e2 Mon Sep 17 00:00:00 2001
From: Vitaliy Filippov <vitalif@yourcmc.ru>
Date: Sun, 21 Aug 2022 00:31:18 +0300
Subject: [PATCH] Fix "tried to overwrite non-zero metadata entry" when during
 a previous metadata flush writing new entry is completed, but zeroing out an
 old one isn't

---
 src/blockstore_init.cpp  | 12 +++++++++++-
 src/blockstore_init.h    |  2 +-
 src/blockstore_write.cpp | 16 ++++++++++++++++
 3 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/src/blockstore_init.cpp b/src/blockstore_init.cpp
index 12e53e43..fa0efedb 100644
--- a/src/blockstore_init.cpp
+++ b/src/blockstore_init.cpp
@@ -211,8 +211,9 @@ resume_1:
     return 0;
 }
 
-void blockstore_init_meta::handle_entries(void* entries, unsigned count, int block_order)
+bool blockstore_init_meta::handle_entries(void* entries, unsigned count, int block_order)
 {
+    bool updated = false;
     for (unsigned i = 0; i < count; i++)
     {
         clean_disk_entry *entry = (clean_disk_entry*)((uint8_t*)entries + i*bs->dsk.clean_entry_size);
@@ -229,6 +230,11 @@ void blockstore_init_meta::handle_entries(void* entries, unsigned count, int blo
                 if (clean_it != clean_db.end())
                 {
                     // free the previous block
+                    // here we have to zero out the entry because otherwise we'll hit
+                    // "tried to overwrite non-zero metadata entry" later
+                    // FIXME: Write it back if modified with inmemory_meta == false
+                    updated = true;
+                    memset(entry, 0, bs->dsk.clean_entry_size);
 #ifdef BLOCKSTORE_DEBUG
                     printf("Free block %lu from %lx:%lx v%lu (new location is %lu)\n",
                         clean_it->second.location >> block_order,
@@ -253,12 +259,16 @@ void blockstore_init_meta::handle_entries(void* entries, unsigned count, int blo
             }
             else
             {
+                // here we also have to zero out the entry
+                updated = true;
+                memset(entry, 0, bs->dsk.clean_entry_size);
 #ifdef BLOCKSTORE_DEBUG
                 printf("Old clean entry %lu: %lx:%lx v%lu\n", done_cnt+i, entry->oid.inode, entry->oid.stripe, entry->version);
 #endif
             }
         }
     }
+    return updated;
 }
 
 blockstore_init_journal::blockstore_init_journal(blockstore_impl_t *bs)
diff --git a/src/blockstore_init.h b/src/blockstore_init.h
index 8c66dfee..48ba06d2 100644
--- a/src/blockstore_init.h
+++ b/src/blockstore_init.h
@@ -15,7 +15,7 @@ class blockstore_init_meta
     uint64_t entries_loaded = 0;
     struct io_uring_sqe *sqe;
     struct ring_data_t *data;
-    void handle_entries(void *entries, unsigned count, int block_order);
+    bool handle_entries(void *entries, unsigned count, int block_order);
     void handle_event(ring_data_t *data);
 public:
     blockstore_init_meta(blockstore_impl_t *bs);
diff --git a/src/blockstore_write.cpp b/src/blockstore_write.cpp
index 233d7733..d1c56e65 100644
--- a/src/blockstore_write.cpp
+++ b/src/blockstore_write.cpp
@@ -269,6 +269,22 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             cancel_all_writes(op, dirty_it, -ENOSPC);
             return 2;
         }
+        if (inmemory_meta)
+        {
+            // Check once more that metadata entry is zeroed (the reverse means a bug or corruption)
+            uint64_t sector = (loc / (dsk.meta_block_size / dsk.clean_entry_size)) * dsk.meta_block_size;
+            uint64_t pos = (loc % (dsk.meta_block_size / dsk.clean_entry_size));
+            clean_disk_entry *entry = (clean_disk_entry*)((uint8_t*)metadata_buffer + sector + pos*dsk.clean_entry_size);
+            if (entry->oid.inode || entry->oid.stripe || entry->version)
+            {
+                printf(
+                    "Fatal error (metadata corruption or bug): tried to write object %lx:%lx v%lu"
+                    " over a non-zero metadata entry %lu with %lx:%lx v%lu\n", op->oid.inode,
+                    op->oid.stripe, op->version, loc, entry->oid.inode, entry->oid.stripe, entry->version
+                );
+                exit(1);
+            }
+        }
         BS_SUBMIT_GET_SQE(sqe, data);
         write_iodepth++;
         dirty_it->second.location = loc << dsk.block_order;