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.

.gitea/workflows/test.yml

@@ -324,6 +324,24 @@ jobs:
           echo ""
         done
 
+  test_no_uring:
+    runs-on: ubuntu-latest
+    needs: build
+    container: ${{env.TEST_IMAGE}}:${{github.sha}}
+    steps:
+    - name: Run test
+      id: test
+      timeout-minutes: 3
+      run: /root/vitastor/tests/test_no_uring.sh
+    - name: Print logs
+      if: always() && steps.test.outcome == 'failure'
+      run: |
+        for i in /root/vitastor/testdata/*.log /root/vitastor/testdata/*.txt; do
+          echo "-------- $i --------"
+          cat $i
+          echo ""
+        done
+
   test_interrupted_rebalance:
     runs-on: ubuntu-latest
     needs: build

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 };
 }

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;

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,19 +94,19 @@ public:
 class journal_flusher_t
 {
     int trim_wanted = 0;
-    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;
 

src/blockstore/v1/stable.cpp

@@ -445,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;
                 }
             }

src/blockstore/v1/sync.cpp

@@ -176,18 +176,15 @@ 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
         {
@@ -195,41 +192,37 @@ void blockstore_impl_t::ack_sync(blockstore_op_t *op)
             assert(unstable_unsynced >= 0);
         }
         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);
-            }
-        }
+        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/client/messenger.cpp

@@ -504,18 +504,21 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd)
     int one = 1;
     setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
     cl->peer_state = PEER_CONNECTED;
-    tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
+    // EPOLLOUT is only required without io_uring, to continue partially sent messages
+    tfd->set_fd_handler(peer_fd, use_sync_send_recv, [this](int peer_fd, int epoll_events)
     {
         handle_peer_epoll(peer_fd, epoll_events);
     });
     // Check OSD number
-    init_tls_client(cl);
+    if (!init_tls_client(cl))
+    {
+        return;
+    }
     check_peer_config(cl);
 }
 
 void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
 {
-    // Mark client as ready (i.e. some data is available)
     auto cl = clients_by_fd.at(peer_fd);
     if (epoll_events & EPOLLRDHUP)
     {
@@ -525,8 +528,9 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
             fprintf(stderr, "[OSD %ju] client %ju disconnected\n", this->osd_num, cl->client_id);
         }
         stop_client(cl->client_id);
+        return;
     }
-    else if (epoll_events & EPOLLIN)
+    if (epoll_events & EPOLLIN)
     {
         // Mark client as ready (i.e. some data is available)
         cl->read_ready++;
@@ -539,6 +543,23 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
                 read_requests();
         }
     }
+    if (epoll_events & EPOLLOUT)
+    {
+        // The socket became writable again - continue the interrupted use_sync_send_recv
+        // read_requests() above may have stopped the client, so look it up again
+        auto cl_it = clients.find(cl->client_id);
+        if (cl_it == clients.end())
+        {
+            return;
+        }
+        cl = cl_it->second;
+        cl->write_ready = true;
+        // send_list is only left non-empty when we're waiting for EPOLLOUT
+        if (cl->send_list.size() && cl->peer_state != PEER_RDMA)
+        {
+            try_send(cl);
+        }
+    }
 }
 
 void osd_messenger_t::on_connect_peer(osd_num_t peer_osd, int errcode, uint64_t client_id)
@@ -770,9 +791,12 @@ void osd_messenger_t::accept_connections(int listen_fd)
         cl->peer_fd = peer_fd;
         cl->peer_state = PEER_CONNECTED;
         cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
-        init_tls_client(cl);
-        // Add FD to epoll
-        tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
+        if (!init_tls_client(cl))
+        {
+            continue;
+        }
+        // Add FD to epoll. EPOLLOUT is only required without io_uring
+        tfd->set_fd_handler(peer_fd, use_sync_send_recv, [this](int peer_fd, int epoll_events)
         {
             handle_peer_epoll(peer_fd, epoll_events);
         });

src/client/messenger.h

@@ -147,6 +147,8 @@ struct osd_client_t
     size_t write_op_pos = 0;
     msghdr write_msg = { 0 };
     int write_state = 0;
+    // Socket is writable. Only tracked with use_sync_send_recv, io_uring doesn't need it
+    bool write_ready = true;
     std::vector<iovec> send_list;
     size_t send_list_size = 0;
     std::deque<osd_op_t*> send_free_ops;
@@ -248,7 +250,7 @@ protected:
 
     void init_tls();
     void destroy_tls();
-    void init_tls_client(osd_client_t *cl);
+    bool init_tls_client(osd_client_t *cl);
     bool do_tls_handshake(osd_client_t *cl, bool from_recv = false);
     bool derive_aes_keys(osd_client_t *cl, bool update_my, bool update_peer);
 
@@ -343,7 +345,8 @@ protected:
     void cancel_op(osd_op_t *op);
 
     bool try_send(osd_client_t *cl);
-    void handle_send(int result, bool prev, bool more, osd_client_t *cl);
+    bool wakeup_send(osd_client_t *cl);
+    bool handle_send(int result, bool prev, bool more, osd_client_t *cl);
     static void post_send_free(osd_op_t *op);
     bool op_write_to(osd_client_t *cl, msgr_op_writer_t & wr);
     void next_write_op(osd_client_t *cl);
@@ -370,7 +373,7 @@ protected:
     void op_decrypt_free(osd_client_t* cl);
 
 #ifdef WITH_RDMA
-    void try_send_rdma(osd_client_t *cl);
+    bool try_send_rdma(osd_client_t *cl);
     bool init_recv_rdma(osd_client_t *cl);
     void handle_rdma_events(msgr_rdma_context_t *rdma_context);
     msgr_rdma_context_t* choose_rdma_context(osd_client_t *cl);

src/client/msgr_encrypt.cpp

@@ -616,7 +616,7 @@ void osd_messenger_t::init_tls()
     }
 }
 
-void osd_messenger_t::init_tls_client(osd_client_t *cl)
+bool osd_messenger_t::init_tls_client(osd_client_t *cl)
 {
     if (gcm_enabled)
     {
@@ -625,13 +625,13 @@ void osd_messenger_t::init_tls_client(osd_client_t *cl)
         cl->hs->init(cl->is_incoming);
         if (cl->hs->out_size())
         {
-            if (cl->write_state == 0)
+            if (!wakeup_send(cl))
             {
-                cl->write_state = CL_WRITE_READY;
-                write_ready_clients.push_back(cl->client_id);
+                return false;
             }
         }
     }
+    return true;
 }
 
 void osd_messenger_t::destroy_tls()

src/client/msgr_rdma.cpp

@@ -571,7 +571,7 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
     cl->rdma_conn->cur_send++;
 }
 
-void osd_messenger_t::try_send_rdma(osd_client_t *cl)
+bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
 {
     auto rc = cl->rdma_conn;
     if (!rc->send_out_size)
@@ -602,7 +602,7 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
         if (cl->io_error)
         {
             stop_client(cl->client_id);
-            return;
+            return false;
         }
         if (copied > 0)
         {
@@ -622,6 +622,7 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
             cl->send_free_ops.push_back(NULL); // end marker
         }
     }
+    return true;
 }
 
 static void try_recv_rdma_wr(osd_client_t *cl, void *buf)
@@ -750,6 +751,7 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context)
             }
         }
     } while (event_count > 0);
+    handle_immediate_ops();
 }
 
 void osd_messenger_t::destroy_rdma_conn(msgr_rdma_connection_t *rdma_conn)

src/client/msgr_rdmacm.cpp

@@ -503,7 +503,6 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev)
     cl->osd_num = peer_osd;
     cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
     cl->rdma_conn = rc;
-    init_tls_client(cl);
     clients[conn->client_id] = cl;
     if (conn->timeout_id >= 0)
     {
@@ -517,6 +516,10 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev)
         fprintf(stderr, "Successfully connected with OSD %ju using RDMA-CM\n", peer_osd);
     }
     // Add initial receive request(s)
+    if (!init_tls_client(cl))
+    {
+        return;
+    }
     init_recv_rdma(cl);
     if (peer_osd)
     {

src/client/msgr_receive.cpp

@@ -458,8 +458,7 @@ void osd_messenger_t::read_requests()
             {
                 result = -errno;
             }
-            // like set_immediate
-            tfd->set_timer_us(0, false, [this, result, cl](int){ handle_read(result, cl); });
+            handle_read(result, cl);
         }
     }
     read_ready_clients.clear();
@@ -490,8 +489,10 @@ void osd_messenger_t::handle_read(int result, osd_client_t *cl)
         }
         stop_client(cl->client_id);
 out_wakeup:
-        if (set_immediate_ops.size())
+        if (ringloop && set_immediate_ops.size())
+        {
             ringloop->wakeup();
+        }
         return;
     }
     bool full_read = false;
@@ -501,7 +502,9 @@ out_wakeup:
         {
             full_read = result >= cl->read_iov.iov_len;
             if (!handle_read_buffer(cl, cl->in_buf, result))
+            {
                 goto out_wakeup;
+            }
         }
         else
         {
@@ -597,13 +600,16 @@ bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size
             bufsize -= done;
             if (cl->hs->out_size())
             {
-                if (cl->write_state == 0)
+                if (!wakeup_send(cl))
                 {
-                    cl->write_state = CL_WRITE_READY;
-                    write_ready_clients.push_back(cl->client_id);
+                    return false;
                 }
             }
-            if (cl->hs->done() && !cl->hs->out_size())
+            if (!cl->hs)
+            {
+                // Deleted by wakeup_send with use_sync_send_recv
+            }
+            else if (cl->hs->done() && !cl->hs->out_size())
             {
                 // Delete hs when done and nothing to send
                 delete cl->hs;

src/client/msgr_send.cpp

@@ -17,6 +17,10 @@
 
 #define GCM_TMP_BUF_SIZE 4096
 
+// How many times to retry sendmsg() in place before falling back to EPOLLOUT.
+// Only used with use_sync_send_recv - io_uring waits for the socket by itself
+#define MSGR_SEND_RETRIES 4
+
 class msgr_op_writer_t
 {
 public:
@@ -467,36 +471,60 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
         return;
     }
 #endif
-    if (!ringloop)
+    try_send(cl);
+}
+
+bool osd_messenger_t::wakeup_send(osd_client_t *cl)
+{
+    if (cl->peer_state == PEER_RDMA)
     {
-        // FIXME: It's worse because it doesn't allow batching
-        while (cl->write_op || cl->write_ops.size())
-        {
-            try_send(cl);
-        }
+        return try_send_rdma(cl);
     }
-    else
+    if (use_sync_send_recv)
     {
-        if (!try_send(cl) && cl->write_state == 0)
-        {
-            cl->write_state = CL_WRITE_READY;
-            write_ready_clients.push_back(cl->client_id);
-        }
-        ringloop->wakeup();
+        // FIXME: Batch send/recv without io_uring too.
+        return try_send(cl);
     }
+    if (cl->write_state == 0)
+    {
+        cl->write_state = CL_WRITE_READY;
+        write_ready_clients.push_back(cl->client_id);
+    }
+    ringloop->wakeup();
+    return true;
 }
 
 bool osd_messenger_t::try_send(osd_client_t *cl)
 {
+    assert(cl->peer_state != PEER_RDMA);
+restart:
     if (cl->peer_state == PEER_STOPPED || cl->peer_fd < 0)
     {
         return true;
     }
-    if (cl->write_msg.msg_iovlen > 0 || !ringloop->space_left() && !use_sync_send_recv)
+    if (cl->write_msg.msg_iovlen > 0)
     {
-        return false;
+        return true;
+    }
+    if (!use_sync_send_recv)
+    {
+        if (!ringloop->space_left())
+        {
+            if (cl->write_state == 0)
+            {
+                cl->write_state = CL_WRITE_READY;
+                write_ready_clients.push_back(cl->client_id);
+            }
+            ringloop->wakeup();
+            return true;
+        }
+    }
+    else if (!cl->write_ready)
+    {
+        // Socket buffer is full, the unsent tail is still in send_list.
+        // Do not copy more data into it, just wait for EPOLLOUT
+        return true;
     }
-    assert(cl->peer_state != PEER_RDMA);
     if (cl->hs)
     {
         // Send handshake message
@@ -523,11 +551,10 @@ copy_ops:
     if (cl->io_error)
     {
         stop_client(cl->client_id);
-        return true;
+        return false;
     }
     if (!cl->send_list.size())
     {
-        cl->write_state = 0;
         return true;
     }
     // get_op_writer_t already guarantees that it's under IOV_MAX
@@ -575,16 +602,29 @@ copy_ops:
     }
     else
     {
-        cl->write_msg.msg_iov = cl->send_list.data();
-        cl->write_msg.msg_iovlen = cl->send_list.size();
-        cl->refs++;
-        int result = sendmsg(cl->peer_fd, &cl->write_msg, MSG_NOSIGNAL);
-        if (result < 0)
+        // Try to first send with a busy loop, then postpone to the EPOLLOUT handler
+        int send_retries = MSGR_SEND_RETRIES;
+        while (cl->send_list.size())
         {
-            result = -errno;
+            cl->write_msg.msg_iov = cl->send_list.data();
+            cl->write_msg.msg_iovlen = cl->send_list.size();
+            cl->refs++;
+            int result = sendmsg(cl->peer_fd, &cl->write_msg, MSG_NOSIGNAL);
+            if (result < 0)
+            {
+                result = -errno;
+            }
+            if (!handle_send(result, false, false, cl))
+            {
+                return false;
+            }
+            if (cl->send_list.size() && --send_retries <= 0)
+            {
+                cl->write_ready = false;
+                return true;
+            }
         }
-        // like set_immediate
-        tfd->set_timer_us(0, false, [this, result, cl](int){ handle_send(result, false, false, cl); });
+        goto restart;
     }
     return true;
 }
@@ -664,21 +704,42 @@ void osd_messenger_t::send_replies()
     {
         uint64_t client_id = write_ready_clients[i];
         auto cl_it = clients.find(client_id);
-        if (cl_it != clients.end() && cl_it->second->peer_state != PEER_RDMA && !try_send(cl_it->second))
+        if (cl_it != clients.end() && cl_it->second->peer_state != PEER_RDMA)
         {
-            write_ready_clients.erase(write_ready_clients.begin(), write_ready_clients.begin() + i);
-            return;
+            if (!ringloop->space_left())
+            {
+                write_ready_clients.erase(write_ready_clients.begin(), write_ready_clients.begin() + i);
+                return;
+            }
+            cl_it->second->write_state = 0;
+            try_send(cl_it->second);
         }
     }
     write_ready_clients.clear();
 }
 
-void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t *cl)
+static void eat_send_list(osd_client_t *cl, size_t n)
+{
+    int i = 0;
+    while (i < cl->send_list.size() && n >= cl->send_list[i].iov_len)
+    {
+        n -= cl->send_list[i].iov_len;
+        i++;
+    }
+    if (i > 0)
+        cl->send_list.erase(cl->send_list.begin(), cl->send_list.begin()+i);
+    if (cl->send_list.size())
+    {
+        cl->send_list[0].iov_base = (uint8_t*)cl->send_list[0].iov_base + n;
+        cl->send_list[0].iov_len -= n;
+    }
+}
+
+bool osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t *cl)
 {
     if (!prev)
     {
         cl->write_msg.msg_iovlen = 0;
-        cl->send_list.clear();
     }
     if (!more)
     {
@@ -690,14 +751,15 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         {
             destroy_client(cl);
         }
-        return;
+        // FIXME: Return error codes instead of true/false (?)
+        return false;
     }
     if (result < 0 && result != -EAGAIN && result != -EINTR)
     {
         // this is a client socket, so don't panic. just disconnect it
         fprintf(stderr, "Client %ju socket write error: %d (%s). Disconnecting client\n", cl->client_id, -result, strerror(-result));
         stop_client(cl->client_id);
-        return;
+        return false;
     }
     if (result >= 0)
     {
@@ -709,14 +771,26 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
                 post_send_free(cl->zc_free_list[i]);
             if (i > 0)
                 cl->zc_free_list.erase(cl->zc_free_list.begin(), cl->zc_free_list.begin()+i+1);
-            return;
+            return true;
         }
-        if (cl->send_list_size > result)
+        if (cl->send_list_size <= result)
+        {
+            cl->send_list.clear();
+            cl->send_list_size = 0;
+        }
+        else if (!use_sync_send_recv)
         {
             fprintf(stderr, "Client %ju socket write error: expected to send "
                 "%zu bytes with MSG_WAITALL but sent %u. Disconnecting client\n", cl->client_id, cl->send_list_size, result);
             stop_client(cl->client_id);
-            return;
+            return false;
+        }
+        else
+        {
+            // Partially sent
+            eat_send_list(cl, result);
+            cl->send_list_size -= result;
+            return true;
         }
         for (auto op: cl->send_free_ops)
         {
@@ -728,9 +802,6 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         if (more)
             cl->zc_free_list.push_back(NULL); // end marker
         cl->send_free_ops.clear();
-        cl->write_state = 0;
-        if (cl->write_op || cl->write_ops.size())
-            cl->write_state = CL_WRITE_READY;
         if ((cl->proto_csum_status & MSGR_CSUM_NEG) && !cl->write_op && !cl->write_ops.size())
         {
             // Checksums negotiated, enable
@@ -750,10 +821,16 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         }
 #endif
     }
-    if (cl->write_state != 0)
+    if ((cl->write_op || cl->write_ops.size() || cl->hs && cl->hs->out_size()) && !use_sync_send_recv)
     {
-        write_ready_clients.push_back(cl->client_id);
+        if (cl->write_state == 0)
+        {
+            cl->write_state = CL_WRITE_READY;
+            write_ready_clients.push_back(cl->client_id);
+        }
+        ringloop->wakeup();
     }
+    return true;
 }
 
 static inline bool op_has_data_for_ssl(osd_op_t *op)

src/test/mock/messenger.cpp

@@ -73,6 +73,11 @@ void osd_messenger_t::accept_connections(int listen_fd)
 {
 }
 
+bool osd_messenger_t::wakeup_send(osd_client_t *cl)
+{
+    return true;
+}
+
 #ifdef WITH_RDMA
 json11::Json osd_messenger_t::connect_rdma(uint64_t client_id, std::string rdma_address, uint64_t client_max_msg)
 {

src/test/test_blockstore.cpp

@@ -362,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");
@@ -976,6 +1091,63 @@ static void test_list_limit()
     free(op.buf);
 }
 
+static void test_intent_write_corruption()
+{
+    printf("\n-- test_intent_write_corruption\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["csum_block_size"] = "16384";
+    test.config["bitmap_granularity"] = "4096";
+    test.config["perfect_csum_update"] = "false";
+    test.init();
+
+    // 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);
+
+
+    // 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);
+
+    // Assert that the read succeeds without checksum corruption errors
+    assert(op.retval == op.len);
+
+    free(op.buf);
+}
+
 // FIXME Add a simple intent_write / big_intent test
 
 int main(int narg, char *args[])
@@ -983,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();

src/test/test_blockstore_v1.cpp

@@ -1127,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();
@@ -1141,5 +1293,7 @@ int main(int narg, char *args[])
     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;
 }

tests/run_tests.sh

@@ -27,6 +27,8 @@ ETCD_SCHEME=https ./test_snapshot.sh
 
 ./test_img_online_resize.sh
 
+./test_no_uring.sh
+
 ./test_interrupted_rebalance.sh
 IMMEDIATE_COMMIT=1 ./test_interrupted_rebalance.sh
 SCHEME=ec ./test_interrupted_rebalance.sh

tests/test_no_uring.sh

@@ -0,0 +1,8 @@
+#!/bin/bash -ex
+
+. `dirname $0`/run_3osds.sh
+
+$VITASTOR_FIO -bs=128k -direct=1 -numjobs=1 -iodepth=4 \
+    -rw=write -pool=1 -inode=1 -size=128M -runtime=10 -no_io_uring=1
+
+format_green OK