Fix false-positive -EDOM checksum errors on overlapping unaligned reads

When csum_block_size > bitmap_granularity and perfect_csum_update = false
(the default behavior), an unaligned INTENT_WRITE modifies the physical
block directly in-place without recalculating and updating the large
block's checksum.

During a subsequent read that requires padding to the checksum block
boundary (e.g., reading adjacent unaligned data), the read logic must
skip checksum verification for the underlying BIG_WRITE because the
physical data has been safely tampered with by the INTENT_WRITE.

Previously, the need_skip logic in blockstore_read.cpp correctly
identified and skipped the verification for overlapping SMALL_WRITEs
but mistakenly omitted INTENT_WRITEs. This caused the padded read to
verify the stale BIG_WRITE checksum against the newly modified physical
data, leading to a fatal false-positive -EDOM (checksum mismatch) IO
error on reads.

This commit fixes the bug by ensuring BS_HEAP_INTENT_WRITE is evaluated
alongside SMALL_WRITE to apply the COPY_BUF_SKIP_CSUM bypass during
overlapping padded reads.

A regression test case (test_intent_write_corruption) is also added to
reproduce this specific data-layout scenario and verify the fix.

src/blockstore/blockstore.h

@@ -77,11 +77,8 @@ Output:
 - retval = number of bytes actually read/written or negative error number
   -EINVAL = invalid input parameters
   -ENOENT = requested object/version does not exist for reads
-  -EEXIST = requested version to write already exists
   -ENOSPC = no space left in the store for writes
-  -EAGAIN = temporary failure, operation should be retried (notably, writes may be retried
-    after stabilizing some journaled entries and freeing up journal space)
-  -EDOM = checksum error
+  -EDOM = checksum error.
 - version = the version actually read or written
 
 ## BS_OP_DELETE
@@ -93,7 +90,7 @@ Input:
 - version = requested version. Treated the same as with BS_OP_WRITE
 
 Output:
-- retval = 0 or negative error number (-EINVAL, -EAGAIN)
+- retval = 0 or negative error number (-EINVAL)
 - version = the version actually written (delete is initially written as an object version)
 
 ## BS_OP_SYNC
@@ -102,7 +99,7 @@ Make sure all previously issued modifications reach physical media.
 
 Input: Nothing except opcode
 Output:
-- retval = 0 or negative error number (-EINVAL, -EAGAIN)
+- retval = 0 or negative error number (-EINVAL)
 
 ## BS_OP_STABLE / BS_OP_ROLLBACK
 
@@ -132,9 +129,7 @@ Input:
 - max_oid = max inode/stripe or 0 to list all objects
 
 Output:
-- retval = total obj_ver_id count or negative error number
-  -ENOMEM = could not allocate memory for the listing
-  -EAGAIN = pool sharding settings are incorrect, please reshard and retry
+- retval = total obj_ver_id count
 - version = stable obj_ver_id count
 - buf = obj_ver_id array allocated by the blockstore. Stable versions come first.
   You must free it yourself after usage with free().

src/blockstore/blockstore_read.cpp

@@ -75,10 +75,11 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
                 return false;
             }
         }
-        if (need_skip && wr->type() == BS_HEAP_SMALL_WRITE &&
+        if (need_skip && (wr->type() == BS_HEAP_SMALL_WRITE || wr->type() == BS_HEAP_INTENT_WRITE) &&
             wr->small().offset < blk_end && wr->small().offset+wr->small().len > blk_start)
         {
             // Small write may mutate big write checksums during flush
+            // Intent write directly mutates physical data in-place
             skip_csum = COPY_BUF_SKIP_CSUM;
         }
         return true;

src/blockstore/blockstore_write.cpp

@@ -34,6 +34,7 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
         assert(it != modified_blocks.end());
         free(it->second.buf);
         modified_blocks.erase(it);
+        unsynced_meta_write_count++;
         heap->complete_block_write(modified_block);
         ringloop->wakeup();
     };
@@ -41,7 +42,6 @@ void blockstore_impl_t::prepare_meta_block_write(uint32_t modified_block)
     io_uring_prep_writev(
         sqe, dsk.meta_fd, &data->iov, 1, dsk.meta_offset + ((uint64_t)modified_block+1)*dsk.meta_block_size
     );
-    unsynced_meta_write_count++;
     pending_modified_blocks.push_back(modified_block);
     modified_blocks[modified_block] = { .sent = false, .buf = buf };
 }
@@ -243,12 +243,6 @@ enospc:
         uint64_t loc = !op->len ? 0 : heap->find_free_buffer_area(op->len);
         if (loc == UINT64_MAX)
         {
-            if (!heap->get_to_compact_count())
-            {
-                op->retval = -EAGAIN;
-                FINISH_OP(op);
-                return 2;
-            }
             PRIV(op)->wait_for = WAIT_COMPACTION;
             PRIV(op)->wait_detail = heap->get_compacted_count();
             flusher->request_trim();

src/blockstore/v1/flush.cpp

@@ -35,7 +35,6 @@ journal_flusher_t::journal_flusher_t(blockstore_impl_t *bs)
 
 journal_flusher_co::journal_flusher_co()
 {
-    wait_state = 0;
     simple_callback_r = [this](ring_data_t* data)
     {
         bs->live = true;
@@ -76,16 +75,6 @@ size_t journal_flusher_t::get_queue_size()
     return flush_queue.size();
 }
 
-bool journal_flusher_t::may_advance()
-{
-    return flush_queue.size() || active_flushers > 0 || trim_possible;
-}
-
-bool journal_flusher_t::may_advance_except_one()
-{
-    return flush_queue.size() || active_flushers > 1 || trim_possible;
-}
-
 void journal_flusher_t::loop()
 {
     target_flusher_count = bs->write_iodepth*2;
@@ -200,7 +189,6 @@ void journal_flusher_t::request_trim()
 
 void journal_flusher_t::mark_trim_possible()
 {
-    trim_possible = true;
     if (trim_wanted > 0)
     {
         dequeuing = true;
@@ -663,10 +651,9 @@ resume_2:
         flusher->sync_to_repeat.erase(repeat_it);
     trim_journal:
         // Clear unused part of the journal every <journal_trim_interval> flushes
-        if ((bs->journal_trim_interval && !((++flusher->journal_trim_counter) % bs->journal_trim_interval) ||
-            flusher->trim_wanted > 0) && flusher->trim_possible)
+        if (bs->journal_trim_interval && !((++flusher->journal_trim_counter) % bs->journal_trim_interval) ||
+            flusher->trim_wanted > 0)
         {
-            flusher->trim_possible = false;
     resume_30:
     resume_31:
     resume_32:
@@ -1508,19 +1495,10 @@ bool journal_flusher_co::trim_journal(int wait_base)
                 printf("Journal flushed\n");
                 exit(0);
             }
-            bs->wakeup_wait_journal();
-        }
-        else if (!flusher->may_advance_except_one())
-        {
-            bs->wakeup_wait_journal();
         }
         flusher->journal_trim_counter = 0;
         flusher->trimming = false;
     }
-    else if (!flusher->may_advance_except_one())
-    {
-        bs->wakeup_wait_journal();
-    }
     return true;
 }
 

src/blockstore/v1/flush.h

@@ -3,34 +3,34 @@
 
 struct copy_buffer_t
 {
-    int copy_flags;
-    uint64_t offset, len, disk_offset;
-    uint64_t journal_sector; // only for reads: sector+1 if used and !journal.inmemory, otherwise 0
-    void *buf;
-    uint8_t *csum_buf;
-    int *dyn_data;
+    int copy_flags = 0;
+    uint64_t offset = 0, len = 0, disk_offset = 0;
+    uint64_t journal_sector = 0; // only for reads: sector+1 if used and !journal.inmemory, otherwise 0
+    void *buf = NULL;
+    uint8_t *csum_buf = NULL;
+    int *dyn_data = NULL;
 };
 
 struct meta_sector_t
 {
-    uint64_t offset, len;
-    int state;
-    void *buf;
-    int usage_count;
+    uint64_t offset = 0, len = 0;
+    int state = 0;
+    void *buf = NULL;
+    int usage_count = 0;
 };
 
 struct flusher_sync_t
 {
-    bool fsync_meta;
-    int ready_count;
-    int state;
+    bool fsync_meta = false;
+    int ready_count = 0;
+    int state = 0;
 };
 
 struct flusher_meta_write_t
 {
-    uint64_t sector, pos;
-    bool submitted;
-    void *buf;
+    uint64_t sector = 0, pos = 0;
+    bool submitted = 0;
+    void *buf = NULL;
     std::map<uint64_t, meta_sector_t>::iterator it;
 };
 
@@ -39,36 +39,36 @@ class journal_flusher_t;
 // Journal flusher coroutine
 class journal_flusher_co
 {
-    blockstore_impl_t *bs;
-    journal_flusher_t *flusher;
-    int wait_state, wait_count, wait_journal_count;
-    struct io_uring_sqe *sqe;
-    struct ring_data_t *data;
+    blockstore_impl_t *bs = NULL;
+    journal_flusher_t *flusher = NULL;
+    int wait_state = 0, wait_count = 0, wait_journal_count = 0;
+    struct io_uring_sqe *sqe = NULL;
+    struct ring_data_t *data = NULL;
 
     std::list<flusher_sync_t>::iterator cur_sync;
 
-    obj_ver_id cur;
+    obj_ver_id cur = {};
     std::map<obj_ver_id, dirty_entry>::iterator dirty_it, dirty_start, dirty_end;
     std::map<object_id, uint64_t>::iterator repeat_it;
     std::function<void(ring_data_t*)> simple_callback_r, simple_callback_rj, simple_callback_w;
 
     bool try_trim = false;
-    bool skip_copy, has_delete, has_writes;
+    bool skip_copy = false, has_delete = false, has_writes = false;
     std::vector<copy_buffer_t> v;
     std::vector<copy_buffer_t>::iterator it;
-    int i;
-    bool fill_incomplete, cleared_incomplete;
-    int read_to_fill_incomplete;
-    int copy_count;
-    uint64_t clean_loc, clean_ver, old_clean_loc, old_clean_ver;
+    int i = 0;
+    bool fill_incomplete = false, cleared_incomplete = false;
+    int read_to_fill_incomplete = 0;
+    int copy_count = 0;
+    uint64_t clean_loc = 0, clean_ver = 0, old_clean_loc = 0, old_clean_ver = 0;
     flusher_meta_write_t meta_old, meta_new;
-    bool clean_init_bitmap;
-    uint64_t clean_bitmap_offset, clean_bitmap_len;
-    uint8_t *clean_init_dyn_ptr;
-    uint8_t *new_clean_bitmap;
+    bool clean_init_bitmap = false;
+    uint64_t clean_bitmap_offset = 0, clean_bitmap_len = 0;
+    uint8_t *clean_init_dyn_ptr = NULL;
+    uint8_t *new_clean_bitmap = NULL;
     std::unordered_set<uint32_t> mangle_csum_blocks;
 
-    uint64_t new_trim_pos;
+    uint64_t new_trim_pos = 0;
 
     friend class journal_flusher_t;
     void scan_dirty();
@@ -94,20 +94,19 @@ public:
 class journal_flusher_t
 {
     int trim_wanted = 0;
-    bool trim_possible = false;
-    bool dequeuing;
-    int min_flusher_count, max_flusher_count, cur_flusher_count, target_flusher_count;
-    int flusher_start_threshold;
-    journal_flusher_co *co;
-    blockstore_impl_t *bs;
+    bool dequeuing = false;
+    int min_flusher_count = 0, max_flusher_count = 0, cur_flusher_count = 0, target_flusher_count = 0;
+    int flusher_start_threshold = 0;
+    journal_flusher_co *co = NULL;
+    blockstore_impl_t *bs = NULL;
     friend class journal_flusher_co;
 
-    int journal_trim_counter;
-    bool trimming;
-    void* journal_superblock;
+    int journal_trim_counter = 0;
+    bool trimming = false;
+    void* journal_superblock = NULL;
 
-    int active_flushers;
-    int syncing_flushers;
+    int active_flushers = 0;
+    int syncing_flushers = 0;
     std::list<flusher_sync_t> syncs;
     std::map<object_id, uint64_t> sync_to_repeat;
 
@@ -118,14 +117,12 @@ class journal_flusher_t
 
     bool try_find_older(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
     bool try_find_other(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
-    bool may_advance_except_one();
 
 public:
     journal_flusher_t(blockstore_impl_t *bs);
     ~journal_flusher_t();
     void loop();
     bool is_trim_wanted() { return trim_wanted; }
-    bool may_advance();
     bool is_active();
     size_t get_queue_size();
     void mark_trim_possible();

src/blockstore/v1/impl.cpp

@@ -212,6 +212,10 @@ void blockstore_impl_t::loop()
                     // ring is full, stop submission
                     break;
                 }
+                else if (PRIV(op)->wait_for == WAIT_JOURNAL)
+                {
+                    PRIV(op)->wait_detail2 = (unstable_writes.size()+unstable_unsynced);
+                }
             }
         }
         if (op_idx != new_idx)
@@ -276,12 +280,6 @@ bool blockstore_impl_t::is_safe_to_stop()
     return true;
 }
 
-void blockstore_impl_t::wakeup_wait_journal()
-{
-    wait_journal_counter++;
-    ringloop->wakeup();
-}
-
 void blockstore_impl_t::check_wait(blockstore_op_t *op)
 {
     if (PRIV(op)->wait_for == WAIT_SQE)
@@ -298,11 +296,12 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
     }
     else if (PRIV(op)->wait_for == WAIT_JOURNAL)
     {
-        if (wait_journal_counter == PRIV(op)->wait_detail)
+        if (journal.used_start == PRIV(op)->wait_detail &&
+            (unstable_writes.size()+unstable_unsynced) == PRIV(op)->wait_detail2)
         {
             // do not submit
 #ifdef BLOCKSTORE_DEBUG
-            printf("Still waiting for flush journal counter %ju\n", PRIV(op)->wait_detail);
+            printf("Still waiting to flush journal offset %08jx\n", PRIV(op)->wait_detail);
 #endif
             return;
         }

src/blockstore/v1/impl.h

@@ -93,7 +93,7 @@ struct blockstore_op_private_t
 {
     // Wait status
     int wait_for;
-    uint64_t wait_detail;
+    uint64_t wait_detail, wait_detail2;
     int pending_ops;
     int op_state;
 
@@ -215,8 +215,6 @@ class blockstore_impl_t: public blockstore_i
     blockstore_init_meta* metadata_init_reader;
     blockstore_init_journal* journal_init_reader;
 
-    uint64_t wait_journal_counter = 0;
-    void wakeup_wait_journal();
     void check_wait(blockstore_op_t *op);
     void init_op(blockstore_op_t *op);
 

src/blockstore/v1/init.cpp

@@ -1000,7 +1000,7 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                         memcpy((uint8_t*)dyn+sizeof(int), dyn_from, dyn_size);
                     }
                     bs->dirty_db.emplace(ov, (dirty_entry){
-                        .state = (BS_ST_SMALL_WRITE | BS_ST_SYNCED | (je->type == JE_SMALL_WRITE_INSTANT ? BS_ST_INSTANT : 0)),
+                        .state = (BS_ST_SMALL_WRITE | BS_ST_SYNCED),
                         .flags = 0,
                         .location = location,
                         .offset = je->small_write.offset,
@@ -1084,7 +1084,7 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                         memcpy((uint8_t*)dyn+sizeof(int), dyn_from, dyn_size);
                     }
                     auto dirty_it = bs->dirty_db.emplace(ov, (dirty_entry){
-                        .state = (BS_ST_BIG_WRITE | BS_ST_SYNCED | (je->type == JE_BIG_WRITE_INSTANT ? BS_ST_INSTANT : 0)),
+                        .state = (BS_ST_BIG_WRITE | BS_ST_SYNCED),
                         .flags = 0,
                         .location = je->big_write.location,
                         .offset = je->big_write.offset,

src/blockstore/v1/journal.cpp

@@ -15,7 +15,6 @@ blockstore_journal_check_t::blockstore_journal_check_t(blockstore_impl_t *bs)
     first_sector = -1;
     next_in_pos = bs->journal.in_sector_pos;
     right_dir = next_pos >= bs->journal.used_start;
-    give_up = false;
 }
 
 // Check if we can write <required> entries of <size> bytes and <data_after> data bytes after them to the journal
@@ -120,20 +119,13 @@ int blockstore_journal_check_t::check_available(blockstore_op_t *op, int entries
         if (bs->log_level > 5)
         {
             printf(
-                "Ran out of journal space (used_start=%08jx, next_free=%08jx, dirty_start=%08jx, queued %zu)\n",
-                bs->journal.used_start, bs->journal.next_free, bs->journal.dirty_start, bs->flusher->get_queue_size()
+                "Ran out of journal space (used_start=%08jx, next_free=%08jx, dirty_start=%08jx)\n",
+                bs->journal.used_start, bs->journal.next_free, bs->journal.dirty_start
             );
         }
-        if (!bs->flusher->may_advance())
-        {
-            give_up = true;
-        }
-        else
-        {
-            PRIV(op)->wait_for = WAIT_JOURNAL;
-            PRIV(op)->wait_detail = bs->wait_journal_counter;
-            bs->flusher->request_trim();
-        }
+        PRIV(op)->wait_for = WAIT_JOURNAL;
+        bs->flusher->request_trim();
+        PRIV(op)->wait_detail = bs->journal.used_start;
         return 0;
     }
     return 1;

src/blockstore/v1/journal.h

@@ -70,7 +70,6 @@ struct blockstore_journal_check_t
     uint64_t next_pos, next_sector, next_in_pos;
     int sectors_to_write, first_sector;
     bool right_dir; // writing to the end or the beginning of the ring buffer
-    bool give_up;
 
     blockstore_journal_check_t(blockstore_impl_t *bs);
     int check_available(blockstore_op_t *op, int required, int size, int data_after);

src/blockstore/v1/rollback.cpp

@@ -63,12 +63,6 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, op->len, sizeof(journal_entry_rollback), 0))
     {
-        if (space_check.give_up)
-        {
-            op->retval = -EAGAIN;
-            FINISH_OP(op);
-            return 2;
-        }
         return 0;
     }
     // There is sufficient space. Check SQEs
@@ -168,10 +162,7 @@ void blockstore_impl_t::mark_rolled_back(const obj_ver_id & ov)
             if (unstab_it != unstable_writes.end())
             {
                 if (max_unstable == 0)
-                {
                     unstable_writes.erase(unstab_it);
-                    wakeup_wait_journal();
-                }
                 else
                     unstab_it->second = max_unstable;
             }

src/blockstore/v1/stable.cpp

@@ -362,12 +362,6 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, op->len, sizeof(journal_entry_stable), 0))
     {
-        if (space_check.give_up)
-        {
-            op->retval = -EAGAIN;
-            FINISH_OP(op);
-            return 2;
-        }
         return 0;
     }
     // There is sufficient space. Check SQEs
@@ -451,9 +445,9 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
                 {
                     break;
                 }
-                if (!IS_STABLE(back_it->second.state))
+                if (!IS_SYNCED(back_it->second.state))
                 {
-                    // There are preceding unstable versions, can't flush <v>
+                    // There are preceding unsynced versions, can't flush <v>
                     return;
                 }
             }
@@ -563,7 +557,6 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
         unstab_it->second <= v.version)
     {
         unstable_writes.erase(unstab_it);
-        wakeup_wait_journal();
     }
 }
 

src/blockstore/v1/sync.cpp

@@ -79,7 +79,6 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
         // 2nd step: Data device is synced, prepare & write journal entries
         // Check space in the journal and journal memory buffers
         blockstore_journal_check_t space_check(this);
-        bool space_ok = true;
         if (dsk.csum_block_size)
         {
             // More complex check because all journal entries have different lengths
@@ -91,34 +90,13 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
                 uint64_t dyn_size = dsk.dirty_dyn_size(dirty_entry.offset, dirty_entry.len);
                 if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size, 0))
                 {
-                    space_ok = false;
-                    break;
+                    return 0;
                 }
             }
         }
         else if (!space_check.check_available(op, PRIV(op)->sync_big_writes.size(),
             sizeof(journal_entry_big_write) + dsk.clean_entry_bitmap_size, 0))
         {
-            space_ok = false;
-        }
-        if (!space_ok)
-        {
-            if (space_check.give_up)
-            {
-                // Return everything back
-                unsynced_big_write_count += PRIV(op)->sync_big_writes.size();
-                PRIV(op)->sync_big_writes.insert(PRIV(op)->sync_big_writes.end(),
-                    unsynced_big_writes.begin(), unsynced_big_writes.end());
-                PRIV(op)->sync_big_writes.swap(unsynced_big_writes);
-                PRIV(op)->sync_big_writes.clear();
-                PRIV(op)->sync_small_writes.insert(PRIV(op)->sync_small_writes.end(),
-                    unsynced_small_writes.begin(), unsynced_small_writes.end());
-                PRIV(op)->sync_small_writes.swap(unsynced_small_writes);
-                PRIV(op)->sync_small_writes.clear();
-                op->retval = -EAGAIN;
-                FINISH_OP(op);
-                return 2;
-            }
             return 0;
         }
         // Check SQEs. Don't bother about merging, submit each journal sector as a separate request
@@ -198,62 +176,53 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
 void blockstore_impl_t::ack_sync(blockstore_op_t *op)
 {
     // Handle states
-    for (auto it = PRIV(op)->sync_big_writes.begin(); it != PRIV(op)->sync_big_writes.end(); it++)
+    auto mark_sync = [&](obj_ver_id & ov)
     {
-#ifdef BLOCKSTORE_DEBUG
-        printf("Ack sync big %jx:%jx v%ju\n", it->oid.inode, it->oid.stripe, it->version);
-#endif
-        auto & unstab = unstable_writes[it->oid];
-        unstab = unstab < it->version ? it->version : unstab;
-        auto dirty_it = dirty_db.find(*it);
+        auto dirty_it = dirty_db.find(ov);
+        bool is_big = ((dirty_it->second.state & BS_ST_TYPE_MASK) == BS_ST_BIG_WRITE);
         dirty_it->second.state = ((dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_SYNCED);
-        if (dirty_it->second.state & BS_ST_INSTANT)
+        uint64_t stable_ver = 0;
+        if (IS_INSTANT(dirty_it->second.state))
         {
-            mark_stable(dirty_it->first);
+            stable_ver = ov.version;
         }
         else
         {
             unstable_unsynced--;
             assert(unstable_unsynced >= 0);
-            wakeup_wait_journal();
         }
         dirty_it++;
-        while (dirty_it != dirty_db.end() && dirty_it->first.oid == it->oid)
+        while (dirty_it != dirty_db.end() && dirty_it->first.oid == ov.oid)
         {
-            if ((dirty_it->second.state & BS_ST_WORKFLOW_MASK) == BS_ST_WAIT_BIG)
+            if (is_big && (dirty_it->second.state & BS_ST_WORKFLOW_MASK) == BS_ST_WAIT_BIG)
             {
                 dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_IN_FLIGHT;
             }
+            else if ((dirty_it->second.state & BS_ST_WORKFLOW_MASK) == BS_ST_SYNCED &&
+                IS_INSTANT(dirty_it->second.state))
+            {
+                stable_ver = dirty_it->first.version;
+            }
             dirty_it++;
         }
-    }
-    for (auto it = PRIV(op)->sync_small_writes.begin(); it != PRIV(op)->sync_small_writes.end(); it++)
+        if (stable_ver != 0)
+        {
+            mark_stable((obj_ver_id){ .oid = ov.oid, .version = stable_ver });
+        }
+    };
+    for (auto & ov: PRIV(op)->sync_big_writes)
     {
 #ifdef BLOCKSTORE_DEBUG
-        printf("Ack sync small %jx:%jx v%ju\n", it->oid.inode, it->oid.stripe, it->version);
+        printf("Ack sync big %jx:%jx v%ju\n", ov.oid.inode, ov.oid.stripe, ov.version);
 #endif
-        auto & unstab = unstable_writes[it->oid];
-        unstab = unstab < it->version ? it->version : unstab;
-        if (dirty_db[*it].state == (BS_ST_DELETE | BS_ST_WRITTEN))
-        {
-            dirty_db[*it].state = (BS_ST_DELETE | BS_ST_SYNCED);
-            // Deletions are treated as immediately stable
-            mark_stable(*it);
-        }
-        else /* (BS_ST_INSTANT?) | BS_ST_SMALL_WRITE | BS_ST_WRITTEN */
-        {
-            dirty_db[*it].state = (dirty_db[*it].state & ~BS_ST_WORKFLOW_MASK) | BS_ST_SYNCED;
-            if (dirty_db[*it].state & BS_ST_INSTANT)
-            {
-                mark_stable(*it);
-            }
-            else
-            {
-                unstable_unsynced--;
-                assert(unstable_unsynced >= 0);
-                wakeup_wait_journal();
-            }
-        }
+        mark_sync(ov);
+    }
+    for (auto & ov: PRIV(op)->sync_small_writes)
+    {
+#ifdef BLOCKSTORE_DEBUG
+        printf("Ack sync small %jx:%jx v%ju\n", ov.oid.inode, ov.oid.stripe, ov.version);
+#endif
+        mark_sync(ov);
     }
     op->retval = 0;
     FINISH_OP(op);

src/blockstore/v1/write.cpp

@@ -39,11 +39,6 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
         dirty_it--; // segfaults when dirty_db is empty
         if (dirty_it != dirty_db.end() && dirty_it->first.oid == op->oid)
         {
-            if (op->opcode == BS_OP_WRITE_STABLE && !IS_INSTANT(dirty_it->second.state))
-            {
-                op->retval = -EINVAL;
-                return false;
-            }
             found = true;
             version = dirty_it->first.version + 1;
             deleted = IS_DELETE(dirty_it->second.state);
@@ -317,11 +312,6 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             sizeof(journal_entry_big_write) + dsk.clean_dyn_size,
             (unstable_writes.size()+unstable_unsynced+((dirty_it->second.state & BS_ST_INSTANT) ? 0 : 1))*journal.block_size))
         {
-            if (space_check.give_up)
-            {
-                cancel_all_writes(op, dirty_it, -EAGAIN);
-                return 2;
-            }
             return 0;
         }
         // Big (redirect) write
@@ -414,11 +404,6 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
                 sizeof(journal_entry_small_write) + dyn_size,
                 op->len + (unstable_writes.size()+unstable_unsynced+((dirty_it->second.state & BS_ST_INSTANT) ? 0 : 1))*journal.block_size))
         {
-            if (space_check.give_up)
-            {
-                cancel_all_writes(op, dirty_it, -EAGAIN);
-                return 2;
-            }
             return 0;
         }
         // There is sufficient space. Check SQE(s)
@@ -574,25 +559,6 @@ resume_2:
         if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size,
             (unstable_writes.size()+unstable_unsynced+((dirty_it->second.state & BS_ST_INSTANT) ? 0 : 1))*journal.block_size))
         {
-            if (space_check.give_up)
-            {
-                auto uo_it = used_clean_objects.find(dirty_it->second.location);
-                if (uo_it != used_clean_objects.end())
-                    uo_it->second.was_freed = true;
-                else
-                    data_alloc->set(dirty_it->second.location / dsk.data_block_size, false);
-                if (!(dirty_it->second.state & BS_ST_INSTANT))
-                {
-                    unstable_unsynced--;
-                    assert(unstable_unsynced >= 0);
-                    wakeup_wait_journal();
-                }
-                if (immediate_commit != IMMEDIATE_ALL)
-                    unsynced_big_write_count--;
-                write_iodepth--;
-                cancel_all_writes(op, dirty_it, -EAGAIN);
-                return 2;
-            }
             return 0;
         }
         BS_SUBMIT_CHECK_SQES(1);
@@ -656,7 +622,6 @@ resume_4:
             {
                 unstable_unsynced--;
                 assert(unstable_unsynced >= 0);
-                wakeup_wait_journal();
             }
         }
         dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK)
@@ -809,11 +774,6 @@ int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, 1, sizeof(journal_entry_del), (unstable_writes.size()+unstable_unsynced)*journal.block_size))
     {
-        if (space_check.give_up)
-        {
-            cancel_all_writes(op, dirty_it, -EAGAIN);
-            return 2;
-        }
         return 0;
     }
     // Write current journal sector only if it's dirty and full, or in the immediate_commit mode

src/client/cluster_client.cpp

@@ -1250,11 +1250,9 @@ resume_2:
         return 1;
     }
     else if (op->retval != 0 && op->opcode != OSD_OP_SYNC && !(op->flags & OP_FLUSH_BUFFER) &&
-        op->retval != -EPIPE && op->retval != -EAGAIN &&
-        (op->retval != -EIO || !client_eio_retry_interval) &&
-        (op->retval != -ENOSPC || !client_retry_enospc))
+        op->retval != -EPIPE && (op->retval != -EIO || !client_eio_retry_interval) && (op->retval != -ENOSPC || !client_retry_enospc))
     {
-        // Fatal error (neither -EPIPE, -EAGAIN, -EIO nor -ENOSPC)
+        // Fatal error (neither -EPIPE, -EIO nor -ENOSPC)
         erase_op(op);
         return 1;
     }
@@ -1613,7 +1611,7 @@ resume_1:
     if (op->retval != 0)
     {
         wb->fsync_error();
-        if (op->retval == -EPIPE || op->retval == -EAGAIN || op->retval == -EIO || op->retval == -ENOSPC)
+        if (op->retval == -EPIPE || op->retval == -EIO || op->retval == -ENOSPC)
         {
             // Retry later
             op->parts.clear();
@@ -1664,6 +1662,9 @@ static inline void mem_or(void *res, const void *r2, unsigned int len)
     }
 }
 
+// Error priority: others > EPERM > EIO > ENOSPC > ETIMEDOUT > EPIPE
+#define ERR_PRIO(e) (((e) == -EPERM ? 5 : ((e) == -EIO ? 4 : ((e) == -ENOSPC ? 3 : ((e) == -ETIMEDOUT ? 2 : ((e) == -EPIPE ? 1 : (!(e) ? 0 : 10)))))))
+
 void cluster_client_t::handle_op_part(cluster_op_part_t *part)
 {
     cluster_op_t *op = part->parent;
@@ -1675,7 +1676,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
         if (ERR_PRIO(part->op.reply.hdr.retval) > ERR_PRIO(op->retval))
             op->retval = part->op.reply.hdr.retval;
         uint64_t stop_client_id = 0;
-        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC && op->retval != -EPERM && op->retval != -EAGAIN)
+        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC && op->retval != -EPERM)
         {
             stop_client_id = part->op.client_id;
             if (op->retval != -EPIPE || log_level > 0)
@@ -1697,11 +1698,11 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
         // So do all these things after modifying operation state, otherwise we may hit reenterability bugs
         // FIXME postpone such things to set_immediate here to avoid bugs
         // Set op->retry_after to retry operation after a short pause (not immediately)
-        if (!op->retry_after && (op->retval == -EPIPE || op->retval == -EAGAIN ||
+        if (!op->retry_after && (op->retval == -EPIPE ||
             op->retval == -EIO && client_eio_retry_interval ||
             op->retval == -ENOSPC && client_retry_enospc))
         {
-            op->retry_after = op->retval != -EPIPE && op->retval != -EAGAIN ? client_eio_retry_interval : client_retry_interval;
+            op->retry_after = op->retval != -EPIPE ? client_eio_retry_interval : client_retry_interval;
         }
         reset_retry_timer(op->retry_after);
         if (stop_client_id)

src/client/cluster_client_list.cpp

@@ -370,7 +370,8 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
             fprintf(stderr, "Failed to get PG %u/%u object list from OSD %ju (retval=%jd), skipping\n",
                 cur_list->pg->lst->pool_id, cur_list->pg->pg_num, cur_list->osd_num, op->reply.hdr.retval);
             if (!cur_list->pg->errcode ||
-                ERR_PRIO(op->reply.hdr.retval) > ERR_PRIO(cur_list->pg->errcode))
+                cur_list->pg->errcode == -EPIPE ||
+                op->reply.hdr.retval != -EPIPE)
             {
                 cur_list->pg->errcode = op->reply.hdr.retval;
             }
@@ -433,7 +434,7 @@ void cluster_client_t::finish_list_pg(inode_list_pg_t *pg, bool retry_epipe)
     auto lst = pg->lst;
     if (pg->inflight_ops == 0)
     {
-        if ((pg->errcode == -EPIPE || pg->errcode == -EAGAIN) && retry_epipe)
+        if (pg->errcode == -EPIPE && retry_epipe)
         {
             // Retry listing after <client_retry_interval> ms on EPIPE
             pg->state = LIST_PG_WAIT_RETRY;

src/client/object_id.h

@@ -111,15 +111,4 @@ namespace std
             return seed;
         }
     };
-
-    template<> struct hash<pool_pg_num_t>
-    {
-        inline size_t operator()(const pool_pg_num_t &s) const
-        {
-            size_t seed = 0;
-            seed ^= ((size_t)s.pool_id + 0xc6a4a7935bd1e995 + (seed << 6) + (seed >> 2));
-            seed ^= ((size_t)s.pg_num + 0xc6a4a7935bd1e995 + (seed << 6) + (seed >> 2));
-            return seed;
-        }
-    };
 }

src/client/osd_ops.h

@@ -57,9 +57,6 @@
 #define OSD_SEC_LOCK_PG 1
 #define OSD_SEC_UNLOCK_PG 2
 
-// Error priority: others > EPERM > EIO > EDOM > ENOSPC > EPIPE > EAGAIN
-#define ERR_PRIO(e) (((e) == -EPERM ? 6 : ((e) == -EIO ? 5 : ((e) == -EDOM ? 4 : ((e) == -ENOSPC ? 3 : ((e) == -EPIPE ? 2 : ((e) == -EAGAIN ? 1 : (!(e) ? 0 : 10))))))))
-
 // common request and reply headers
 struct __attribute__((__packed__)) osd_op_header_t
 {

src/osd/osd_flush.cpp

@@ -81,9 +81,18 @@ void osd_t::handle_flush_op(bool rollback, pool_id_t pool_id, pg_num_t pg_num, p
     if (retval != 0)
     {
         fb->flush_errors++;
-        printf("Error while doing flush on OSD %ju: %d (%s)\n", peer_osd, retval, strerror(-retval));
-        if (peer_osd != this->osd_num && retval != -EAGAIN)
+        if (peer_osd == this->osd_num)
         {
+            throw std::runtime_error(
+                std::string(rollback
+                    ? "Error while doing local rollback operation: "
+                    : "Error while doing local stabilize operation: "
+                ) + strerror(-retval)
+            );
+        }
+        else
+        {
+            printf("Error while doing flush on OSD %ju: %d (%s)\n", osd_num, retval, strerror(-retval));
             auto peer_it = msgr.osd_peers.find(peer_osd);
             if (peer_it != msgr.osd_peers.end())
             {

src/osd/osd_primary_subops.cpp

@@ -311,10 +311,11 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop)
     blockstore_op_t *bs_op = subop->bs_op;
     int64_t expected = bs_op->opcode == BS_OP_READ || bs_op->opcode == BS_OP_WRITE
         || bs_op->opcode == BS_OP_WRITE_STABLE ? bs_op->len : 0;
-    if (bs_op->retval != expected && bs_op->opcode != BS_OP_READ && bs_op->retval != -EAGAIN &&
-        (bs_op->opcode != BS_OP_WRITE && bs_op->opcode != BS_OP_WRITE_STABLE || bs_op->retval != -ENOSPC))
+    if (bs_op->retval != expected && bs_op->opcode != BS_OP_READ &&
+        (bs_op->opcode != BS_OP_WRITE && bs_op->opcode != BS_OP_WRITE_STABLE ||
+        bs_op->retval != -ENOSPC))
     {
-        // die on any error except ENOSPC or EAGAIN during write
+        // die on any error except ENOSPC during write
         if (bs_op->opcode == BS_OP_WRITE || bs_op->opcode == BS_OP_WRITE_STABLE)
         {
             printf(
@@ -427,7 +428,7 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
 #endif
         if (!(op_data->flags & OP_DATA_SKIP_VER_CHECK))
         {
-            if (op_data->fact_ver != version && op_data->done > 0 &&
+            if (op_data->fact_ver != version && (op_data->errors > 0 || op_data->done > 0) &&
                 (version != 0 && op_data->fact_ver != 0 || !(op_data->flags & OP_DATA_ENOENT_OK)))
             {
                 fprintf(
@@ -472,12 +473,16 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
             );
         }
         subop->rmw_buf = NULL;
-        // Error priority: ENOSPC > others > EIO > EDOM > EPIPE > EAGAIN
-        if (op_data->errcode == 0 || ERR_PRIO(retval) > ERR_PRIO(op_data->errcode))
+        // Error priority: ENOSPC > others > EIO > EDOM > EPIPE
+        if (op_data->errcode == 0 ||
+            retval == -ENOSPC && op_data->errcode != -ENOSPC ||
+            retval == -EIO && (op_data->errcode == -EDOM || op_data->errcode == -EPIPE) ||
+            retval == -EDOM && (op_data->errcode == -EPIPE) ||
+            retval != -EIO && retval != -EDOM && retval != -EPIPE)
         {
             op_data->errcode = retval;
         }
-        if (subop->client_id && retval != -EDOM && retval != -ERANGE && retval != -EAGAIN &&
+        if (subop->client_id && retval != -EDOM && retval != -ERANGE &&
             (retval != -ENOSPC || opcode != OSD_OP_SEC_WRITE && opcode != OSD_OP_SEC_WRITE_STABLE) &&
             (retval != -EIO || opcode != OSD_OP_SEC_READ))
         {
@@ -485,32 +490,6 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
             msgr.stop_client(subop->client_id);
             op_data->drops++;
         }
-        else if (opcode == OSD_OP_SEC_STABILIZE || opcode == OSD_OP_SEC_ROLLBACK)
-        {
-            // STABILIZE/ROLLBACK should never fail, but if they do, we repeer the PGs
-            assert(subop->iov.count == 1);
-            obj_ver_id *vers = (obj_ver_id*)subop->iov.buf[0].iov_base;
-            size_t n = subop->iov.buf[0].iov_len/sizeof(obj_ver_id);
-            robin_hood::unordered_flat_set<pool_pg_num_t> to_repeer;
-            for (size_t i = 0; i < n; i++)
-            {
-                to_repeer.insert((pool_pg_num_t){
-                    .pool_id = INODE_POOL(vers[n].oid.inode),
-                    .pg_num = map_to_pg(vers[n].oid),
-                });
-            }
-            ringloop->set_immediate([this, to_repeer = std::move(to_repeer)]()
-            {
-                for (auto & ppg: to_repeer)
-                {
-                    auto pg_it = pgs.find(ppg);
-                    if (pg_it != pgs.end() && (pg_it->second.state & PG_ACTIVE))
-                    {
-                        repeer_pg(pg_it->second);
-                    }
-                }
-            });
-        }
         // Increase op_data->errors after stop_client to prevent >= n_subops running twice
         op_data->errors++;
     }
@@ -724,7 +703,6 @@ void osd_t::submit_primary_stab_subops(osd_op_t *cur_op)
         {
             clock_gettime(CLOCK_REALTIME, &subops[i].tv_begin);
             subops[i].op_type = (uint64_t)cur_op;
-            subops[i].iov.push_back(op_data->unstable_writes + stab_osd.start, stab_osd.len * sizeof(obj_ver_id));
             subops[i].bs_op = new blockstore_op_t((blockstore_op_t){
                 .opcode = BS_OP_STABLE,
                 .callback = [subop = &subops[i], this](blockstore_op_t *bs_subop)
@@ -746,7 +724,7 @@ void osd_t::submit_primary_stab_subops(osd_op_t *cur_op)
                     .magic = SECONDARY_OSD_OP_MAGIC,
                     .opcode = OSD_OP_SEC_STABILIZE,
                 },
-                .len = (uint64_t)stab_osd.len * sizeof(obj_ver_id),
+                .len = (uint64_t)(stab_osd.len * sizeof(obj_ver_id)),
                 .flags = cur_op->client_id == SELF_CLIENT && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0,
             } };
             subops[i].iov.push_back(op_data->unstable_writes + stab_osd.start, stab_osd.len * sizeof(obj_ver_id));
@@ -800,7 +778,6 @@ void osd_t::submit_primary_rollback_subops(osd_op_t *cur_op, const uint64_t* osd
             {
                 clock_gettime(CLOCK_REALTIME, &subop->tv_begin);
                 subop->op_type = (uint64_t)cur_op;
-                subop->iov.push_back(op_data->unstable_writes + i, sizeof(obj_ver_id));
                 subop->bs_op = new blockstore_op_t((blockstore_op_t){
                     .opcode = BS_OP_ROLLBACK,
                     .callback = [subop, this](blockstore_op_t *bs_subop)

src/osd/osd_primary_write.cpp

@@ -258,21 +258,9 @@ resume_4:
         return;
     }
 resume_5:
-    if (!(pg.state & PG_ACTIVE))
-    {
-        if (op_data->subops)
-        {
-            delete[] op_data->subops;
-            op_data->subops = NULL;
-        }
-        pg.ver_override.erase(op_data->oid);
-        deref_object_state(pg, &op_data->object_state, true);
-        pg_cancel_write_queue(pg, cur_op, op_data->oid, -EPIPE);
-        return;
-    }
     if (op_data->errors > 0)
     {
-        // Handle ENOSPC/EAGAIN/EDOM/ERANGE/EIO. If some subops fail, but others succeed,
+        // Handle ENOSPC/EDOM/ERANGE/EIO. If some subops fail, but others succeed,
         // next writes to the same object will also fail because they'll try
         // to overwrite the same version number which will result in EEXIST.
         // To fix it, we should mark the object as degraded for replicas,
@@ -281,11 +269,6 @@ resume_5:
         {
             if (pg.scheme != POOL_SCHEME_REPLICATED)
             {
-                if (op_data->subops)
-                {
-                    delete[] op_data->subops;
-                    op_data->subops = NULL;
-                }
                 submit_primary_rollback_subops(cur_op, pg.cur_set.data());
 resume_11:
                 if (op_data->n_subops > 0)
@@ -294,17 +277,21 @@ resume_11:
                     return;
                 }
 resume_12:
-                if (op_data->unstable_writes)
-                {
-                    delete[] op_data->unstable_writes;
-                    op_data->unstable_writes = NULL;
-                }
-                // Ignore ROLLBACK errors - submit_primary_subops will repeer the PG if it fails
+                // Ignore ROLLBACK errors - submit_primary_subops will drop the connection if it fails
+                delete[] op_data->unstable_writes;
+                op_data->unstable_writes = NULL;
             }
             else
             {
                 pg.ver_override.erase(op_data->oid);
                 mark_partial_write(pg, cur_op);
+                if (op_data->subops)
+                {
+                    delete[] op_data->subops;
+                    op_data->subops = NULL;
+                }
+                pg_cancel_write_queue(pg, cur_op, op_data->oid, op_data->errcode);
+                return;
             }
         }
         if (op_data->subops)
@@ -317,6 +304,13 @@ resume_12:
         cur_op->reply.hdr.retval = op_data->errcode;
         goto continue_others;
     }
+    if (!(pg.state & PG_ACTIVE))
+    {
+        pg.ver_override.erase(op_data->oid);
+        deref_object_state(pg, &op_data->object_state, true);
+        pg_cancel_write_queue(pg, cur_op, op_data->oid, -EPIPE);
+        return;
+    }
     if (pg.scheme != POOL_SCHEME_REPLICATED)
     {
         // Remove version override just after the write, but before stabilizing
@@ -561,7 +555,6 @@ resume_6:
                 return false;
             }
 resume_7:
-            // Ignore STABILIZE errors - submit_primary_subops will repeer the PG if it fails
             // FIXME: Free those in the destructor?
             delete op_data->unstable_write_osds;
             delete[] op_data->unstable_writes;

src/test/test_blockstore.cpp

@@ -137,13 +137,12 @@ struct bs_test_t
         }
     }
 
-    void exec_op(blockstore_op_t *op, bool print = true)
+    void exec_op(blockstore_op_t *op)
     {
         bool done = false;
         op->callback = [&](blockstore_op_t *op)
         {
-            if (print)
-                printf("op opcode=%ju completed retval=%jd\n", op->opcode, op->retval);
+            printf("op opcode=%ju completed retval=%jd\n", op->opcode, op->retval);
             done = true;
         };
         bs->enqueue_op(op);
@@ -363,6 +362,121 @@ static void test_fsync(bool separate_meta)
     free(op2.buf);
 }
 
+static void test_fsync_meta_before_complete()
+{
+    printf("\n-- test_fsync_meta_before_complete\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["disable_data_fsync"] = "0";
+    test.config["immediate_commit"] = "none";
+    test.config["meta_device"] = "./test_meta.bin";
+    test.config["disable_meta_fsync"] = "0";
+    test.config["meta_device_size"] = "33554432";
+    test.config["meta_device_sect"] = "4096";
+    test.config["data_offset"] = "0";
+    test.init();
+    test.meta_disk->trace = 1;
+
+    // Write
+    printf("writing 16K+4K v1\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 16384;
+    op.len = 4096;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
+    memset(op.buf, 0xaa, 4096);
+
+    // Block metadata write execution
+    uint64_t mod_off = 0;
+    void *mod_buf = NULL;
+    ring_data_t *mod_data = NULL;
+    test.sqe_handler = [&](io_uring_sqe *sqe)
+    {
+        if (sqe->fd == MOCK_META_FD && sqe->opcode == IORING_OP_WRITEV)
+        {
+            assert(sqe->len == 1);
+            mod_off = sqe->off;
+            mod_buf = ((iovec*)sqe->addr)[0].iov_base;
+            mod_data = (ring_data_t*)sqe->user_data;
+            assert(((iovec*)sqe->addr)[0].iov_len == test.bs->dsk.meta_block_size);
+            return true;
+        }
+        return false;
+    };
+
+    bool wr_done = false;
+    op.callback = [&](blockstore_op_t *op)
+    {
+        printf("op opcode=%ju completed retval=%jd\n", op->opcode, op->retval);
+        wr_done = true;
+    };
+    test.bs->enqueue_op(&op);
+
+    while (!mod_buf)
+        test.ringloop->loop();
+
+    // Sync
+    printf("syncing\n");
+    blockstore_op_t sync_op;
+    sync_op.opcode = BS_OP_SYNC;
+    test.exec_op(&sync_op);
+    assert(sync_op.retval == 0);
+
+    // Now send the metadata write
+    {
+        ring_data_t data;
+        io_uring_sqe sqe;
+        iovec iov;
+        sqe.opcode = IORING_OP_WRITEV;
+        sqe.off = test.bs->dsk.meta_offset + mod_off;
+        sqe.addr = (uint64_t)&iov;
+        sqe.len = 1;
+        sqe.rw_flags = 0;
+        iov.iov_base = mod_buf;
+        iov.iov_len = test.bs->dsk.meta_block_size;
+        sqe.user_data = (uint64_t)mod_data;
+        bool ok = test.meta_disk->submit(&sqe);
+        assert(ok);
+        test.ringloop->mark_completed(mod_data);
+    }
+
+    while (!wr_done)
+        test.ringloop->loop();
+    assert(op.retval == op.len);
+
+    // Sync again
+    printf("syncing again\n");
+    test.exec_op(&sync_op);
+    assert(sync_op.retval == 0);
+
+    // Discard and restart
+    printf("destroying again\n");
+    test.destroy_bs();
+    test.meta_disk->discard_buffers(true, 0);
+    test.init();
+
+    // Check that the write isn't lost
+    printf("checking for OK\n");
+    blockstore_op_t op2;
+    op2.opcode = BS_OP_READ;
+    op2.oid = { .inode = 1, .stripe = 0 };
+    op2.version = UINT64_MAX;
+    op2.offset = 0;
+    op2.len = 128*1024;
+    op2.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    test.exec_op(&op2);
+    assert(op2.retval == op2.len);
+    assert(is_zero(op2.buf, 16*1024));
+    assert(memcmp(op2.buf+16*1024, op.buf, 4*1024) == 0);
+    assert(is_zero(op2.buf+20*1024, 108*1024));
+
+    free(op.buf);
+    free(op2.buf);
+}
+
 static void test_intent_over_unstable()
 {
     printf("\n-- test_intent_over_unstable\n");
@@ -977,60 +1091,61 @@ static void test_list_limit()
     free(op.buf);
 }
 
-static void test_write_no_space_eagain()
+static void test_intent_write_corruption()
 {
-    printf("\n-- test_write_no_space_eagain\n");
+    printf("\n-- test_intent_write_corruption\n");
 
     bs_test_t test;
     test.default_cfg();
-    // 1 MB journal (data_offset 32 MB - journal_offset)
-    test.config["journal_offset"] = std::to_string(31*1024*1024);
+    test.config["csum_block_size"] = "16384";
+    test.config["bitmap_granularity"] = "4096";
+    test.config["perfect_csum_update"] = "false";
     test.init();
-    printf("blockstore initialized\n");
 
-    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
-    memset(buf, 0xAA, 4096);
+    // 1. Write BIG_WRITE (0-48K)
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 49152; // 48K
+    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);
 
-    // fill the journal with unstable writes
-    int i = 0, blocked_retval = 0;
-    auto do_write = [&]()
-    {
-        blockstore_op_t op;
-        op.opcode = BS_OP_WRITE;
-        op.oid = { .inode = 1, .stripe = (uint64_t)(i % 4) * 0x20000 };
-        op.version = 1 + i / 4;
-        op.offset = 0;
-        op.len = 4096;
-        op.buf = buf;
-        test.exec_op(&op, false);
-        if (op.retval != (int)op.len)
-            blocked_retval = op.retval;
-    };
-    for (i = 0; i < 100000 && !blocked_retval; i++)
-        do_write();
-    printf("write #%d refused with retval=%d\n", i, blocked_retval);
-    assert(blocked_retval == -EAGAIN);
 
-    // check that a stabilize operation and one more write can proceed
-    {
-        blockstore_op_t op;
-        op.opcode = BS_OP_STABLE;
-        op.len = 1;
-        op.buf = (uint8_t*)malloc_or_die(1 * sizeof(obj_ver_id));
-        ((obj_ver_id*)op.buf)[0] = {
-            .oid = { .inode = 1, .stripe = 0 },
-            .version = 2,
-        };
-        test.exec_op(&op);
-        assert(op.retval == 0);
-        free(op.buf);
-    }
+    // 2. Write unaligned INTENT_WRITE B (8K-12K)
+    free(op.buf);
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 2;
+    op.offset = 8192; // 8K
+    op.len = 4096; // 4K
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op.len);
+    memset(op.buf, 0xbb, op.len);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+    
+    // 3. Read 12-40K
+    // This will hit the hole logic where B provided 8-12K, but we don't even request it.
+    // The request is for 12-40K.
+    // 12K is not aligned to 16K, so A must pad by reading 0-16K.
+    // However, 8-12K of the physical block has been overwritten by B.
+    // A will try to verify the 0-16K block with its old checksum and fail.
+    free(op.buf);
+    op.opcode = BS_OP_READ;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = UINT64_MAX; // latest version
+    op.offset = 12288; // 12K
+    op.len = 28672; // 28K (so end = 40K)
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op.len);
+    test.exec_op(&op);
 
-    blocked_retval = 0;
-    do_write();
-    assert(!blocked_retval);
+    // Assert that the read succeeds without checksum corruption errors
+    assert(op.retval == op.len);
 
-    free(buf);
+    free(op.buf);
 }
 
 // FIXME Add a simple intent_write / big_intent test
@@ -1040,7 +1155,9 @@ int main(int narg, char *args[])
     test_simple();
     test_fsync(false);
     test_fsync(true);
+    test_fsync_meta_before_complete();
     test_intent_over_unstable();
+    test_intent_write_corruption();
     test_padded_csum_intent(false);
     test_padded_csum_intent(true);
     test_padded_csum_sparse_leading_hole();
@@ -1053,6 +1170,5 @@ int main(int narg, char *args[])
     test_fsync_almost_batch_big();
     test_fsync_batch_big();
     test_list_limit();
-    test_write_no_space_eagain();
     return 0;
 }

src/test/test_blockstore_v1.cpp

@@ -883,75 +883,6 @@ static void test_meta_tail_reload()
     printf("blockstore re-initialized with an overflowing entry\n");
 }
 
-static void test_write_no_space_eagain()
-{
-    printf("\n-- test_write_no_space_eagain\n");
-
-    bs_test_t test;
-    test.default_cfg();
-    // 1 MB journal (data_offset 32 MB - journal_offset), so it fills in a few hundred writes
-    test.config["journal_offset"] = std::to_string(31*1024*1024);
-    test.init();
-    printf("blockstore initialized\n");
-
-    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
-    memset(buf, 0xAA, 4096);
-
-    // fill the journal with unstable writes
-    int i = 0, blocked_retval = 0;
-    auto do_write = [&]()
-    {
-        blockstore_op_t op;
-        op.opcode = BS_OP_WRITE;
-        op.oid = { .inode = 1, .stripe = (uint64_t)(i % 4) * 0x20000 };
-        op.version = 1 + i / 4;
-        op.offset = 0;
-        op.len = 4096;
-        op.buf = buf;
-        test.exec_op(&op);
-        if (op.retval != (int)op.len)
-            blocked_retval = op.retval;
-    };
-    for (i = 0; i < 100000 && !blocked_retval; i++)
-        do_write();
-    printf("write #%d refused with retval=%d\n", i, blocked_retval);
-    assert(blocked_retval == -EAGAIN);
-
-    // check that a stabilize operation and one more write can proceed
-    // journal space_check is simplified so it requires slightly more space than actually needed
-    {
-        blockstore_op_t op;
-        op.opcode = BS_OP_STABLE;
-        op.len = 4;
-        op.buf = (uint8_t*)malloc_or_die(4 * sizeof(obj_ver_id));
-        ((obj_ver_id*)op.buf)[0] = {
-            .oid = { .inode = 1, .stripe = 0 },
-            .version = 2,
-        };
-        ((obj_ver_id*)op.buf)[1] = {
-            .oid = { .inode = 1, .stripe = 0x20000 },
-            .version = 2,
-        };
-        ((obj_ver_id*)op.buf)[2] = {
-            .oid = { .inode = 1, .stripe = 0x40000 },
-            .version = 1,
-        };
-        ((obj_ver_id*)op.buf)[3] = {
-            .oid = { .inode = 1, .stripe = 0x60000 },
-            .version = 1,
-        };
-        test.exec_op(&op);
-        assert(op.retval == 0);
-        free(op.buf);
-    }
-
-    blocked_retval = 0;
-    do_write();
-    assert(!blocked_retval);
-
-    free(buf);
-}
-
 // Check null dereference in read_bitmap over DELETE
 static void test_read_bitmap_of_deleted_object()
 {
@@ -1196,6 +1127,158 @@ static void test_used_blocks_replay_delete_over_unstable_big_write()
     free(op.buf);
 }
 
+static void test_delete_over_ec_write()
+{
+    printf("\n-- test_delete_over_ec_write\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.init();
+    printf("blockstore initialized\n");
+
+    blockstore_op_t op;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    memset(op.buf, 0xAA, 128*1024);
+
+    printf("write v1 0+128k (big, stable)\n");
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+    assert(test.used_blocks() > 0);
+
+    printf("write v2 0+4k (big, unstable)\n");
+    op.opcode = BS_OP_WRITE;
+    op.version = 2;
+    op.len = 4*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("delete v3 - should mark all previous versions stable\n");
+    op.opcode = BS_OP_DELETE;
+    op.version = 3;
+    op.offset = 0;
+    op.len = 0;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 2);
+        assert(dirty);
+        assert(IS_STABLE(dirty->state));
+    }
+
+    printf("restarting the blockstore (journal replay)\n");
+    test.destroy_bs();
+    test.init();
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 2);
+        if (!dirty)
+        {
+            dirty = test.find_dirty_entry(op.oid, 3);
+            assert(!dirty);
+        }
+        else
+        {
+            assert(IS_STABLE(dirty->state));
+            dirty = test.find_dirty_entry(op.oid, 3);
+            assert(IS_DELETE(dirty->state));
+        }
+    }
+
+    free(op.buf);
+}
+
+static void test_delete_over_ec_write_unstable()
+{
+    printf("\n-- test_delete_over_ec_write_unstable\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["disable_data_fsync"] = "0";
+    test.config["immediate_commit"] = "none";
+    test.init();
+    printf("blockstore initialized\n");
+
+    blockstore_op_t op;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    memset(op.buf, 0xAA, 128*1024);
+
+    printf("write v1 0+128k (big, stable)\n");
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("write v2 0+4k (big, unstable)\n");
+    op.opcode = BS_OP_WRITE;
+    op.version = 2;
+    op.len = 4*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("delete v3 - should mark all previous versions stable, but only after sync\n");
+    bool del_done = false;
+    op.opcode = BS_OP_DELETE;
+    op.version = 3;
+    op.offset = 0;
+    op.len = 0;
+    op.callback = [&](blockstore_op_t *op)
+    {
+        del_done = true;
+    };
+    test.exec_op(&op);
+    assert(op.retval == 0);
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 3);
+        assert(dirty);
+        assert(!IS_SYNCED(dirty->state));
+        assert(!IS_STABLE(dirty->state));
+    }
+
+    blockstore_op_t sync_op;
+    sync_op.opcode = BS_OP_SYNC;
+    test.exec_op(&sync_op);
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 2);
+        assert(dirty);
+        assert(IS_STABLE(dirty->state));
+        dirty = test.find_dirty_entry(op.oid, 3);
+        assert(dirty);
+        assert(IS_STABLE(dirty->state));
+    }
+
+    printf("restarting the blockstore (journal replay)\n");
+    test.destroy_bs();
+    test.init();
+
+    {
+        auto dirty = test.find_dirty_entry(op.oid, 2);
+        if (!dirty)
+        {
+            dirty = test.find_dirty_entry(op.oid, 3);
+            assert(!dirty);
+        }
+        else
+        {
+            assert(IS_STABLE(dirty->state));
+            dirty = test.find_dirty_entry(op.oid, 3);
+            assert(IS_DELETE(dirty->state));
+        }
+    }
+
+    free(op.buf);
+}
+
 int main(int narg, char *args[])
 {
     test_preserve_corruption();
@@ -1207,9 +1290,10 @@ int main(int narg, char *args[])
     test_read_retry_on_ring_full_1M_csum4k_clean();
     test_read_free_clean_loc_used();
     test_meta_tail_reload();
-    test_write_no_space_eagain();
     test_read_bitmap_of_deleted_object();
     test_used_blocks_replay_stable_over_small_write();
     test_used_blocks_replay_delete_over_unstable_big_write();
+    test_delete_over_ec_write();
+    test_delete_over_ec_write_unstable();
     return 0;
 }