Write a test for store v1 & journal corruption preservation, fix MULTIPLE bugs

src/blockstore/v1/flush.cpp

@@ -71,6 +71,11 @@ bool journal_flusher_t::is_active()
     return active_flushers > 0 || dequeuing;
 }
 
+size_t journal_flusher_t::get_queue_size()
+{
+    return flush_queue.size();
+}
+
 void journal_flusher_t::loop()
 {
     target_flusher_count = bs->write_iodepth*2;
@@ -384,6 +389,7 @@ stop_flusher:
         wait_state = 0;
         return true;
     }
+    copy_count = 0;
     try_trim = true;
     cur.oid = flusher->flush_queue.front();
     cur.version = flusher->flush_versions[cur.oid];
@@ -511,6 +517,31 @@ resume_2:
             {
                 uo_it->second.was_changed = true;
             }
+            if (!bs->journal.inmemory)
+            {
+                // Verify journaled data checksums (but not COALESCED)
+                for (it = v.begin(); it != v.end(); it++)
+                {
+                    if (it->copy_flags == COPY_BUF_JOURNAL)
+                    {
+                        iovec iov = { .iov_base = it->buf, .iov_len = it->len };
+                        bs->verify_journal_checksums(
+                            it->csum_buf, it->offset, &iov, 1,
+                            [&](uint32_t bad_block, uint32_t calc_csum, uint32_t stored_csum)
+                            {
+                                printf(
+                                    "Checksum mismatch in object %jx:%jx v%ju in journal at 0x%jx, checksum block #%u: got %08x, expected %08x\n",
+                                    cur.oid.inode, cur.oid.stripe, cur.version, it->disk_offset,
+                                    bad_block / bs->dsk.csum_block_size, calc_csum, stored_csum
+                                );
+                                bad_block += it->offset;
+                                assert(!(bad_block % bs->dsk.csum_block_size) && bad_block < bs->dsk.data_block_size);
+                                mangle_csum_blocks.insert(bad_block);
+                            }
+                        );
+                    }
+                }
+            }
         }
         // Submit data writes
         for (it = v.begin(); it != v.end(); it++)
@@ -634,6 +665,7 @@ resume_2:
         }
         // All done
         flusher->active_flushers--;
+        copy_count = 0; // used by is_mutated()...
         wait_state = 0;
         goto resume_0;
     }
@@ -815,35 +847,21 @@ bool journal_flusher_co::clear_incomplete_csum_block_bits(int wait_base)
                 bs->verify_padded_checksums(new_clean_bitmap, new_clean_bitmap + 2*bs->dsk.clean_entry_bitmap_size,
                     v[i].offset, &iov, 1, [&](uint32_t bad_block, uint32_t calc_csum, uint32_t stored_csum)
                 {
-                    printf("Checksum mismatch in object %jx:%jx v%ju in data area at offset 0x%jx+0x%x: got %08x, expected %08x\n",
+                    printf("Checksum mismatch in object %jx:%jx v%ju in data area at offset 0x%jx+0x%x during flush: got %08x, expected %08x\n",
                         cur.oid.inode, cur.oid.stripe, old_clean_ver, old_clean_loc, bad_block, calc_csum, stored_csum);
-                    for (uint32_t j = 0; j < bs->dsk.csum_block_size; j += bs->dsk.bitmap_granularity)
+                    assert(!(bad_block % bs->dsk.csum_block_size) && bad_block < bs->dsk.data_block_size);
-                    {
+                    mangle_csum_blocks.insert(bad_block);
-                        // Simplest method of mangling: flip one byte in every sector
-                        ((uint8_t*)v[i].buf)[j+bad_block-v[i].offset] ^= 0xff;
-                    }
                 });
             }
             else
             {
                 bs->verify_journal_checksums(v[i].csum_buf, v[i].offset, &iov, 1, [&](uint32_t bad_block, uint32_t calc_csum, uint32_t stored_csum)
                 {
-                    printf("Checksum mismatch in object %jx:%jx v%ju in journal at offset 0x%jx+0x%x (block offset 0x%jx): got %08x, expected %08x\n",
+                    printf("Checksum mismatch in object %jx:%jx v%ju in journal at offset 0x%jx+0x%x (block offset 0x%jx) during flush: got %08x, expected %08x\n",
                         cur.oid.inode, cur.oid.stripe, old_clean_ver,
                         v[i].disk_offset, bad_block, v[i].offset, calc_csum, stored_csum);
-                    bad_block += (v[i].offset/bs->dsk.csum_block_size) * bs->dsk.csum_block_size;
+                    assert(!(bad_block % bs->dsk.csum_block_size) && bad_block < bs->dsk.data_block_size);
-                    uint32_t bad_block_end = bad_block + bs->dsk.csum_block_size + (v[i].offset/bs->dsk.csum_block_size) * bs->dsk.csum_block_size;
+                    mangle_csum_blocks.insert(bad_block);
-                    if (bad_block < v[i].offset)
-                        bad_block = v[i].offset;
-                    if (bad_block_end > v[i].offset+v[i].len)
-                        bad_block_end = v[i].offset+v[i].len;
-                    bad_block -= v[i].offset;
-                    bad_block_end -= v[i].offset;
-                    for (uint32_t j = bad_block; j < bad_block_end; j += bs->dsk.bitmap_granularity)
-                    {
-                        // Simplest method of mangling: flip one byte in every sector
-                        ((uint8_t*)v[i].buf)[j] ^= 0xff;
-                    }
                 });
             }
         }
@@ -952,6 +970,11 @@ void journal_flusher_co::calc_block_checksums(uint32_t *new_data_csums, bool ski
     }
     // `v` should contain aligned items, possibly split into pieces
     assert(!block_done);
+    for (uint32_t mangle_block: mangle_csum_blocks)
+    {
+        // Flip 1 bit
+        new_data_csums[mangle_block / bs->dsk.csum_block_size] ^= 1;
+    }
 }
 
 void journal_flusher_co::scan_dirty()
@@ -1118,6 +1141,7 @@ bool journal_flusher_co::read_dirty(int wait_base)
     if (wait_state == wait_base)        goto resume_0;
     else if (wait_state == wait_base+1) goto resume_1;
     wait_count = wait_journal_count = 0;
+    mangle_csum_blocks.clear();
     if (bs->journal.inmemory && !read_to_fill_incomplete)
     {
         // Happy path: nothing to read :)

src/blockstore/v1/flush.h

@@ -66,6 +66,7 @@ class journal_flusher_co
     uint64_t clean_bitmap_offset, clean_bitmap_len;
     uint8_t *clean_init_dyn_ptr;
     uint8_t *new_clean_bitmap;
+    std::unordered_set<uint32_t> mangle_csum_blocks;
 
     uint64_t new_trim_pos;
 
@@ -123,6 +124,7 @@ public:
     void loop();
     bool is_trim_wanted() { return trim_wanted; }
     bool is_active();
+    size_t get_queue_size();
     void mark_trim_possible();
     void request_trim();
     void release_trim();

src/blockstore/v1/impl.cpp

@@ -6,11 +6,12 @@
 
 namespace v1 {
 
-blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i *ringloop, timerfd_manager_t *tfd)
+blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i *ringloop, timerfd_manager_t *tfd, bool mock_mode)
 {
     assert(sizeof(blockstore_op_private_t) <= BS_OP_PRIVATE_DATA_SIZE);
     this->tfd = tfd;
     this->ringloop = ringloop;
+    dsk.mock_mode = mock_mode;
     ring_consumer.loop = [this]() { loop(); };
     ringloop->register_consumer(&ring_consumer);
     initialized = 0;

src/blockstore/v1/impl.h

@@ -30,6 +30,8 @@
 
 //#define BLOCKSTORE_DEBUG
 
+struct bs_test_t;
+
 namespace v1 {
 
 #include "journal.h"
@@ -122,6 +124,7 @@ typedef uint64_t pool_pg_id_t;
 
 class blockstore_impl_t: public blockstore_i
 {
+    friend struct ::bs_test_t;
     blockstore_disk_t dsk;
 
     /******* OPTIONS *******/
@@ -220,6 +223,7 @@ class blockstore_impl_t: public blockstore_i
 
     // Read
     int dequeue_read(blockstore_op_t *read_op);
+    void release_clean(blockstore_op_t *op);
     void find_holes(std::vector<copy_buffer_t> & read_vec, uint32_t item_start, uint32_t item_end,
         std::function<int(int, bool, uint32_t, uint32_t)> callback);
     int fulfill_read(blockstore_op_t *read_op,
@@ -281,7 +285,7 @@ class blockstore_impl_t: public blockstore_i
 
 public:
 
-    blockstore_impl_t(blockstore_config_t & config, ring_loop_i *ringloop, timerfd_manager_t *tfd);
+    blockstore_impl_t(blockstore_config_t & config, ring_loop_i *ringloop, timerfd_manager_t *tfd, bool mock_mode = false);
     ~blockstore_impl_t();
 
     void parse_config(blockstore_config_t & config);

src/blockstore/v1/read.cpp

@@ -101,8 +101,8 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *read_op,
                     .copy_flags = COPY_BUF_JOURNAL|COPY_BUF_CSUM_FILL,
                     .offset = blk_begin,
                     .len = blk_end-blk_begin,
-                    .csum_buf = (csum + (blk_begin/dsk.csum_block_size -
+                    .csum_buf = (!csum ? NULL : (csum + (blk_begin/dsk.csum_block_size -
-                        item_start/dsk.csum_block_size) * (dsk.data_csum_type & 0xFF)),
+                        item_start/dsk.csum_block_size) * (dsk.data_csum_type & 0xFF))),
                     .dyn_data = dyn_data,
                 });
                 if (dyn_data)
@@ -134,7 +134,7 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *read_op,
                 // If we don't track it then we may IN THEORY read another object's data:
                 // submit read -> remove the object -> flush remove -> overwrite with another object -> finish read
                 // Very improbable, but possible
-                PRIV(read_op)->clean_loc_used = 1;
+                PRIV(read_op)->clean_loc_used = UINT64_MAX;
             }
             rv.insert(rv.begin() + pos, el);
             fulfilled += el.len;
@@ -178,7 +178,7 @@ int blockstore_impl_t::fill_partial_checksum_blocks(std::vector<copy_buffer_t> &
     uint32_t end_block = 0;
     while (start_block <= last_block)
     {
-        if (read_range_fulfilled(rv, fulfilled, read_buf, clean_entry_bitmap,
+        if (read_range_fulfilled(rv, fulfilled, read_buf, from_journal ? NULL : clean_entry_bitmap,
             start_block*dsk.csum_block_size < read_offset ? read_offset : start_block*dsk.csum_block_size,
             (start_block+1)*dsk.csum_block_size > read_end ? read_end : (start_block+1)*dsk.csum_block_size))
         {
@@ -190,7 +190,7 @@ int blockstore_impl_t::fill_partial_checksum_blocks(std::vector<copy_buffer_t> &
             // Find a sequence of checksum blocks required to be read
             end_block = start_block;
             while ((end_block+1)*dsk.csum_block_size < read_end &&
-                !read_range_fulfilled(rv, fulfilled, read_buf, clean_entry_bitmap,
+                !read_range_fulfilled(rv, fulfilled, read_buf, from_journal ? NULL : clean_entry_bitmap,
                     (end_block+1)*dsk.csum_block_size < read_offset ? read_offset : (end_block+1)*dsk.csum_block_size,
                     (end_block+2)*dsk.csum_block_size > read_end ? read_end : (end_block+2)*dsk.csum_block_size))
             {
@@ -202,7 +202,7 @@ int blockstore_impl_t::fill_partial_checksum_blocks(std::vector<copy_buffer_t> &
                 .copy_flags = COPY_BUF_CSUM_FILL | (from_journal ? COPY_BUF_JOURNALED_BIG : 0),
                 .offset = start_block*dsk.csum_block_size,
                 .len = (end_block-start_block)*dsk.csum_block_size,
-                // save clean_entry_bitmap if we're reading clean data from the journal
+                // save clean_entry_bitmap if we're reading clean data from the journal -- for checksums
                 .csum_buf = from_journal ? clean_entry_bitmap : NULL,
                 .dyn_data = dyn_data,
             });
@@ -226,6 +226,11 @@ bool blockstore_impl_t::read_range_fulfilled(std::vector<copy_buffer_t> & rv, ui
     {
         if (alloc)
             return 0;
+        if (!clean_entry_bitmap)
+        {
+            all_done = false;
+            return 0;
+        }
         int diff = 0;
         uint32_t bmp_start = cur_start/dsk.bitmap_granularity;
         uint32_t bmp_end = cur_end/dsk.bitmap_granularity;
@@ -323,7 +328,7 @@ bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uin
             {
                 iov[n_iov++] = (struct iovec){ (uint8_t*)op->buf+cur_start-op->offset, lim_end-cur_start };
                 rv.insert(rv.begin() + pos, (copy_buffer_t){
-                    .copy_flags = COPY_BUF_DATA,
+                    .copy_flags = COPY_BUF_DATA|COPY_BUF_COALESCED,
                     .offset = cur_start,
                     .len = lim_end-cur_start,
                 });
@@ -376,7 +381,7 @@ bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uin
     {
         // Reads running parallel to flushes of the same clean block may read
         // a mixture of old and new data. So we don't verify checksums for such blocks.
-        PRIV(op)->clean_loc_used = 1;
+        PRIV(op)->clean_loc_used = UINT64_MAX;
     }
     return true;
 }
@@ -515,6 +520,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *read_op)
     return 2;
 undo_read:
     // need to wait. undo added requests, don't dequeue op
+    release_clean(read_op);
     if (dsk.csum_block_size > dsk.bitmap_granularity)
     {
         for (auto & vec: rv)
@@ -535,6 +541,32 @@ undo_read:
     return 0;
 }
 
+void blockstore_impl_t::release_clean(blockstore_op_t *op)
+{
+    if (PRIV(op)->clean_loc_used == UINT64_MAX)
+    {
+        PRIV(op)->clean_loc_used = 0;
+    }
+    if (PRIV(op)->clean_loc_used)
+    {
+        // Release clean data block
+        auto uo_it = used_clean_objects.find(PRIV(op)->clean_loc_used - 1);
+        if (uo_it != used_clean_objects.end())
+        {
+            uo_it->second.refs--;
+            if (uo_it->second.refs <= 0)
+            {
+                if (uo_it->second.was_freed)
+                {
+                    data_alloc->set((PRIV(op)->clean_loc_used - 1) / dsk.data_block_size, false);
+                }
+                used_clean_objects.erase(uo_it);
+            }
+        }
+        PRIV(op)->clean_loc_used = 0;
+    }
+}
+
 int blockstore_impl_t::pad_journal_read(std::vector<copy_buffer_t> & rv, copy_buffer_t & cp,
     // FIXME Passing dirty_entry& would be nicer
     uint64_t dirty_offset, uint64_t dirty_end, uint64_t dirty_loc, uint8_t *csum_ptr, int *dyn_data,
@@ -615,7 +647,7 @@ bool blockstore_impl_t::fulfill_clean_read(blockstore_op_t *read_op, uint64_t &
                 return false;
             }
         }
-        PRIV(read_op)->clean_loc_used = req > 0;
+        PRIV(read_op)->clean_loc_used = req > 0 ? UINT64_MAX : 0;
     }
     else if (from_journal)
     {
@@ -679,13 +711,13 @@ bool blockstore_impl_t::fulfill_clean_read(blockstore_op_t *read_op, uint64_t &
         }
     }
     // Increment reference counter if clean data is being read from the disk
-    if (PRIV(read_op)->clean_loc_used)
+    if (PRIV(read_op)->clean_loc_used == UINT64_MAX)
     {
         auto & uo = used_clean_objects[clean_loc];
         uo.refs++;
         if (dsk.csum_block_size && flusher->is_mutated(clean_loc))
             uo.was_changed = true;
-        PRIV(read_op)->clean_loc_used = clean_loc;
+        PRIV(read_op)->clean_loc_used = clean_loc + 1;
     }
     return true;
 }
@@ -725,12 +757,18 @@ bool blockstore_impl_t::verify_padded_checksums(uint8_t *clean_entry_bitmap, uin
         while (pos < iov[i].iov_len)
         {
             uint32_t start = pos;
-            uint8_t bit = (clean_entry_bitmap[bmp_pos >> 3] >> (bmp_pos & 0x7)) & 1;
+            uint8_t bit = 1;
-            while (pos < iov[i].iov_len && ((clean_entry_bitmap[bmp_pos >> 3] >> (bmp_pos & 0x7)) & 1) == bit)
+            if (clean_entry_bitmap)
             {
-                pos += dsk.bitmap_granularity;
+                bit = (clean_entry_bitmap[bmp_pos >> 3] >> (bmp_pos & 0x7)) & 1;
-                bmp_pos++;
+                while (pos < iov[i].iov_len && ((clean_entry_bitmap[bmp_pos >> 3] >> (bmp_pos & 0x7)) & 1) == bit)
+                {
+                    pos += dsk.bitmap_granularity;
+                    bmp_pos++;
+                }
             }
+            else
+                pos = iov[i].iov_len;
             uint32_t len = pos-start;
             auto buf = (uint8_t*)iov[i].iov_base+start;
             while (block_done+len >= dsk.csum_block_size)
@@ -807,7 +845,7 @@ bool blockstore_impl_t::verify_clean_padded_checksums(blockstore_op_t *op, uint6
 {
     uint32_t offset = clean_loc % dsk.data_block_size;
     if (from_journal)
-        return verify_padded_checksums(dyn_data, dyn_data + dsk.clean_entry_bitmap_size, offset, iov, n_iov, bad_block_cb);
+        return verify_padded_checksums(NULL, dyn_data + dsk.clean_entry_bitmap_size, offset, iov, n_iov, bad_block_cb);
     clean_loc = (clean_loc / dsk.data_block_size) * dsk.data_block_size;
     if (!dyn_data)
     {
@@ -835,7 +873,7 @@ void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op
             void *meta_block = NULL;
             if (dsk.csum_block_size > dsk.bitmap_granularity)
             {
-                for (int i = rv.size()-1; i >= 0 && (rv[i].copy_flags & COPY_BUF_CSUM_FILL); i--)
+                for (int i = 0; i < rv.size(); i++)
                 {
                     if (rv[i].copy_flags & COPY_BUF_META_BLOCK)
                     {
@@ -845,8 +883,36 @@ void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op
                         rv[i].buf = NULL;
                         continue;
                     }
-                    struct iovec *iov = (struct iovec*)((uint8_t*)rv[i].buf + (rv[i].len & 0xFFFFFFFF));
+                    if (rv[i].copy_flags & COPY_BUF_COALESCED)
-                    int n_iov = rv[i].len >> 32;
+                    {
+                        // Sub-block shared with another read. Skip
+                        continue;
+                    }
+                    if ((rv[i].copy_flags & COPY_BUF_JOURNAL) && journal.inmemory)
+                    {
+                        // Do not check journal checksums in-memory
+                        continue;
+                    }
+                    iovec single_iov = {};
+                    iovec *iov = NULL;
+                    int n_iov = 0;
+                    if (rv[i].copy_flags & COPY_BUF_CSUM_FILL)
+                    {
+                        // Padded, buffer list passed using a 'creepy way'
+                        iov = (struct iovec*)((uint8_t*)rv[i].buf + (rv[i].len & 0xFFFFFFFF));
+                        n_iov = rv[i].len >> 32;
+                    }
+                    else
+                    {
+                        // Not padded, buffer is fully within the input buffer
+                        assert(op->buf);
+                        assert(rv[i].csum_buf);
+                        iov = &single_iov;
+                        n_iov = 1;
+                        assert(rv[i].offset >= op->offset);
+                        assert(rv[i].offset + rv[i].len <= op->offset + op->len);
+                        single_iov = { .iov_base = op->buf + rv[i].offset - op->offset, .iov_len = rv[i].len };
+                    }
                     bool ok = true;
                     if (rv[i].copy_flags & COPY_BUF_JOURNAL)
                     {
@@ -944,23 +1010,7 @@ void blockstore_impl_t::handle_read_event(ring_data_t *data, blockstore_op_t *op
                 meta_block = NULL;
             }
         }
-        if (PRIV(op)->clean_loc_used)
+        release_clean(op);
-        {
-            // Release clean data block
-            auto uo_it = used_clean_objects.find(PRIV(op)->clean_loc_used);
-            if (uo_it != used_clean_objects.end())
-            {
-                uo_it->second.refs--;
-                if (uo_it->second.refs <= 0)
-                {
-                    if (uo_it->second.was_freed)
-                    {
-                        data_alloc->set(PRIV(op)->clean_loc_used / dsk.data_block_size, false);
-                    }
-                    used_clean_objects.erase(uo_it);
-                }
-            }
-        }
         if (!journal.inmemory)
         {
             // Release journal sector usage

src/test/CMakeLists.txt

@@ -71,6 +71,12 @@ add_dependencies(build_tests test_blockstore)
 target_link_libraries(test_blockstore vitastor_blk vitastor_common ${ISAL_LIBRARIES})
 add_test(NAME test_blockstore COMMAND test_blockstore)
 
+# test_blockstore_v1
+add_executable(test_blockstore_v1 EXCLUDE_FROM_ALL test_blockstore_v1.cpp ringloop_mock.cpp)
+add_dependencies(build_tests test_blockstore_v1)
+target_link_libraries(test_blockstore_v1 vitastor_blk vitastor_common ${ISAL_LIBRARIES})
+add_test(NAME test_blockstore_v1 COMMAND test_blockstore_v1)
+
 ## test_shit
 #add_executable(test_shit test_shit.cpp osd_peering_pg.cpp)
 #target_link_libraries(test_shit ${LIBURING_LIBRARIES} m)

src/test/test_blockstore_v1.cpp

@@ -0,0 +1,286 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
+
+#include <malloc.h>
+#include "str_util.h"
+#include "ringloop_mock.h"
+#include "blockstore/v1/impl.h"
+
+struct bs_test_t
+{
+    blockstore_config_t config;
+    disk_mock_t *data_disk = NULL;
+    disk_mock_t *meta_disk = NULL;
+    std::function<bool(io_uring_sqe*)> sqe_handler;
+    ring_loop_mock_t *ringloop = NULL;
+    timerfd_manager_t *tfd = NULL;
+    v1::blockstore_impl_t *bs = NULL;
+
+    ~bs_test_t()
+    {
+        destroy();
+    }
+
+    void destroy_bs()
+    {
+        if (bs)
+        {
+            delete bs;
+            bs = NULL;
+        }
+    }
+
+    blockstore_disk_t& dsk()
+    {
+        return bs->dsk;
+    }
+
+    v1::journal_flusher_t* flusher()
+    {
+        return bs->flusher;
+    }
+
+    void destroy()
+    {
+        while (bs && !bs->is_safe_to_stop())
+            ringloop->loop();
+        destroy_bs();
+        if (tfd)
+        {
+            delete tfd;
+            tfd = NULL;
+        }
+        if (meta_disk)
+        {
+            delete meta_disk;
+            meta_disk = NULL;
+        }
+        if (data_disk)
+        {
+            delete data_disk;
+            data_disk = NULL;
+        }
+        if (ringloop)
+        {
+            delete ringloop;
+            ringloop = NULL;
+        }
+    }
+
+    void default_cfg()
+    {
+        config["data_device"] = "./test_data.bin";
+        config["data_device_size"] = "1073741824";
+        config["data_device_sect"] = "4096";
+        config["meta_offset"] = "0";
+        config["journal_offset"] = "16777216";
+        config["data_offset"] = "33554432";
+        config["disable_data_fsync"] = "1";
+        config["immediate_commit"] = "all";
+        config["log_level"] = "10";
+        config["meta_format"] = "2";
+    }
+
+    void init()
+    {
+        if (!ringloop)
+        {
+            ringloop = new ring_loop_mock_t(RINGLOOP_DEFAULT_SIZE, [&](io_uring_sqe *sqe)
+            {
+                if (sqe_handler && sqe_handler(sqe))
+                {
+                }
+                else if (sqe->fd == MOCK_DATA_FD)
+                {
+                    bool ok = data_disk->submit(sqe);
+                    assert(ok);
+                    ringloop->mark_completed((ring_data_t*)sqe->user_data);
+                }
+                else if (sqe->fd == MOCK_META_FD)
+                {
+                    bool ok = meta_disk->submit(sqe);
+                    assert(ok);
+                    ringloop->mark_completed((ring_data_t*)sqe->user_data);
+                }
+                else
+                {
+                    assert(0);
+                }
+            });
+        }
+        if (!tfd)
+        {
+            tfd = new timerfd_manager_t(nullptr);
+        }
+        if (!data_disk)
+        {
+            data_disk = new disk_mock_t("data disk", parse_size(config["data_device_size"]), config["disable_data_fsync"] != "1");
+            data_disk->clear(0, parse_size(config["data_offset"]));
+        }
+        uint64_t meta_size = parse_size(config["meta_device_size"]);
+        if (meta_size && !meta_disk)
+        {
+            meta_disk = new disk_mock_t("meta disk", meta_size, config["disable_meta_fsync"] != "1");
+            meta_disk->clear(0, meta_size);
+        }
+        if (!bs)
+        {
+            bs = new v1::blockstore_impl_t(config, ringloop, tfd, true);
+            while (!bs->is_started())
+                ringloop->loop();
+        }
+    }
+
+    void exec_op(blockstore_op_t *op)
+    {
+        bool done = false;
+        op->callback = [&](blockstore_op_t *op)
+        {
+            done = true;
+        };
+        bs->enqueue_op(op);
+        while (!done)
+            ringloop->loop();
+        op->callback = nullptr;
+    }
+};
+
+static bool memcheck(uint8_t *buf, uint8_t byte, size_t len)
+{
+    for (size_t i = 0; i < len; i++)
+        if (buf[i] != byte)
+            return false;
+    return true;
+}
+
+// Check that journaled data corruption is preserved after flushing
+// TODO more variations:
+// - block 4k, block 16k
+// - write at 16k, write at 36k
+// - read with corrupted, read without corrupted
+// - corrupt data, corrupt journal
+// - inmemory, not inmemory
+// - partial data checksum block read, full read
+// - clean object at 1, at 0 (to check regressions with clean_loc_used)
+static void test_preserve_corruption()
+{
+    printf("\n-- test_preserve_corruption\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["inmemory_journal"] = "0";
+    test.config["data_csum_type"] = "crc32c";
+    test.config["csum_block_size"] = "16384";
+    test.init();
+    printf("blockstore initialized\n");
+
+    // Big_write without external bitmap(!) - also checks if journaled big_writes
+    // are handled correctly (they don't have an external bitmap)
+    printf("write v1 0+128k\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op.len);
+    memset(op.buf, 0xAA, op.len);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+    // Small_write
+    printf("write v2 16+4k\n");
+    op.version = 2;
+    op.offset = 16384;
+    op.len = 4096;
+    memset(op.buf, 0xAB, 4096);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+    // Check that it's not compacted
+    assert(test.flusher()->get_queue_size());
+
+    // Read and verify original data before corruption
+    uint64_t small_write_offset = 0, small_write_len = 0;
+    test.sqe_handler = [&](io_uring_sqe *sqe)
+    {
+        if (sqe->off >= test.dsk().journal_offset && sqe->off < test.dsk().journal_offset + test.dsk().journal_len)
+        {
+            auto data = ((ring_data_t*)sqe->user_data);
+            small_write_offset = sqe->off;
+            small_write_len = data->iov.iov_len;
+        }
+        return false;
+    };
+    printf("read v2 (before corruption)\n");
+    blockstore_op_t read_op;
+    read_op.opcode = BS_OP_READ;
+    read_op.oid = { .inode = 1, .stripe = 0 };
+    read_op.version = 2;
+    read_op.offset = 0;
+    read_op.len = 128*1024;
+    read_op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, read_op.len);
+    test.exec_op(&read_op);
+    assert(read_op.retval == read_op.len);
+    assert(memcheck(read_op.buf, 0xAA, 16*1024));
+    assert(memcheck(read_op.buf + 16*1024, 0xAB, 4*1024));
+    assert(memcheck(read_op.buf + 20*1024, 0xAA, 108*1024));
+    assert(small_write_offset && small_write_len);
+    test.sqe_handler = nullptr;
+
+    // Corrupt data on the disk
+    printf("corrupting journaled data\n");
+    {
+        uint8_t *buf = (uint8_t*)malloc_or_die(small_write_len);
+        memset(buf, 0xBB, small_write_len);
+        io_uring_sqe sqe;
+        ring_data_t data = {};
+        iovec v = { .iov_base = buf, .iov_len = small_write_len };
+        sqe.opcode = IORING_OP_WRITEV;
+        sqe.off = small_write_offset;
+        sqe.addr = (uint64_t)&v;
+        sqe.len = 1;
+        sqe.rw_flags = RWF_DSYNC;
+        sqe.user_data = (uint64_t)&data;
+        bool ok = test.data_disk->submit(&sqe);
+        assert(ok);
+        assert(data.res == small_write_len);
+        free(buf);
+    }
+
+    // Read corrupted - should finish with -EDOM
+    printf("read v2 (corrupted) - should fail with -EDOM\n");
+    read_op.version = 2;
+    test.exec_op(&read_op);
+    assert(read_op.retval == -EDOM);
+
+    // Check that it's still not compacted
+    assert(test.flusher()->get_queue_size());
+
+    // Trigger compaction and intercept journal read
+    printf("triggering compaction\n");
+    test.flusher()->request_trim();
+    while (test.flusher()->get_queue_size())
+        test.ringloop->loop();
+    while (test.flusher()->is_active())
+        test.ringloop->loop();
+    test.flusher()->release_trim();
+    assert(!test.flusher()->get_queue_size());
+    printf("compaction complete\n");
+
+    // Store v1 can't cancel compaction because the journal is a ring buffer
+    // so it compacts the object but preserves corruption
+    printf("read v2 (after compaction) - should fail with -EDOM\n");
+    read_op.version = UINT64_MAX;
+    test.exec_op(&read_op);
+    assert(read_op.retval == -EDOM);
+
+    free(op.buf);
+    free(read_op.buf);
+}
+
+int main(int narg, char *args[])
+{
+    test_preserve_corruption();
+    return 0;
+}