Fix iterate_compaction for commit with non-latest version

src/blockstore/blockstore_flush.cpp

@@ -104,9 +104,10 @@ void journal_flusher_t::release_trim()
 void journal_flusher_t::dump_diagnostics()
 {
     printf(
-        "Compaction queue: %u items, data: %ju/%ju blocks used, meta: %ju/%ju bytes used, %u/%ju blocks nearfull\n",
+        "Compaction queue: %u/%u items, data: %ju/%ju blocks used, buffer: %ju/%ju bytes used, meta: %ju/%ju bytes used, %u/%ju blocks nearfull\n",
-        bs->heap->get_to_compact_count(),
+        bs->heap->get_compact_queue_size(), bs->heap->get_to_compact_count(),
         bs->heap->get_data_used_space()/bs->dsk.data_block_size, bs->dsk.block_count,
+        bs->heap->get_buffer_area_used_space(), bs->dsk.journal_len,
         bs->heap->get_meta_used_space(), bs->heap->get_meta_total_space(),
         bs->heap->get_meta_nearfull_blocks(), bs->dsk.meta_area_size/bs->dsk.meta_block_size-1
     );
@@ -220,6 +221,7 @@ resume_16:
                 goto resume_0;
             }
         }
+        assert(false);
     }
     flusher->flushing.insert(cur_oid);
 resume_1:
@@ -286,7 +288,9 @@ resume_1:
     flusher->active_flushers++;
     if (bs->log_level > 10)
     {
-        printf("Compacting %jx:%jx l%ju .. l%ju\n", cur_oid.inode, cur_oid.stripe, compact_info.clean_wr->lsn, compact_info.compact_lsn);
+        printf("Compacting %jx:%jx v%ju..v%ju / l%ju..l%ju\n", cur_oid.inode, cur_oid.stripe,
+            compact_info.clean_wr->version, compact_info.compact_version,
+            compact_info.clean_wr->lsn, compact_info.compact_lsn);
     }
     overwrite_start = overwrite_end = 0;
     if (read_vec.size() > 0)
@@ -723,7 +727,7 @@ bool journal_flusher_co::fsync_buffer(int wait_base)
     else if (wait_state == wait_base+1) goto resume_1;
     else if (wait_state == wait_base+2) goto resume_2;
     if (bs->dsk.disable_journal_fsync && bs->dsk.disable_meta_fsync && bs->dsk.disable_data_fsync ||
-        !bs->unsynced_big_write_count && !bs->unsynced_small_write_count)
+        !bs->unsynced_big_write_count && !bs->unsynced_small_write_count && !bs->unsynced_meta_write_count)
     {
         return true;
     }

src/blockstore/blockstore_heap.cpp

@@ -1347,7 +1347,8 @@ int blockstore_heap_t::add_commit(heap_entry_t *obj, uint64_t version, uint32_t
         }
         if (wr->type() == BS_HEAP_COMMIT)
         {
-            commit_version = wr->version;
+            if (commit_version < wr->version)
+                commit_version = wr->version;
             wr = prev(wr);
             continue;
         }
@@ -1397,7 +1398,8 @@ int blockstore_heap_t::add_rollback(heap_entry_t *obj, uint64_t version, uint32_
         }
         if (wr->type() == BS_HEAP_COMMIT)
         {
-            commit_version = wr->version;
+            if (commit_version < wr->version)
+                commit_version = wr->version;
             wr = prev(wr);
             continue;
         }
@@ -1581,7 +1583,8 @@ void blockstore_heap_t::iterate_with_stable(heap_entry_t *obj, uint64_t max_lsn,
         }
         else if (old_wr->type() == BS_HEAP_COMMIT)
         {
-            commit_version = old_wr->version;
+            if (commit_version < old_wr->version)
+                commit_version = old_wr->version;
         }
         else
         {
@@ -1639,7 +1642,8 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
                 res.compact_version = wr->version;
             }
             res.do_delete = false;
-            commit_version = wr->version;
+            if (commit_version < wr->version)
+                commit_version = wr->version;
             continue;
         }
         bool rolled_back = (wr->version > rollback_version);

src/blockstore/blockstore_sync.cpp

@@ -90,7 +90,8 @@ int blockstore_impl_t::do_sync(blockstore_op_t *op, int base_state)
         // Wait for flusher-initiated sync
         return 0;
     }
-    if (dsk.disable_journal_fsync && dsk.disable_meta_fsync && dsk.disable_data_fsync || !unsynced_big_write_count && !unsynced_small_write_count)
+    if (dsk.disable_journal_fsync && dsk.disable_meta_fsync && dsk.disable_data_fsync ||
+        !unsynced_big_write_count && !unsynced_small_write_count && !unsynced_meta_write_count)
     {
         // We can return immediately because sync only syncs previous writes
         unsynced_big_write_count = unsynced_small_write_count = unsynced_meta_write_count = 0;

src/test/test_heap.cpp

@@ -470,6 +470,52 @@ void test_compact(bool csum, bool stable)
     printf("OK test_compact %s %s\n", stable ? "stable" : "unstable", csum ? "csum" : "no_csum");
 }
 
+void test_iterate_compaction()
+{
+    int res;
+    blockstore_disk_t dsk;
+    _test_init(dsk, false);
+    std::vector<uint8_t> buffer_area(dsk.journal_device_size);
+
+    {
+        blockstore_heap_t heap(&dsk, buffer_area.data());
+        heap.finish_load();
+
+        // Case: BIG_STABLE(v1 l1) -> SMALL(v2 l2) -> SMALL(v3 l3) -> SMALL(v4 l4) -> COMMIT(v2 l5) -> SMALL(v5 l6) -> COMMIT(v5 l7)
+        uint32_t mblock = 0;
+        _test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 4096, buffer_area.data());
+        _test_small_write(heap, dsk, 1, 0, 2, 0, 4096, 0, false, buffer_area.data(), false);
+        _test_small_write(heap, dsk, 1, 0, 3, 4*1024, 4096, 4096, false, buffer_area.data(), false);
+        _test_small_write(heap, dsk, 1, 0, 4, 8*1024, 4096, 8*1024, false, buffer_area.data(), false);
+        object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
+        auto obj = heap.read_entry(oid);
+        res = heap.add_commit(obj, 2, &mblock);
+        assert(res == 0);
+        heap.start_block_write(mblock);
+        heap.complete_block_write(mblock);
+        _test_small_write(heap, dsk, 1, 0, 5, 12*1024, 4096, 12*1024, false, buffer_area.data(), false);
+        obj = heap.read_entry(oid);
+        res = heap.add_commit(obj, 5, &mblock);
+        assert(res == 0);
+        heap.start_block_write(mblock);
+        heap.complete_block_write(mblock);
+
+        assert(heap.get_fsynced_lsn() == 7);
+        int small_writes = 0;
+        obj = heap.read_entry(oid);
+        auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
+        {
+            small_writes++;
+        });
+        assert(compact_info.compact_lsn == 7);
+        assert(compact_info.compact_version == 5);
+        assert(compact_info.clean_wr->lsn == 1);
+        assert(small_writes == 4);
+    }
+
+    printf("OK test_iterate_compaction\n");
+}
+
 void test_modify_bitmap()
 {
     blockstore_disk_t dsk;
@@ -1167,6 +1213,7 @@ int main(int narg, char *args[])
     test_compact(true, false);
     test_compact(false, true);
     test_compact(false, false);
+    test_iterate_compaction();
     test_modify_bitmap();
     test_recheck(false, true, false);
     test_recheck(false, false, false);