Allow 1 intent_write over big_write in fsync mode

src/blockstore/blockstore_heap.cpp

@@ -71,8 +71,7 @@ bool heap_write_t::is_compacted(uint64_t compacted_lsn)
 
 bool heap_write_t::can_be_collapsed(blockstore_heap_t *heap)
 {
-    return flags == BS_HEAP_INTENT_WRITE ||
-        !heap->dsk->csum_block_size || heap->dsk->csum_block_size == heap->dsk->bitmap_granularity ||
+    return !heap->dsk->csum_block_size || heap->dsk->csum_block_size == heap->dsk->bitmap_granularity ||
         !(offset % heap->dsk->csum_block_size) && !(len % heap->dsk->csum_block_size);
 }
 
@@ -1991,7 +1990,7 @@ void blockstore_heap_t::mark_lsn_fsynced(uint64_t lsn)
 {
     if (lsn > fsynced_lsn)
     {
-        assert(lsn >= first_inflight_lsn && lsn <= completed_lsn);
+        assert(lsn <= completed_lsn);
         fsynced_lsn = lsn;
         deref_overwrites(lsn);
     }

src/blockstore/blockstore_write.cpp

@@ -148,11 +148,11 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     }
     // Only one INTENT_WRITE is allowed at a time, but in fact,
     // parallel writes to the same object are forbidden anyway
-    else if (dsk.disable_data_fsync &&
-        op->opcode == BS_OP_WRITE_STABLE &&
+    else if (op->opcode == BS_OP_WRITE_STABLE &&
         op->len > 0 && op->len <= dsk.atomic_write_size &&
+        // One intent-write is allowed even with fsyncs because BIG_WRITE is always counted as fsynced
         (obj->get_writes()->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE) ||
-        obj->get_writes()->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE)))
+        obj->get_writes()->flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE) && dsk.disable_data_fsync))
     {
         // Direct intent-write
         BS_SUBMIT_CHECK_SQES(1);