Implement actual client operation cancel

Add a test for operation cancel
Throw replies away without allocating buffers(?)
2026-09-25 18:07:09 +03:00 · 2026-09-24 11:55:52 +03:00 · 2026-09-24 11:55:51 +03:00 · 2026-09-24 11:55:00 +03:00
44 changed files with 940 additions and 686 deletions
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -2,7 +2,7 @@ cmake_minimum_required(VERSION 2.8...3.30)

 project(vitastor)

-set(VITASTOR_VERSION "3.2.2")
+set(VITASTOR_VERSION "3.2.1")

 include(CTest)

--- a/csi/Makefile
+++ b/csi/Makefile
@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v3.2.2
+VITASTOR_VERSION ?= v3.2.1

 all: build push

--- a/csi/deploy/004-csi-nodeplugin.yaml
+++ b/csi/deploy/004-csi-nodeplugin.yaml
@@ -49,7 +49,7 @@ spec:
            capabilities:
              add: ["SYS_ADMIN"]
            allowPrivilegeEscalation: true
-          image: vitalif/vitastor-csi:v3.2.2
+          image: vitalif/vitastor-csi:v3.2.1
          args:
            - "--node=$(NODE_ID)"
            - "--endpoint=$(CSI_ENDPOINT)"
--- a/csi/deploy/007-csi-provisioner.yaml
+++ b/csi/deploy/007-csi-provisioner.yaml
@@ -87,7 +87,7 @@ spec:
            - name: socket-dir
              mountPath: /csi
        - name: csi-attacher
-          image: k8s.gcr.io/sig-storage/csi-attacher:v3.2.2
+          image: k8s.gcr.io/sig-storage/csi-attacher:v3.2.1
          args:
            - "--v=5"
            - "--csi-address=$(ADDRESS)"
@@ -121,7 +121,7 @@ spec:
            privileged: true
            capabilities:
              add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v3.2.2
+          image: vitalif/vitastor-csi:v3.2.1
          args:
            - "--node=$(NODE_ID)"
            - "--endpoint=$(CSI_ENDPOINT)"
--- a/csi/src/config.go
+++ b/csi/src/config.go
@@ -5,7 +5,7 @@ package vitastor

 const (
    vitastorCSIDriverName    = "csi.vitastor.io"
-    vitastorCSIDriverVersion = "3.2.2"
+    vitastorCSIDriverVersion = "3.2.1"
 )

 // Config struct fills the parameters of request or user input
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,4 +1,4 @@
-vitastor (3.2.2-1) unstable; urgency=medium
+vitastor (3.2.1-1) unstable; urgency=medium

  * Bugfixes

--- a/docker/Makefile
+++ b/docker/Makefile
@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v3.2.2
+VITASTOR_VERSION ?= v3.2.1

 all: build push

--- a/docker/etc/vitastor/docker.conf
+++ b/docker/etc/vitastor/docker.conf
@@ -4,7 +4,7 @@
 #

 # Desired Vitastor version
-VITASTOR_VERSION=v3.2.2
+VITASTOR_VERSION=v3.2.1

 # Additional arguments for all containers
 # For example, you may want to specify a custom logging driver here
--- a/docs/installation/docker.en.md
+++ b/docs/installation/docker.en.md
@@ -26,9 +26,9 @@ at Vitastor Kubernetes operator: https://github.com/Antilles7227/vitastor-operat
 The instruction is very simple.

 1. Download a Docker image of the desired version: \
-   `docker pull vitalif/vitastor:v3.2.2`
+   `docker pull vitalif/vitastor:v3.2.1`
 2. Install scripts to the host system: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.2 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.1 install.sh`
 3. Reload udev rules: \
   `udevadm control --reload-rules`
 4. Enable the vitastor-host service: \
--- a/docs/installation/docker.ru.md
+++ b/docs/installation/docker.ru.md
@@ -25,9 +25,9 @@ Vitastor можно установить в Docker/Podman. При этом etcd,
 Инструкция по установке максимально простая.

 1. Скачайте Docker-образ желаемой версии: \
-   `docker pull vitalif/vitastor:v3.2.2`
+   `docker pull vitalif/vitastor:v3.2.1`
 2. Установите скрипты в хост-систему командой: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.2 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.1 install.sh`
 3. Перезагрузите правила udev: \
   `udevadm control --reload-rules`
 4. Включите сервис vitastor-host: \
--- a/mon/package.json
+++ b/mon/package.json
@@ -1,6 +1,6 @@
 {
  "name": "vitastor-mon",
-  "version": "3.2.2",
+  "version": "3.2.1",
  "description": "Vitastor SDS monitor service",
  "main": "mon-main.js",
  "scripts": {
--- a/node-binding/package.json
+++ b/node-binding/package.json
@@ -1,6 +1,6 @@
 {
  "name": "vitastor",
-  "version": "3.2.2",
+  "version": "3.2.1",
  "description": "Low-level native bindings to Vitastor client library",
  "main": "index.js",
  "keywords": [
--- a/patches/cinder-vitastor.py
+++ b/patches/cinder-vitastor.py
@@ -50,7 +50,7 @@ from cinder.volume import configuration
 from cinder.volume import driver
 from cinder.volume import volume_utils

-VITASTOR_VERSION = '3.2.2'
+VITASTOR_VERSION = '3.2.1'

 LOG = logging.getLogger(__name__)

--- a/rpm/vitastor-el10.spec
+++ b/rpm/vitastor-el10.spec
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.2.2
+Version:        3.2.1
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage

 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-3.2.2.el10.tar.gz
+Source0:        vitastor-3.2.1.el10.tar.gz

 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-c++
--- a/rpm/vitastor-el7.spec
+++ b/rpm/vitastor-el7.spec
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.2.2
+Version:        3.2.1
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage

 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-3.2.2.el7.tar.gz
+Source0:        vitastor-3.2.1.el7.tar.gz

 BuildRequires:  gperftools-devel
 BuildRequires:  devtoolset-9-gcc-c++
--- a/rpm/vitastor-el8.spec
+++ b/rpm/vitastor-el8.spec
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.2.2
+Version:        3.2.1
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage

 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-3.2.2.el8.tar.gz
+Source0:        vitastor-3.2.1.el8.tar.gz

 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-toolset-9-gcc-c++
--- a/rpm/vitastor-el9.spec
+++ b/rpm/vitastor-el9.spec
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.2.2
+Version:        3.2.1
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage

 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-3.2.2.el9.tar.gz
+Source0:        vitastor-3.2.1.el9.tar.gz

 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-c++
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -21,7 +21,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 endif()
 set(ENABLE_COVERAGE false CACHE BOOL "Enable code coverage")

-add_definitions(-DVITASTOR_VERSION="3.2.2")
+add_definitions(-DVITASTOR_VERSION="3.2.1")
 add_definitions(-D_GNU_SOURCE -D_LARGEFILE64_SOURCE -D_FILE_OFFSET_BITS=64 -Wall -Wno-sign-compare -Wno-comment -Wno-parentheses -Wno-pointer-arith -fdiagnostics-color=always -fno-omit-frame-pointer -fvisibility=hidden -I ${CMAKE_SOURCE_DIR}/src)
 add_link_options(-fno-omit-frame-pointer)
 if (${WITH_ASAN})
--- a/src/blockstore/blockstore_flush.cpp
+++ b/src/blockstore/blockstore_flush.cpp
@@ -213,25 +213,18 @@ resume_0:
    res = bs->heap->get_next_compact(cur_oid);
    if ((bs->intent_write_counter >= bs->journal_trim_interval || flusher->force_fsync > 0) && co_id == 0)
    {
+        // Advance fsynced_lsn every <journal_trim_interval> intent writes
+        bs->intent_write_counter = 0;
 resume_17:
 resume_18:
 resume_19:
        if (!fsync_buffer(17))
-        {
            return false;
-        }
-        if (bs->intent_write_counter >= bs->journal_trim_interval)
-        {
-            // Advance fsynced_lsn every <journal_trim_interval> intent writes
-            bs->intent_write_counter = 0;
 resume_29:
 resume_30:
 resume_31:
-            if (!trim_lsn(29))
-            {
-                return false;
-            }
-        }
+        if (!trim_lsn(29))
+            return false;
    }
    if (res == ENOENT && flusher->force_start > 0 && co_id == 0 &&
        (!bs->dsk.disable_journal_fsync || !bs->dsk.disable_meta_fsync || !bs->dsk.disable_data_fsync))
--- a/src/blockstore/blockstore_heap.cpp
+++ b/src/blockstore/blockstore_heap.cpp
@@ -197,10 +197,7 @@ uint32_t *heap_entry_t::get_checksum(blockstore_heap_t *heap)
 {
    if (type() == BS_HEAP_SMALL_WRITE || type() == BS_HEAP_INTENT_WRITE)
    {
-        // a zero-length small_write still has a checksum slot reserved by
-        // get_small_entry_size when block checksums are off, and its entry
-        // checksum covers it - so it has to be filled with a defined value
-        if (heap->dsk->csum_block_size)
+        if (heap->dsk->csum_block_size || small().len == 0)
            return NULL;
        return (uint32_t*)((uint8_t*)this + sizeof(heap_small_write_t) + heap->dsk->clean_entry_bitmap_size);
    }
@@ -1266,12 +1263,6 @@ bool blockstore_heap_t::calc_checksums(heap_entry_t *wr, uint8_t *data, bool set
            len = wr->big_intent().len;
        else
            assert(0);
-        if (!len)
-        {
-            if (set)
-                *wr_csum = 0;
-            return true;
-        }
        uint32_t real_csum = 0;
        if (dsk->data_csum_type == BLOCKSTORE_CSUM_XXH3_32)
            real_csum = (uint32_t)XXH3_64bits(data, len);
@@ -2445,11 +2436,9 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
            res.do_delete = false;
            res.compact_lsn = 0;
            res.compact_version = 0;
-            if ((!under_pressure || stable) && (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_BIG_INTENT || wr->type() == BS_HEAP_DELETE))
+            if (!under_pressure && (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_DELETE))
            {
                // We may postpone compaction if we have an unstable overwrite when not under pressure
-                // We MUST postpone if the overwrite is STABLE (even if under pressure), 
-                // because add_compact will reject STABLE overwrites with EBUSY.
                return res;
            }
            continue;
--- a/src/client/cluster_client.cpp
+++ b/src/client/cluster_client.cpp
@@ -14,11 +14,6 @@

 cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json config, std::unique_ptr<etcd_state_client_t> st_cli_ptr)
 {
-    auto on_fatal_error_hook = [](const std::string& error)
-    {
-        throw std::runtime_error(error);
-    };
-
    wb = new writeback_cache_t();

    cli_config = config.object_items();
@@ -28,13 +23,12 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
    this->ringloop = ringloop;
    this->tfd = tfd;

-    msgr = new osd_messenger_t();
-    msgr->osd_num = 0;
-    msgr->tfd = tfd;
-    msgr->ringloop = ringloop;
-    msgr->repeer_pgs = [this](osd_num_t peer_osd)
+    msgr.osd_num = 0;
+    msgr.tfd = tfd;
+    msgr.ringloop = ringloop;
+    msgr.repeer_pgs = [this](osd_num_t peer_osd)
    {
-        if (msgr->osd_peers.find(peer_osd) != msgr->osd_peers.end())
+        if (msgr.osd_peers.find(peer_osd) != msgr.osd_peers.end())
        {
            // peer_osd just connected
            continue_ops();
@@ -51,16 +45,14 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
            }
        }
    };
-    msgr->exec_op = [this](osd_op_t *op)
+    msgr.exec_op = [this](osd_op_t *op)
    {
        // Garbage in
        fprintf(stderr, "Can't handle incoming operation from client %ju\n", op->client_id);
-        msgr->stop_client(op->client_id);
+        msgr.stop_client(op->client_id);
        delete op;
    };
-    auto exit_hook = msgr->on_fatal_error_hook;
-    msgr->on_fatal_error_hook = on_fatal_error_hook;
-    msgr->parse_config(config, true);
+    msgr.parse_config(config, true);

    st_cli = std::move(st_cli_ptr);
    st_cli->on_load_config_hook = [this](json11::Json::object & cfg) { on_load_config_hook(cfg); };
@@ -72,7 +64,6 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
    st_cli->on_load_pgs_hook = [this](bool success) { on_load_pgs_hook(success); };
    st_cli->on_reload_hook = [this]() { this->st_cli->load_global_config(); };
    st_cli->on_inode_change_hook = [this](uint64_t inode, bool removed) { on_change_inode_hook(inode, removed); };
-    st_cli->on_fatal_error_hook = on_fatal_error_hook;

    st_cli->parse_config(config);
    st_cli->infinite_start = false;
@@ -81,30 +72,11 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
        st_cli->infinite_start = config["client_infinite_start"].bool_value();
    }
    st_cli->load_global_config();
-
-    if (ringloop)
-    {
-        ringloop->loop();
-    }
-
-    msgr->on_fatal_error_hook = exit_hook;
-    st_cli->on_fatal_error_hook = exit_hook;
-}
-
-void cluster_client_t::set_on_fatal_error_hook(std::function<void(const std::string&)> cb)
-{
-    st_cli->on_fatal_error_hook = cb;
-    msgr->on_fatal_error_hook = cb;
-}
-
-void cluster_client_t::fatal_error(const std::string & error)
-{
-    (std::function<void(const std::string&)>(msgr->on_fatal_error_hook))(error);
 }

 cluster_client_t::~cluster_client_t()
 {
-    aborted = true;
+    vault_destroy();
    if (retry_timeout_id >= 0)
    {
        tfd->clear_timer(retry_timeout_id);
@@ -117,21 +89,13 @@ cluster_client_t::~cluster_client_t()
        list_retry_timeout_id = -1;
        list_retry_time = {};
    }
-    msgr->repeer_pgs = [](osd_num_t){};
+    msgr.repeer_pgs = [](osd_num_t){};
    if (ringloop)
    {
        ringloop->unregister_consumer(&consumer);
    }
    delete wb;
    wb = NULL;
-    delete msgr;
-    msgr = NULL;
-    while (op_queue_head)
-    {
-        op_queue_head->retval = -ECANCELED;
-        erase_op(op_queue_head);
-    }
-    vault_destroy();
 }

 cluster_op_t::~cluster_op_t()
@@ -186,15 +150,13 @@ std::vector<osd_num_t> cluster_op_t::get_left_on_dead()

 void cluster_client_t::continue_raw_ops(osd_num_t peer_osd)
 {
-    if (aborted)
-        return;
    auto it = raw_ops.find(peer_osd);
    while (it != raw_ops.end() && it->first == peer_osd)
    {
        auto op = it->second;
        op->op_type = OSD_OP_OUT;
-        op->client_id = msgr->osd_peers.at(peer_osd)->client_id;
-        msgr->outbox_push(op);
+        op->client_id = msgr.osd_peers.at(peer_osd)->client_id;
+        msgr.outbox_push(op);
        raw_ops.erase(it++);
    }
 }
@@ -203,14 +165,14 @@ void cluster_client_t::init_msgr()
 {
    if (msgr_initialized)
        return;
-    msgr->init();
+    msgr.init();
    msgr_initialized = true;
    if (ringloop)
    {
        consumer.loop = [this]()
        {
-            msgr->read_requests();
-            msgr->send_replies();
+            msgr.read_requests();
+            msgr.send_replies();
            this->ringloop->submit();
        };
        ringloop->register_consumer(&consumer);
@@ -520,7 +482,7 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_co
    }
    // vault
    vault_parse_config();
-    msgr->parse_config(config, false);
+    msgr.parse_config(config, false);
    st_cli->parse_config(config);
    st_cli->load_pgs();
 }
@@ -681,9 +643,9 @@ bool cluster_client_t::get_immediate_commit(uint64_t inode)
 void cluster_client_t::on_change_osd_state_hook(uint64_t peer_osd)
 {
    osd_tree_metrics.erase(peer_osd);
-    if (msgr->wanted_peers.find(peer_osd) != msgr->wanted_peers.end())
+    if (msgr.wanted_peers.find(peer_osd) != msgr.wanted_peers.end())
    {
-        msgr->connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
+        msgr.connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
        continue_lists();
    }
 }
@@ -799,6 +761,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
    op->cur_inode = op->inode;
    op->retval = 0;
    op->state = 0;
+    op->canceled = false;
    op->retry_after = 0;
    op->inflight_count = 0;
    op->done_count = 0;
@@ -898,15 +861,115 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
    }
 }

-void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
+// Try to cancel a previously started operation
+//
+// A canceled operation is completed with -ECANCELED, either immediately (from inside
+// this function) or later, when the messenger finishes canceling its sub-operations.
+//
+// Some operations can't be canceled at all - in that case they just complete normally.
+// The result of a canceled write is undefined - some parts of it may be written and
+// some may be not, so the write-back copy of its data is dropped, too.
+//
+// Returns true if the operation is already completed, i.e. if its callback is already called
+bool cluster_client_t::cancel(cluster_op_t *op)
 {
-    if (aborted)
+    if (op->canceled || (op->flags & OP_FLUSH_BUFFER))
    {
+        // Already canceled, or an internal write-back flush which is never canceled
+        return false;
+    }
+    // Operations which aren't started yet are just removed from their queues
+    for (auto queue: std::initializer_list<std::vector<cluster_op_t*>*>{ &offline_ops, &key_wait_ops, &wb->writeback_overflow })
+    {
+        for (auto op_it = queue->begin(); op_it != queue->end(); op_it++)
+        {
+            if (*op_it == op)
+            {
+                queue->erase(op_it);
+                op->canceled = true;
+                op->retval = -ECANCELED;
+                auto cb = std::move(op->callback);
+                cb(op);
+                return true;
+            }
+        }
+    }
+    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version)
+    {
+        // CAS operations aren't in the operation queue and have their own callback
+        if (!op->parts.size() || !(op->parts[0].flags & PART_SENT) ||
+            (op->parts[0].flags & (PART_DONE|PART_ERROR)))
+        {
+            // Not sent yet (waiting for a preliminary sync) => can't cancel it
+            return false;
+        }
+        op->canceled = true;
+        if (!msgr.cancel_out(&op->parts[0].op))
+        {
+            // The callback will be called with -ECANCELED by the messenger
+            return false;
+        }
        op->retval = -ECANCELED;
        auto cb = std::move(op->callback);
        cb(op);
-        return;
+        return true;
    }
+    op->canceled = true;
+    op->retry_after = 0;
+    // ERR_PRIO(-ECANCELED) is the highest, so replies to the remaining parts
+    // won't override it and won't schedule a retry in handle_op_part()
+    op->retval = -ECANCELED;
+    // msgr.cancel_out() never calls callbacks itself, so the operation
+    // can not be completed in the middle of this loop
+    for (auto & part: op->parts)
+    {
+        if ((part.flags & PART_SENT) && !(part.flags & (PART_DONE|PART_ERROR)) &&
+            msgr.cancel_out(&part.op))
+        {
+            // Canceled synchronously => the callback won't be called at all
+            part.flags |= PART_ERROR;
+            op->inflight_count--;
+        }
+    }
+    if (op->inflight_count > 0)
+    {
+        // The rest of the parts will be canceled asynchronously
+        return false;
+    }
+    finish_canceled(op);
+    return true;
+}
+
+void cluster_client_t::finish_canceled(cluster_op_t *op)
+{
+    assert(op->canceled && !op->inflight_count);
+    op->retval = -ECANCELED;
+    op->retry_after = 0;
+    if (op->opcode == OSD_OP_SYNC)
+    {
+        if (op->parts.size())
+        {
+            // Syncs were already sent - roll the write-back cache state back and
+            // mark all unsynced OSDs as dirty again so the next sync retries them
+            wb->fsync_error();
+            for (auto & part: op->parts)
+            {
+                if (!(part.flags & PART_DONE))
+                    dirty_osds.insert(part.osd_num);
+            }
+        }
+    }
+    else if (op->flush_id)
+    {
+        // The result of a canceled write is undefined, so drop the write-repeat copy
+        wb->delete_flush(op->inode, op->offset, op->len, op->flush_id);
+        op->flush_id = 0;
+    }
+    erase_op(op);
+}
+
+void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
+{
    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1)
    {
        // Atomic writes to multiple stripes are unsupported
@@ -923,14 +986,7 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
        sync_op->flags = 0;
        sync_op->callback = [this, op](cluster_op_t* sync_op)
        {
-            if (sync_op->retval != 0)
-            {
-                op->retval = sync_op->retval;
-                auto cb = std::move(op->callback);
-                cb(op);
-            }
-            else
-                execute_cas(op, true);
+            execute_cas(op, true);
            delete sync_op;
        };
        execute_internal(sync_op);
@@ -946,25 +1002,18 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
    op->needs_reslice = false;
    int res = try_send(op, 0, [this, op](osd_op_t *part)
    {
-        if (aborted)
-        {
-            op->retval = -ECANCELED;
-            auto cb = std::move(op->callback);
-            cb(op);
-            return;
-        }
        int expected = part->req.hdr.opcode == OSD_OP_DELETE ? 0 : part->req.rw.len;
        op->retval = part->reply.hdr.retval;
        if (op->retval != expected && op->retval >= 0)
            op->retval = -EIO;
        op->retval = op->retval == -EPIPE ? -EINTR : op->retval;
-        auto peer_it = msgr->osd_peers.find(op->parts[0].osd_num);
+        auto peer_it = msgr.osd_peers.find(op->parts[0].osd_num);
        if (op->retval != expected || (op->flags & OP_IMMEDIATE_COMMIT))
        {
            auto cb = std::move(op->callback);
            cb(op);
        }
-        else if (peer_it == msgr->osd_peers.end())
+        else if (peer_it == msgr.osd_peers.end())
        {
            // Care must be taken to make sure that the client doesn't reconnect to the OSD
            // before executing the previously completed operation callback (!)
@@ -996,7 +1045,7 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
                    cb(op);
                },
            };
-            msgr->outbox_push(part);
+            msgr.outbox_push(part);
        }
    });
    if (res == TRY_SEND_CONNECTING || res == TRY_SEND_OFFLINE)
@@ -1127,25 +1176,33 @@ bool cluster_client_t::check_rw(cluster_op_t *op)

 void cluster_client_t::execute_raw(osd_num_t osd_num, osd_op_t *op)
 {
-    auto peer_it = msgr->osd_peers.find(osd_num);
-    if (peer_it != msgr->osd_peers.end())
+    auto peer_it = msgr.osd_peers.find(osd_num);
+    if (peer_it != msgr.osd_peers.end())
    {
        op->op_type = OSD_OP_OUT;
        op->client_id = peer_it->second->client_id;
-        msgr->outbox_push(op);
+        msgr.outbox_push(op);
    }
    else
    {
-        if (msgr->wanted_peers.find(osd_num) == msgr->wanted_peers.end())
-            msgr->connect_peer(osd_num, st_cli->peer_states[osd_num]);
+        if (msgr.wanted_peers.find(osd_num) == msgr.wanted_peers.end())
+            msgr.connect_peer(osd_num, st_cli->peer_states[osd_num]);
        raw_ops.emplace(osd_num, op);
    }
 }

 int cluster_client_t::continue_rw(cluster_op_t *op)
 {
-    if (aborted)
-        return 0;
+    if (op->canceled)
+    {
+        if (op->inflight_count > 0)
+        {
+            // Wait for the remaining parts to be canceled
+            return 0;
+        }
+        finish_canceled(op);
+        return 1;
+    }
    if (op->state == 0)
        goto resume_0;
    else if (op->state == 1)
@@ -1570,8 +1627,8 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
                primary_osd = nearest_osd;
        }
        part->osd_num = primary_osd;
-        auto peer_it = msgr->osd_peers.find(primary_osd);
-        if (peer_it != msgr->osd_peers.end())
+        auto peer_it = msgr.osd_peers.find(primary_osd);
+        if (peer_it != msgr.osd_peers.end())
        {
            osd_client_t *cl = peer_it->second;
            part->flags |= PART_SENT|PART_VALID;
@@ -1615,12 +1672,12 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
                },
            };
            part->op.iov = part->iov;
-            msgr->outbox_push(&part->op);
+            msgr.outbox_push(&part->op);
            return TRY_SEND_OK;
        }
-        else if (msgr->wanted_peers.find(primary_osd) == msgr->wanted_peers.end())
+        else if (msgr.wanted_peers.find(primary_osd) == msgr.wanted_peers.end())
        {
-            msgr->connect_peer(primary_osd, st_cli->peer_states[primary_osd]);
+            msgr.connect_peer(primary_osd, st_cli->peer_states[primary_osd]);
            return TRY_SEND_CONNECTING;
        }
    }
@@ -1629,8 +1686,16 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o

 int cluster_client_t::continue_sync(cluster_op_t *op)
 {
-    if (aborted)
-        return 0;
+    if (op->canceled)
+    {
+        if (op->inflight_count > 0)
+        {
+            // Wait for the remaining parts to be canceled
+            return 0;
+        }
+        finish_canceled(op);
+        return 1;
+    }
    if (op->state == 1)
        goto resume_1;
    if (!dirty_osds.size())
@@ -1644,8 +1709,8 @@ int cluster_client_t::continue_sync(cluster_op_t *op)
    for (auto do_it = dirty_osds.begin(); do_it != dirty_osds.end(); )
    {
        osd_num_t sync_osd = *do_it;
-        auto peer_it = msgr->osd_peers.find(sync_osd);
-        if (peer_it == msgr->osd_peers.end())
+        auto peer_it = msgr.osd_peers.find(sync_osd);
+        if (peer_it == msgr.osd_peers.end())
            dirty_osds.erase(do_it++);
        else
            do_it++;
@@ -1698,7 +1763,7 @@ resume_1:

 void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part)
 {
-    osd_client_t *cl = msgr->osd_peers.at(part->osd_num);
+    osd_client_t *cl = msgr.osd_peers.at(part->osd_num);
    part->flags |= PART_SENT;
    op->inflight_count++;
    part->op = (osd_op_t){
@@ -1715,7 +1780,7 @@ void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part)
            handle_op_part(part);
        },
    };
-    msgr->outbox_push(&part->op);
+    msgr.outbox_push(&part->op);
 }

 static inline void mem_or(void *res, const void *r2, unsigned int len)
@@ -1739,7 +1804,8 @@ 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 && op->retval != -EAGAIN && op->retval != -ECANCELED)
        {
            stop_client_id = part->op.client_id;
            if (op->retval != -EPIPE || log_level > 0)
@@ -1770,7 +1836,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
        reset_retry_timer(op->retry_after);
        if (stop_client_id)
        {
-            msgr->stop_client(stop_client_id);
+            msgr.stop_client(stop_client_id);
        }
        op->inflight_count--;
        if (op->inflight_count == 0 && !op->retry_after)
--- a/src/client/cluster_client.h
+++ b/src/client/cluster_client.h
@@ -65,6 +65,7 @@ protected:
    uint64_t cur_inode; // for snapshot reads
    bool needs_reslice: 1;
    bool deoptimise_snapshot: 1;
+    bool canceled = false;
    int retry_after = 0;
    int inflight_count = 0, done_count = 0;
    timespec wait_up_until = {};
@@ -176,12 +177,11 @@ class __attribute__((visibility("default"))) cluster_client_t
    std::multimap<osd_num_t, osd_op_t*> raw_ops;
    int continuing_ops = 0;
    bool msgr_initialized = false;
-    bool aborted = false;

 public:
    std::unique_ptr<etcd_state_client_t> st_cli;
-    osd_messenger_t* msgr = NULL;

+    osd_messenger_t msgr;
    void init_msgr();

    json11::Json::object cli_config, file_config, etcd_global_config;
@@ -191,11 +191,11 @@ public:
    cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json config, std::unique_ptr<etcd_state_client_t> st_cli);
    ~cluster_client_t();
    void execute(cluster_op_t *op);
+    bool cancel(cluster_op_t *op);
    void execute_raw(osd_num_t osd_num, osd_op_t *op);
    bool is_ready();
    void on_ready(std::function<void(void)> fn);
    bool flush();
-    void set_on_fatal_error_hook(std::function<void(const std::string&)>);

    bool get_immediate_commit(uint64_t inode);

@@ -203,7 +203,6 @@ public:
        int status, int pgs_left, pg_num_t pg_num, std::set<object_id>&& objects)> pg_callback);

 protected:
-    void fatal_error(const std::string& error);
    void continue_ops(int time_passed = 0);

    std::shared_ptr<inode_cache_t> inode_cache_get(inode_t ino);
@@ -235,6 +234,7 @@ protected:
    int continue_sync(cluster_op_t *op);
    void send_sync(cluster_op_t *op, cluster_op_part_t *part);
    void handle_op_part(cluster_op_part_t *part);
+    void finish_canceled(cluster_op_t *op);
    void copy_part_bitmap(cluster_op_t *op, cluster_op_part_t *part);
    void erase_op(cluster_op_t *op);
    void calc_wait(cluster_op_t *op);
--- a/src/client/cluster_client_list.cpp
+++ b/src/client/cluster_client_list.cpp
@@ -301,14 +301,14 @@ int cluster_client_t::start_pg_listing(inode_list_pg_t *pg)
    bool conn = true;
    for (osd_num_t peer_osd: all_peers)
    {
-        if (msgr->osd_peers.find(peer_osd) == msgr->osd_peers.end())
+        if (msgr.osd_peers.find(peer_osd) == msgr.osd_peers.end())
        {
            // Initiate connection
            if (st_cli->peer_states[peer_osd].is_null())
            {
                return LIST_PG_WAIT_ACTIVE;
            }
-            msgr->connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
+            msgr.connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
            conn = false;
        }
    }
@@ -346,7 +346,7 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
    osd_op_t *op = new osd_op_t();
    op->op_type = OSD_OP_OUT;
    // Already checked that it exists above, but anyway
-    op->client_id = msgr->osd_peers.at(cur_list->osd_num)->client_id;
+    op->client_id = msgr.osd_peers.at(cur_list->osd_num)->client_id;
    op->req = (osd_any_op_t){
        .sec_list = {
            .header = {
@@ -425,7 +425,7 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
        finish_list_pg(cur_list->pg, true);
        continue_listing(cur_list->pg->lst);
    };
-    msgr->outbox_push(op);
+    msgr.outbox_push(op);
 }

 void cluster_client_t::finish_list_pg(inode_list_pg_t *pg, bool retry_epipe)
--- a/src/client/cluster_client_real.cpp
+++ b/src/client/cluster_client_real.cpp
@@ -98,8 +98,8 @@ void cluster_client_t::vault_load_keys()
        vault_http_ctx = http_context_init(tfd, vault_client_cert, vault_client_key, vault_ca, true, error);
        if (!vault_http_ctx)
        {
-            fatal_error("Failed to initialize HTTP context for Vault: "+error);
-            return;
+            fprintf(stderr, "Failed to initialize HTTP context for Vault: %s\n", error.c_str());
+            exit(1);
        }
        vault_http_cli = http_init(vault_http_ctx);
    }
--- a/src/client/etcd_state_client.cpp
+++ b/src/client/etcd_state_client.cpp
@@ -10,20 +10,6 @@
 #include "str_util.h"
 #include "json_util.h"

-etcd_state_client_t::etcd_state_client_t()
-{
-    this->on_fatal_error_hook = [](const std::string & error)
-    {
-        fprintf(stderr, "%s\n", error.c_str());
-        exit(1);
-    };
-}
-
-void etcd_state_client_t::fatal_error(const std::string & error)
-{
-    (std::function<void(const std::string&)>(on_fatal_error_hook))(error);
-}
-
 etcd_state_client_t::~etcd_state_client_t()
 {
    for (auto watch: watches)
--- a/src/client/etcd_state_client.h
+++ b/src/client/etcd_state_client.h
@@ -144,7 +144,6 @@ protected:
    std::set<osd_num_t> seen_peers;
    bool new_pg_config = false;

-    void fatal_error(const std::string & error);
    void add_etcd_url(std::string);
    void reset_pg_exists();
    void clean_nonexistent_pgs();
@@ -193,7 +192,6 @@ public:
    std::function<void()> on_reload_hook;
    std::function<void(inode_t, bool)> on_inode_change_hook;
    std::function<void(http_co_t *)> on_start_watcher_hook;
-    std::function<void(const std::string&)> on_fatal_error_hook;

    json11::Json::object serialize_inode_cfg(inode_config_t *cfg);
    inode_config_t deserialize_inode_cfg(uint64_t inode_num, json11::Json value, uint64_t mod_revision);
@@ -217,7 +215,6 @@ public:
    inode_watch_t* watch_inode(std::string name);
    void close_watch(inode_watch_t* watch);
    int address_count();
-    etcd_state_client_t();
    virtual ~etcd_state_client_t();

    static uint32_t parse_immediate_commit(const std::string & immediate_commit_str, uint32_t default_value);
--- a/src/client/etcd_state_client_http.cpp
+++ b/src/client/etcd_state_client_http.cpp
@@ -49,11 +49,7 @@ void etcd_state_client_http_t::etcd_add_watch(json11::Json watch)

 std::string etcd_state_client_http_t::get_username()
 {
-    if (!get_http_ctx())
-    {
-        return "";
-    }
-    return http_context_get_ssl_cn(http_ctx);
+    return http_context_get_ssl_cn(get_http_ctx());
 }

 http_context_t *etcd_state_client_http_t::get_http_ctx()
@@ -64,7 +60,8 @@ http_context_t *etcd_state_client_http_t::get_http_ctx()
        http_ctx = http_context_init(tfd, etcd_client_cert, etcd_client_key, etcd_ca, true, error);
        if (!http_ctx)
        {
-            fatal_error("Failed to initialize HTTP context: "+error);
+            fprintf(stderr, "Failed to initialize HTTP context: %s\n", error.c_str());
+            exit(1);
        }
    }
    return http_ctx;
@@ -73,11 +70,7 @@ http_context_t *etcd_state_client_http_t::get_http_ctx()
 void etcd_state_client_http_t::etcd_call_oneshot(const std::string & etcd_url, const std::string & api, json11::Json payload,
    int timeout, std::function<void(std::string, json11::Json)> callback)
 {
-    if (!get_http_ctx())
-    {
-        return;
-    }
-    auto http_cli = http_init(http_ctx);
+    auto http_cli = http_init(get_http_ctx());
    http_json_post(http_cli, etcd_url+api, payload, "", { .timeout = timeout }, [http_cli, callback](http_message_t *response)
    {
        std::string err;
@@ -147,11 +140,7 @@ void etcd_state_client_http_t::etcd_call_selected(const std::string & api, json1
            callback(err, data);
    };
    if (!keepalive_client)
-    {
-        if (!get_http_ctx())
-            return;
-        keepalive_client = http_init(http_ctx);
-    }
+        keepalive_client = http_init(get_http_ctx());
    http_request(keepalive_client, url.addr, req, { .timeout = timeout, .keepalive = true, .ssl = url.ssl }, cb);
 }

@@ -170,12 +159,8 @@ void etcd_state_client_http_t::pick_next_etcd(std::function<void()> cb)
 {
    if (!etcd_addresses.size() && !etcd_local.size())
    {
-        fatal_error("etcd_address is missing in Vitastor configuration");
-        return;
-    }
-    if (!get_http_ctx())
-    {
-        return;
+        fprintf(stderr, "etcd_address is missing in Vitastor configuration\n");
+        exit(1);
    }
    if (selected_etcd_url.addr != "")
    {
@@ -211,7 +196,7 @@ void etcd_state_client_http_t::pick_next_etcd(std::function<void()> cb)
    for (auto & url: etcd_name_urls)
    {
        resolve_count++;
-        http_resolve(http_ctx, url.ssl, url.addr, [this, url](const std::string & error, const std::vector<std::string>& addresses)
+        http_resolve(get_http_ctx(), url.ssl, url.addr, [this, url](const std::string & error, const std::vector<std::string>& addresses)
        {
            if (error != "")
                fprintf(stderr, "Error resolving %s: %s\n", url.addr.c_str(), error.c_str());
@@ -243,8 +228,8 @@ void etcd_state_client_http_t::pick_next_etcd_on_resolve()
 {
    if (!etcd_urls_to_try.size())
    {
-        fatal_error("None of etcd_address could be resolved");
-        return;
+        fprintf(stderr, "None of etcd_address could be resolved\n");
+        exit(1);
    }
    if (!rand_initialized)
    {
@@ -289,11 +274,7 @@ void etcd_state_client_http_t::start_etcd_watcher_selected()
            etcd_watch_revision_config, etcd_watch_revision_osd, etcd_watch_revision_pg);
    }
    if (!etcd_watch_ws)
-    {
-        if (!get_http_ctx())
-            return;
-        etcd_watch_ws = http_init(http_ctx);
-    }
+        etcd_watch_ws = http_init(get_http_ctx());
    else
        http_close(etcd_watch_ws);
    open_websocket(etcd_watch_ws, url.addr, url.hostname, url.path+"/watch", { .timeout = etcd_slow_timeout, .ssl = url.ssl },
@@ -343,17 +324,20 @@ void etcd_state_client_http_t::start_etcd_watcher_selected()
                                etcd_watch_revision_config = etcd_watch_revision_osd = etcd_watch_revision_pg = 0;
                                on_reload_hook();
                            }
+                            return;
                        }
                        else
                        {
-                            fatal_error("Revisions before "+data["result"]["compact_revision"].as_string()+" were compacted by etcd, exiting");
+                            fprintf(stderr, "Revisions before %ju were compacted by etcd, exiting\n",
+                                data["result"]["compact_revision"].uint64_value());
+                            exit(1);
                        }
                    }
                    else
                    {
-                        fatal_error("Watch canceled by etcd, reason: "+data["result"]["cancel_reason"].string_value()+", exiting");
+                        fprintf(stderr, "Watch canceled by etcd, reason: %s, exiting\n", data["result"]["cancel_reason"].string_value().c_str());
+                        exit(1);
                    }
-                    return;
                }
                // Save revision only if it's present in the message - because sometimes etcd sends something without a header, like:
                // {"error": {"grpc_code": 14, "http_code": 503, "http_status": "Service Unavailable", "message": "error reading from server: EOF"}}
@@ -508,9 +492,9 @@ void etcd_state_client_http_t::load_global_config()
    {
        if (err != "")
        {
+            fprintf(stderr, "Error reading configuration from etcd: %s\n", err.c_str());
            if (infinite_start)
            {
-                fprintf(stderr, "Error reading configuration from etcd: %s\n", err.c_str());
                tfd->set_timer(etcd_slow_timeout, false, [this](int timer_id)
                {
                    load_global_config();
@@ -518,7 +502,7 @@ void etcd_state_client_http_t::load_global_config()
            }
            else
            {
-                fatal_error("Error reading configuration from etcd: "+err);
+                exit(1);
            }
        }
    });
--- a/src/client/messenger.cpp
+++ b/src/client/messenger.cpp
@@ -17,20 +17,6 @@
 #include "msgr_rdma.h"
 #endif

-osd_messenger_t::osd_messenger_t()
-{
-    this->on_fatal_error_hook = [](const std::string & error)
-    {
-        fprintf(stderr, "%s\n", error.c_str());
-        exit(1);
-    };
-}
-
-void osd_messenger_t::fatal_error(const std::string & error)
-{
-    (std::function<void(const std::string&)>(on_fatal_error_hook))(error);
-}
-
 void osd_messenger_t::init()
 {
    init_tls();
@@ -212,7 +198,7 @@ osd_messenger_t::~osd_messenger_t()
    destroy_tls();
 }

-int osd_messenger_t::parse_proto_checksums(const json11::Json & val, int default_value)
+static int parse_proto_checksums(const json11::Json & val, int default_value)
 {
    if (val.is_string())
    {
@@ -230,9 +216,9 @@ int osd_messenger_t::parse_proto_checksums(const json11::Json & val, int default
    }
    else if (val.is_null())
        return default_value;
-    fatal_error("proto_checksums should be \"full\", \"payload\", \"gcm\", \"none\""
-        ", \"\" or null (default), but it is: "+val.dump());
-    return default_value;
+    fprintf(stderr, "proto_checksums should be \"full\", \"payload\", \"gcm\", \"none\""
+        ", \"\" or null (default), but it is: %s\n", val.dump().c_str());
+    exit(1);
 }

 void osd_messenger_t::parse_config(const json11::Json & config, bool init)
--- a/src/client/messenger.h
+++ b/src/client/messenger.h
@@ -57,6 +57,7 @@
 #define MSGR_SENDP_IN 2
 #define MSGR_SENDP_OUT 4
 #define MSGR_SENDP_REF 8
+#define MSGR_SENDP_CANCELED 16

 struct osd_client_t;

@@ -151,6 +152,7 @@ struct osd_client_t
    // Write state
    std::deque<osd_op_t *> write_ops;
    osd_op_t *write_op = NULL;
+    bool write_op_cancel = false;
    size_t write_op_pos = 0;
    msghdr write_msg = { 0 };
    int write_state = 0;
@@ -257,7 +259,6 @@ protected:
    bool gcm_enabled = false;
    msgr_handshake_ctx_i *hs_ctx = NULL;

-    void fatal_error(const std::string & error);
    void init_tls();
    void destroy_tls();
    bool init_tls_client(osd_client_t *cl);
@@ -313,17 +314,16 @@ public:
    void stop_client(uint64_t client_id, bool force_delete = false);
    void destroy_client(osd_client_t *cl);
    void outbox_push(osd_op_t *cur_op);
+    bool cancel_out(osd_op_t *cur_op);
    std::function<void(osd_op_t*)> exec_op;
    std::function<void(osd_num_t)> repeer_pgs;
    std::function<void(osd_num_t)> break_pg_locks;
    std::function<bool(osd_client_t*, json11::Json)> check_config_hook;
    std::function<void(osd_client_t*)> handshake_hook;
-    std::function<void(const std::string&)> on_fatal_error_hook;
    void read_requests();
    void send_replies();
    void accept_connections(int listen_fd);
    void destroy_iothreads();
-    osd_messenger_t();
    ~osd_messenger_t();

    static json11::Json::object read_config(const json11::Json & config);
@@ -347,7 +347,6 @@ public:
    void measure_exec(osd_op_t *cur_op);

 protected:
-    int parse_proto_checksums(const json11::Json & val, int default_value);
    void try_connect_peer(uint64_t osd_num);
    void try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_host, int peer_port);
    void handle_peer_epoll(int peer_fd, int epoll_events);
@@ -356,6 +355,7 @@ protected:
    void check_peer_config(osd_client_t *cl);
    void cancel_osd_ops(osd_client_t *cl);
    void cancel_op(osd_op_t *op);
+    osd_op_t *copy_out_op_to_ignore(osd_op_t *cur_op);

    bool try_send(osd_client_t *cl);
    bool wakeup_send(osd_client_t *cl);
--- a/src/client/msgr_encrypt.cpp
+++ b/src/client/msgr_encrypt.cpp
@@ -597,9 +597,10 @@ void osd_messenger_t::init_tls()
        if (client_cert.empty() || client_key.empty() || osd_ca.empty() || osd_num && client_ca.empty())
        {
            if (osd_num)
-                fatal_error("Vitastor transport encryption requires osd_cert, osd_pkey, osd_ca, client_ca options for OSDs");
+                fprintf(stderr, "Vitastor transport encryption requires osd_cert, osd_pkey, osd_ca, client_ca options for OSDs\n");
            else
-                fatal_error("Vitastor transport encryption requires cert, pkey and osd_ca options\n");
+                fprintf(stderr, "Vitastor transport encryption requires cert, pkey and osd_ca options\n");
+            exit(1);
        }
        else
        {
@@ -608,8 +609,8 @@ void osd_messenger_t::init_tls()
            hs_ctx = msgr_handshake_ctx_i::create_ctx();
            if (!hs_ctx->init(client_cert, client_key, osd_ca, client_ca, admin_ca))
            {
-                gcm_enabled = false;
-                fatal_error("Error: "+hs_ctx->get_error());
+                fprintf(stderr, "Error: %s\n", hs_ctx->get_error().c_str());
+                exit(1);
            }
 #endif
        }
--- a/src/client/msgr_handshake.cpp
+++ b/src/client/msgr_handshake.cpp
@@ -205,8 +205,6 @@ bool msgr_handshake_ctx_t::init(const std::string & pem_cert, const std::string
        return on_error("EVP_KDF_fetch: ");
    kdf_ctx = EVP_KDF_CTX_new(kdf);
    EVP_KDF_free(kdf);
-    if (!kdf_ctx)
-        return on_error("EVP_KDF_CTX_new: ");
    return true;
 }

@@ -247,9 +245,6 @@ bool msgr_handshake_ctx_t::on_error(const std::string & prefix)
 bool msgr_handshake_ctx_t::derive_kdf(const uint8_t* insecret, size_t insecret_len,
    const uint8_t* salt, size_t salt_len, const char *label, uint8_t *outsecret, size_t outsize)
 {
-    // Some OpenSSL versions (for example 3.0.13, 3.3.3, 3.5.0) ignore empty salt,
-    // so without reset the salt from the previous call (key rotation) would be used
-    EVP_KDF_CTX_reset(kdf_ctx);
    OSSL_PARAM params[5];
    int n = 0;
    params[n++] = OSSL_PARAM_construct_utf8_string("digest", (char*)"sha384", (size_t)7);
@@ -410,20 +405,14 @@ bool msgr_handshake_t::verify(const uint8_t *signature, size_t signature_len)
    if (EVP_DigestVerifyUpdate(md_ctx.get(), full_handshake.data(), full_handshake.size()) <= 0)
        return on_error("EVP_DigestVerifyUpdate: ");
    if (EVP_DigestVerifyFinal(md_ctx.get(), signature, signature_len) <= 0)
-    {
-        // Clear errors of the invalid signature, otherwise the next on_error() would report them
-        ERR_clear_error();
-        error = "Handshake signature verification failed";
-        state = MSGR_HS_ERROR;
        return false;
-    }
    return true;
 }

 bool msgr_handshake_t::encrypt(const uint8_t* src, size_t len, uint8_t* dest)
 {
    std::unique_ptr<EVP_CIPHER_CTX, decltype(&EVP_CIPHER_CTX_free)> enc_ctx(EVP_CIPHER_CTX_new(), EVP_CIPHER_CTX_free);
-    if (!enc_ctx)
+    if (!ctx)
        return on_error("EVP_CIPHER_CTX_new: ");
    if (EVP_EncryptInit_ex(enc_ctx.get(), EVP_aes_256_gcm(), NULL, hs_key.data(), hs_key.data() + AES_256_GCM_KEY_SIZE) <= 0)
        return on_error("EVP_EncryptInit AES-256-GCM: ");
@@ -449,7 +438,7 @@ bool msgr_handshake_t::decrypt(const uint8_t *src, size_t & len, uint8_t* dest)
        return false;
    }
    std::unique_ptr<EVP_CIPHER_CTX, decltype(&EVP_CIPHER_CTX_free)> dec_ctx(EVP_CIPHER_CTX_new(), EVP_CIPHER_CTX_free);
-    if (!dec_ctx)
+    if (!ctx)
        return on_error("EVP_CIPHER_CTX_new: ");
    if (EVP_DecryptInit_ex(dec_ctx.get(), EVP_aes_256_gcm(), NULL, peer_hs_key.data(), peer_hs_key.data() + AES_256_GCM_KEY_SIZE) <= 0)
        return on_error("EVP_DecryptInit AES-256-GCM: ");
--- a/src/client/msgr_op.h
+++ b/src/client/msgr_op.h
@@ -187,13 +187,17 @@ struct __attribute__((visibility("default"))) osd_op_t
    void *rmw_buf = NULL;
    std::shared_ptr<osd_op_enc_t> enc;
    uint64_t csum = 0; // network layer checksum
+    uint64_t send_free_pos = 0;
+    // Read the reply and throw it away instead of copying it anywhere.
+    // Used for stub operations left in place of canceled ones
+    bool ignore_reply = false;
    osd_primary_op_data_t* op_data = NULL;
    std::function<void(osd_op_t*)> callback;

    osd_op_buf_list_t iov;

    ~osd_op_t();
-    void cancel();
+    void cancel(int retval);

    bool is_recovery_related();
    uint64_t calc_data_checksum();
--- a/src/client/msgr_receive.cpp
+++ b/src/client/msgr_receive.cpp
@@ -179,14 +179,15 @@ public:
        }
        else if (flags & RDR_GCM)
        {
-            // Here, dst == NULL is not allowed
-            assert(dst != NULL);
+            // Here, dst == NULL means "decrypt in place and throw the data away".
+            // The data still has to be decrypted to keep the GCM stream and tag in sync
+            uint8_t *out = (dst ? dst+from : curbuf+done);
 #ifdef WITH_ISAL_CRYPTO
-            int r = isal_aes_gcm_dec_256_update(&cl->peer_key_isal, cl->dec_ctx, dst+from, curbuf+done, n);
+            int r = isal_aes_gcm_dec_256_update(&cl->peer_key_isal, cl->dec_ctx, out, curbuf+done, n);
            assert(!r);
 #else
            int actual_out;
-            if (EVP_DecryptUpdate(cl->dec_ctx, dst+from, &actual_out, curbuf+done, n) != 1)
+            if (EVP_DecryptUpdate(cl->dec_ctx, out, &actual_out, curbuf+done, n) != 1)
            {
                fprintf(stderr, "DecryptUpdate error: ");
                ERR_print_errors_fp(stderr);
@@ -196,7 +197,7 @@ public:
 #endif
            if (cl->read_csum_state && !(flags & RDR_NO_CSUM))
            {
-                XXH3_64bits_update(cl->read_csum_state, dst+from, n);
+                XXH3_64bits_update(cl->read_csum_state, out, n);
            }
            done += n;
        }
@@ -307,6 +308,8 @@ public:
    get_op_reader_t(osd_messenger_t* msgr, osd_client_t* cl):
        cl(cl), from(cl->read_op_pos), mpos(false)
    {
+        // Discarded replies are always read through in_buf - this reader needs real buffers
+        assert(!cl->read_op->ignore_reply);
        if (cl->read_op->op_type == OSD_OP_OUT &&
            cl->read_op->reply.hdr.opcode == OSD_OP_READ &&
            cl->read_op->reply.hdr.retval > 0)
@@ -397,7 +400,8 @@ void osd_messenger_t::read_requests()
            continue;
        }
        auto cl = cl_it->second;
-        if (!rxbounce && cl->read_op && cl->read_op_pos >= OSD_PACKET_SIZE && cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size)
+        if (!rxbounce && cl->read_op && !cl->read_op->ignore_reply && cl->read_op_pos >= OSD_PACKET_SIZE &&
+            cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size)
        {
            get_op_reader_t rdr(this, cl);
            if (!op_read_from(cl, rdr))
@@ -570,11 +574,15 @@ void osd_messenger_t::handle_immediate_ops()
            else
                delete op;
        }
-        else
+        else if (op->callback)
        {
            // Copy lambda to be unaffected by `delete op`
            std::function<void(osd_op_t*)>(op->callback)(op);
        }
+        else
+        {
+            delete op;
+        }
    }
 }

@@ -859,12 +867,12 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
        }
        if (op->reply.hdr.retval >= 0 && bmp_len > 0)
        {
-            assert(op->bitmap);
+            assert(op->ignore_reply || op->bitmap);
            cl->read_op_size += bmp_len;
        }
        if (op->reply.hdr.retval > 0)
        {
-            assert(op->iov.count > 0);
+            assert(op->ignore_reply || op->iov.count > 0);
            cl->read_op_size += op->reply.hdr.retval;
        }
    }
@@ -878,7 +886,8 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
                cl->client_id, op->reply.hdr.retval);
            return false;
        }
-        op->buf = malloc_or_die(cl->read_op_size);
+        if (!op->ignore_reply)
+            op->buf = malloc_or_die(cl->read_op_size);
    }
    else if (op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP && op->reply.hdr.retval > 0)
    {
@@ -891,8 +900,11 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
            return false;
        }
        cl->read_op_size = op->reply.hdr.retval;
-        free(op->buf);
-        op->buf = malloc_or_die(cl->read_op_size);
+        if (!op->ignore_reply)
+        {
+            free(op->buf);
+            op->buf = malloc_or_die(cl->read_op_size);
+        }
    }
    else if (op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG && op->reply.hdr.retval > 0)
    {
@@ -903,14 +915,20 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
                cl->client_id, op->reply.hdr.retval, MAX_SIMPLE_PAYLOAD_SIZE);
            return false;
        }
-        free(op->buf);
-        op->buf = malloc_or_die(op->reply.hdr.retval);
+        if (!op->ignore_reply)
+        {
+            free(op->buf);
+            op->buf = malloc_or_die(op->reply.hdr.retval);
+        }
    }
    else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0)
    {
        cl->read_op_size = op->reply.describe.result_bytes;
-        free(op->buf);
-        op->buf = malloc_or_die(op->reply.describe.result_bytes);
+        if (!op->ignore_reply)
+        {
+            free(op->buf);
+            op->buf = malloc_or_die(op->reply.describe.result_bytes);
+        }
    }
    if (cl->proto_csum_status == MSGR_CSUM_FULL ||
        cl->read_op_size > 0 && cl->proto_csum_status == MSGR_CSUM_PAYLOAD)
@@ -974,7 +992,37 @@ bool osd_messenger_t::op_read_from(osd_client_t *cl, msgr_op_reader_t & rdr)
        if (!rdr.read(op->reply.buf, OSD_PACKET_SIZE, RDR_GCM | (cl->proto_csum_status == MSGR_CSUM_PAYLOAD ? RDR_NO_CSUM : 0)))
            return false;
 switched_type:
-        if (op->reply.hdr.opcode == OSD_OP_SEC_READ)
+        if (op->ignore_reply)
+        {
+            // Reply to a canceled operation - read it into nowhere and throw it away.
+            // Read flags have to exactly match the normal path below, otherwise
+            // the GCM stream and the protocol checksum would desync
+            if (op->reply.hdr.opcode == OSD_OP_SEC_READ || op->reply.hdr.opcode == OSD_OP_READ)
+            {
+                // Conditions mirror allocate_reply_buffers() which defines cl->read_op_size,
+                // so that we always consume exactly the framed length
+                uint32_t bmp_len = (op->reply.hdr.opcode == OSD_OP_SEC_READ
+                    ? op->reply.sec_rw.attr_len : op->reply.rw.bitmap_len);
+                if (op->reply.hdr.retval >= 0 && bmp_len > 0 && !rdr.read(NULL, bmp_len, RDR_GCM))
+                    return false;
+                // Note that RDR_XTS is intentionally not passed - the stub has no keys and
+                // XTS is length-preserving, so both the GCM stream and the checksum match
+                if (op->reply.hdr.retval > 0 && !rdr.read(NULL, op->reply.hdr.retval,
+                    (cl->proto_csum_status == MSGR_CSUM_GCM ? RDR_GCM : 0)))
+                    return false;
+            }
+            else
+            {
+                // All other replies have a single payload region, always read with RDR_GCM
+                size_t payload = cl->read_op_size;
+                if (cl->proto_csum_status == MSGR_CSUM_FULL ||
+                    payload > 0 && cl->proto_csum_status == MSGR_CSUM_PAYLOAD)
+                    payload -= 8;
+                if (payload > 0 && !rdr.read(NULL, payload, RDR_GCM))
+                    return false;
+            }
+        }
+        else if (op->reply.hdr.opcode == OSD_OP_SEC_READ)
        {
            if (op->reply.sec_rw.attr_len > 0)
            {
--- a/src/client/msgr_send.cpp
+++ b/src/client/msgr_send.cpp
@@ -436,7 +436,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
    auto cl_it = clients.find(cur_op->client_id);
    if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
    {
-        cur_op->cancel();
+        cur_op->cancel(-EPIPE);
        return;
    }
    osd_client_t *cl = cl_it->second;
@@ -494,6 +494,81 @@ bool osd_messenger_t::wakeup_send(osd_client_t *cl)
    return true;
 }

+// Try to cancel an outbound operation
+// Returns true when canceled synchronously, false if canceling is impossible or will be
+// performed asynchronously and the caller has to wait for the callback to be called
+bool osd_messenger_t::cancel_out(osd_op_t *cur_op)
+{
+    assert(cur_op->client_id);
+    assert(cur_op->op_type == OSD_OP_OUT);
+    auto cl_it = clients.find(cur_op->client_id);
+    if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
+    {
+        // Client is stopped => messenger will call the callback anyway
+        return false;
+    }
+    osd_client_t *cl = cl_it->second;
+    auto sent_it = cl->sent_ops.find(cur_op->req.hdr.id);
+    if (sent_it == cl->sent_ops.end())
+    {
+        // Reply header is already received => callback is pending
+        return false;
+    }
+    if (sent_it->second != cur_op)
+    {
+        // Already canceled
+        return false;
+    }
+    // Make a copy to have some space to place the reply to and forget the original object
+    if (cl->write_op == cur_op)
+    {
+        // Operation is being sent => wait for full send, then cancel it
+        sent_it->second = copy_out_op_to_ignore(cur_op);
+        cl->write_op_cancel = true;
+        return false;
+    }
+    for (auto wr_it = cl->write_ops.begin(); wr_it != cl->write_ops.end(); wr_it++)
+    {
+        if (*wr_it == cur_op)
+        {
+            // Operation is only queued => just remove it from the send queue
+            cl->sent_ops.erase(sent_it);
+            cl->write_ops.erase(wr_it);
+            cur_op->callback = NULL;
+            return true;
+        }
+    }
+    sent_it->second = copy_out_op_to_ignore(cur_op);
+    if (cur_op->send_free_pos >= cl->next_send_free)
+    {
+        // In progress of sending => free it when the send_free reference is cleared
+        assert(cur_op->send_free_pos < cl->next_send_free + cl->send_free.size());
+        auto & sf = cl->send_free[cur_op->send_free_pos - cl->next_send_free];
+        assert((osd_op_t*)sf.buf == cur_op);
+        sf.flags |= MSGR_SENDP_CANCELED;
+        return false;
+    }
+    // Operation is fully sent and not referenced by the send_free queue
+    cur_op->callback = NULL;
+    return true;
+}
+
+// Make a stub operation to swallow the reply to a canceled operation.
+// The reply body is read into nowhere, so no data buffers are needed here -
+// only the request header, which is required to validate the reply
+osd_op_t *osd_messenger_t::copy_out_op_to_ignore(osd_op_t *cur_op)
+{
+    osd_op_t *op_copy = new osd_op_t;
+    op_copy->op_type = OSD_OP_OUT;
+    op_copy->ignore_reply = true;
+    op_copy->tv_begin = cur_op->tv_begin;
+    op_copy->client_id = cur_op->client_id;
+    op_copy->osd_num = cur_op->osd_num;
+    op_copy->bitmap_len = cur_op->bitmap_len;
+    memcpy(op_copy->req.buf, cur_op->req.buf, OSD_PACKET_SIZE);
+    return op_copy;
+}
+
 bool osd_messenger_t::try_send(osd_client_t *cl)
 {
    assert(cl->peer_state != PEER_RDMA);
@@ -686,7 +761,11 @@ size_t osd_messenger_t::copy_ops_to_with(osd_client_t *cl, uint8_t *dst, size_t
            break;
        }
        assert(cl->send_free.size() < UINT64_MAX - cl->next_send_free);
-        cl->send_free.push_back({ .buf = op, .flags = (uint32_t)((op->op_type == OSD_OP_IN ? MSGR_SENDP_IN : MSGR_SENDP_OUT) | MSGR_SENDP_REF) });
+        uint32_t flags = ((op->op_type == OSD_OP_IN ? MSGR_SENDP_IN : MSGR_SENDP_OUT) | MSGR_SENDP_REF);
+        if (cl->write_op_cancel)
+            flags |= MSGR_SENDP_CANCELED;
+        op->send_free_pos = cl->next_send_free + cl->send_free.size();
+        cl->send_free.push_back({ .buf = op, .flags = flags });
    }
    return wr.get_done();
 }
@@ -694,6 +773,7 @@ size_t osd_messenger_t::copy_ops_to_with(osd_client_t *cl, uint8_t *dst, size_t
 void osd_messenger_t::next_write_op(osd_client_t *cl)
 {
    cl->write_op = cl->write_ops.front();
+    cl->write_op_cancel = false;
    cl->write_ops.pop_front();
    if (cl->proto_csum_status == MSGR_CSUM_FULL || cl->proto_csum_status == MSGR_CSUM_PAYLOAD)
    {
--- a/src/client/msgr_stop.cpp
+++ b/src/client/msgr_stop.cpp
@@ -22,24 +22,24 @@ void osd_client_t::cancel_ops()
    sent_ops.clear();
    for (auto op: cancel_ops)
    {
-        op->cancel();
+        op->cancel(-EPIPE);
    }
 }

-void osd_op_t::cancel()
+void osd_op_t::cancel(int retval)
 {
    if (op_type == OSD_OP_OUT && callback)
    {
        reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
        reply.hdr.id = req.hdr.id;
        reply.hdr.opcode = req.hdr.opcode;
-        reply.hdr.retval = -EPIPE;
+        reply.hdr.retval = retval;
        // Copy lambda to be unaffected by `delete this`
        (std::function<void(osd_op_t*)>(callback))(this);
    }
    else
    {
-        // This function is only called in stop_client(), so it's fine to destroy the operation
+        // It's fine to just free responses to incoming ops
        delete this;
    }
 }
@@ -57,6 +57,8 @@ void osd_client_t::unref_send_free(uint64_t from, uint64_t to)
        {
            if (it->flags & MSGR_SENDP_IN)
                delete (osd_op_t*)it->buf;
+            else if (it->flags & MSGR_SENDP_CANCELED)
+                ((osd_op_t*)it->buf)->cancel(-ECANCELED);
            else if (it->flags & MSGR_SENDP_BUF)
                free(it->buf);
        }
@@ -67,6 +69,8 @@ void osd_client_t::unref_send_free(uint64_t from, uint64_t to)
        {
            if (it->flags & MSGR_SENDP_IN)
                delete (osd_op_t*)it->buf;
+            else if (it->flags & MSGR_SENDP_CANCELED)
+                ((osd_op_t*)it->buf)->cancel(-ECANCELED);
            else if (it->flags & MSGR_SENDP_BUF)
                free(it->buf);
            to++;
@@ -204,14 +208,21 @@ osd_client_t::~osd_client_t()
    if (read_op)
    {
        // read_op may be an incoming op or a continued response for an outbound op
-        read_op->cancel();
+        read_op->cancel(-EPIPE);
        read_op = NULL;
    }
    // Delete unsent replies - before cancel_ops() because it may delete outbound operations.
    if (write_op)
    {
        if (write_op->op_type == OSD_OP_IN)
+        {
            delete write_op;
+        }
+        else if (write_op_cancel)
+        {
+            // not in sent_ops anymore - should be canceled here
+            write_op->cancel(-ECANCELED);
+        }
        write_op = NULL;
    }
    for (auto op: write_ops)
--- a/src/client/qemu_driver.c
+++ b/src/client/qemu_driver.c
@@ -1160,7 +1160,7 @@ static void vitastor_co_read_bitmap_cb(void *opaque, long retval, uint8_t *bitma
 }

 static int coroutine_fn vitastor_co_block_status(BlockDriverState *bs,
-#if QEMU_VERSION_MAJOR > 10 || QEMU_VERSION_MAJOR == 10 && QEMU_VERSION_MINOR >= 1 || defined BDRV_WANT_PRECISE
+#if QEMU_VERSION_MAJOR > 10 || QEMU_VERSION_MAJOR == 10 && QEMU_VERSION_MINOR >= 1
    unsigned int mode,
 #else
    bool want_zero,
@@ -1171,7 +1171,7 @@ static int coroutine_fn vitastor_co_block_status(BlockDriverState *bs,
    // Not allocated => return 0
    // Error => return -errno
    // Set pnum to length of the extent, `*map` = `offset`, `*file` = `bs`
-#if QEMU_VERSION_MAJOR > 10 || QEMU_VERSION_MAJOR == 10 && QEMU_VERSION_MINOR >= 1 || defined BDRV_WANT_PRECISE
+#if QEMU_VERSION_MAJOR > 10 || QEMU_VERSION_MAJOR == 10 && QEMU_VERSION_MINOR >= 1
    int want_zero = (mode == BDRV_WANT_PRECISE);
 #endif
    VitastorRPC task;
--- a/src/client/vitastor.pc.in
+++ b/src/client/vitastor.pc.in
@@ -6,7 +6,7 @@ includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@

 Name: Vitastor
 Description: Vitastor client library
-Version: 3.2.2
+Version: 3.2.1
 Libs: -L${libdir} -lvitastor_client
 Cflags: -I${includedir}

--- a/src/client/vitastor_c.cpp
+++ b/src/client/vitastor_c.cpp
@@ -30,18 +30,6 @@ struct vitastor_c

    QEMUSetFDHandler *aio_set_fd_handler = NULL;
    void *aio_ctx = NULL;
-
-    ~vitastor_c()
-    {
-        if (ringloop)
-            delete ringloop;
-        if (epmgr)
-            delete epmgr;
-        else if (tfd)
-            delete tfd;
-        if (cli)
-            delete cli;
-    }
 };

 extern "C" {
@@ -115,16 +103,7 @@ vitastor_c *vitastor_c_create_qemu(QEMUSetFDHandler *aio_set_fd_handler, void *a
        rdma_device, rdma_port_num, rdma_gid_index, rdma_mtu, log_level
    );
    auto self = vitastor_c_create_qemu_common(aio_set_fd_handler, aio_context);
-    try
-    {
-        self->cli = cluster_client_t::create(NULL, self->tfd, cfg_json);
-    }
-    catch (std::exception & e)
-    {
-        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
-        delete self;
-        return NULL;
-    }
+    self->cli = cluster_client_t::create(NULL, self->tfd, cfg_json);
    return self;
 }

@@ -132,45 +111,47 @@ vitastor_c *vitastor_c_create_qemu_uring(QEMUSetFDHandler *aio_set_fd_handler, v
    const char *config_path, const char *etcd_host, const char *etcd_prefix,
    int use_rdma, const char *rdma_device, int rdma_port_num, int rdma_gid_index, int rdma_mtu, int log_level)
 {
+    ring_loop_t *ringloop = NULL;
+    try
+    {
+        ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
+    }
+    catch (std::exception & e)
+    {
+        return NULL;
+    }
    json11::Json cfg_json = vitastor_c_common_config(
        config_path, etcd_host, etcd_prefix, use_rdma,
        rdma_device, rdma_port_num, rdma_gid_index, rdma_mtu, log_level
    );
    auto self = vitastor_c_create_qemu_common(aio_set_fd_handler, aio_context);
-    try
-    {
-        self->ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
-        self->cli = cluster_client_t::create(self->ringloop, self->tfd, cfg_json);
-    }
-    catch (std::exception & e)
-    {
-        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
-        delete self;
-        return NULL;
-    }
+    self->ringloop = ringloop;
+    self->cli = cluster_client_t::create(self->ringloop, self->tfd, cfg_json);
+    ringloop->loop();
    return self;
 }

 vitastor_c *vitastor_c_create_uring(const char *config_path, const char *etcd_host, const char *etcd_prefix,
    int use_rdma, const char *rdma_device, int rdma_port_num, int rdma_gid_index, int rdma_mtu, int log_level)
 {
+    ring_loop_t *ringloop = NULL;
+    try
+    {
+        ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
+    }
+    catch (std::exception & e)
+    {
+        return NULL;
+    }
    json11::Json cfg_json = vitastor_c_common_config(
        config_path, etcd_host, etcd_prefix, use_rdma,
        rdma_device, rdma_port_num, rdma_gid_index, rdma_mtu, log_level
    );
    vitastor_c *self = new vitastor_c;
-    try
-    {
-        self->ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
-        self->epmgr = new epoll_manager_t(self->ringloop);
-        self->cli = cluster_client_t::create(self->ringloop, self->epmgr->tfd, cfg_json);
-    }
-    catch (std::exception & e)
-    {
-        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
-        delete self;
-        return NULL;
-    }
+    self->ringloop = ringloop;
+    self->epmgr = new epoll_manager_t(self->ringloop);
+    self->cli = cluster_client_t::create(self->ringloop, self->epmgr->tfd, cfg_json);
+    ringloop->loop();
    return self;
 }

@@ -185,6 +166,15 @@ int vitastor_c_uring_register_eventfd(vitastor_c *client)

 vitastor_c *vitastor_c_create_uring_json(const char **options, int options_len)
 {
+    ring_loop_t *ringloop = NULL;
+    try
+    {
+        ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
+    }
+    catch (std::exception & e)
+    {
+        return NULL;
+    }
    json11::Json::object cfg;
    for (int i = 0; i < options_len-1; i += 2)
    {
@@ -199,18 +189,10 @@ vitastor_c *vitastor_c_create_uring_json(const char **options, int options_len)
    }
    json11::Json cfg_json(cfg);
    vitastor_c *self = new vitastor_c;
-    try
-    {
-        self->ringloop = new ring_loop_t(RINGLOOP_DEFAULT_SIZE);
-        self->epmgr = new epoll_manager_t(self->ringloop);
-        self->cli = cluster_client_t::create(self->ringloop, self->epmgr->tfd, cfg_json);
-    }
-    catch (std::exception & e)
-    {
-        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
-        delete self;
-        return NULL;
-    }
+    self->ringloop = ringloop;
+    self->epmgr = new epoll_manager_t(self->ringloop);
+    self->cli = cluster_client_t::create(self->ringloop, self->epmgr->tfd, cfg_json);
+    ringloop->loop();
    return self;
 }

@@ -223,17 +205,8 @@ vitastor_c *vitastor_c_create_epoll_json(const char **options, int options_len)
    }
    json11::Json cfg_json(cfg);
    vitastor_c *self = new vitastor_c;
-    try
-    {
-        self->epmgr = new epoll_manager_t(NULL);
-        self->cli = cluster_client_t::create(NULL, self->epmgr->tfd, cfg_json);
-    }
-    catch (std::exception & e)
-    {
-        fprintf(stderr, "Failed to create Vitastor client: %s\n", e.what());
-        delete self;
-        return NULL;
-    }
+    self->epmgr = new epoll_manager_t(NULL);
+    self->cli = cluster_client_t::create(NULL, self->epmgr->tfd, cfg_json);
    return self;
 }

@@ -244,6 +217,13 @@ void* vitastor_c_get_internal_client(vitastor_c *client)

 void vitastor_c_destroy(vitastor_c *client)
 {
+    delete client->cli;
+    if (client->epmgr)
+        delete client->epmgr;
+    else if (client->tfd)
+        delete client->tfd;
+    if (client->ringloop)
+        delete client->ringloop;
    delete client;
 }

--- a/src/test/CMakeLists.txt
+++ b/src/test/CMakeLists.txt
@@ -27,6 +27,17 @@ target_link_libraries(stub_uring_osd
 	${RDMACM_LIBRARIES}
 )

+# test_messenger
+add_executable(test_messenger EXCLUDE_FROM_ALL test_messenger.cpp)
+target_link_libraries(test_messenger
+	vitastor_net
+	${LIBURING_LIBRARIES}
+	${IBVERBS_LIBRARIES}
+	${RDMACM_LIBRARIES}
+)
+add_dependencies(build_tests test_messenger)
+add_test(NAME test_messenger COMMAND test_messenger)
+
 # test_allocator
 add_executable(test_allocator EXCLUDE_FROM_ALL test_allocator.cpp)
 target_link_libraries(test_allocator util)
@@ -88,9 +99,3 @@ add_test(NAME test_blockstore_chaos COMMAND test_blockstore_chaos)
 # test_atomic
 add_executable(test_atomic test_atomic.cpp)
 target_link_libraries(test_atomic util)
-
-# test_msgr_kdf
-add_executable(test_msgr_kdf EXCLUDE_FROM_ALL test_msgr_kdf.cpp)
-target_link_libraries(test_msgr_kdf vitastor_common)
-add_dependencies(build_tests test_msgr_kdf)
-add_test(NAME test_msgr_kdf COMMAND test_msgr_kdf)
--- a/src/test/mock/messenger.cpp
+++ b/src/test/mock/messenger.cpp
@@ -25,7 +25,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
    auto cl_it = clients.find(cur_op->client_id);
    if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
    {
-        cur_op->cancel();
+        cur_op->cancel(-EPIPE);
        return;
    }
    auto cl = cl_it->second;
@@ -34,6 +34,35 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
    cl->sent_ops[cur_op->req.hdr.id] = cur_op;
 }

+// Set to true to make cancel_out() pretend that operations are already sent
+// and have to be canceled asynchronously
+bool mock_msgr_async_cancel = false;
+
+bool osd_messenger_t::cancel_out(osd_op_t *cur_op)
+{
+    assert(cur_op->client_id);
+    assert(cur_op->op_type == OSD_OP_OUT);
+    auto cl_it = clients.find(cur_op->client_id);
+    if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
+    {
+        return false;
+    }
+    auto cl = cl_it->second;
+    auto sent_it = cl->sent_ops.find(cur_op->req.hdr.id);
+    if (sent_it == cl->sent_ops.end() || sent_it->second != cur_op)
+    {
+        return false;
+    }
+    if (mock_msgr_async_cancel)
+    {
+        // The caller is expected to "complete" the operation with -ECANCELED itself
+        return false;
+    }
+    cl->sent_ops.erase(sent_it);
+    cur_op->callback = NULL;
+    return true;
+}
+
 void osd_messenger_t::parse_config(const json11::Json & config, bool init)
 {
 }
--- a/src/test/test_cluster_client.cpp
+++ b/src/test/test_cluster_client.cpp
@@ -12,6 +12,9 @@
 #include "cluster_client_impl.h"
 #include "msgr_encrypt.h"

+// Defined in test/mock/messenger.cpp
+extern bool mock_msgr_async_cancel;
+
 class cluster_client_test_t
 {
 public:
@@ -53,6 +56,9 @@ void configure_single_pg_pool(etcd_state_client_mock_t *mock)
    });
 }

+// Last operation posted by test_write()/test_read()/test_sync(), for cancellation tests
+cluster_op_t *last_posted_op = NULL;
+
 int *test_write(cluster_client_t *cli, uint64_t offset, uint64_t len, uint8_t c, std::function<void()> cb = NULL, bool instant = false)
 {
    printf("Post write %jx+%jx\n", offset, len);
@@ -78,6 +84,7 @@ int *test_write(cluster_client_t *cli, uint64_t offset, uint64_t len, uint8_t c,
        if (cb != NULL)
            cb();
    };
+    last_posted_op = op;
    cli->execute(op);
    if (instant)
    {
@@ -113,6 +120,7 @@ int *test_read(cluster_client_t *cli, uint64_t offset, uint64_t len, std::functi
        if (cb != NULL)
            cb();
    };
+    last_posted_op = op;
    cli->execute(op);
    return r;
 }
@@ -129,10 +137,11 @@ int *test_sync(cluster_client_t *cli)
        if (*r == -1000)
            printf("Error: Not allowed to complete yet\n");
        assert(*r != -1000);
-        *r = op->retval == 0 ? 1 : 0;
+        *r = op->retval == 0 ? 1 : op->retval;
        printf("Done sync r=%d\n", op->retval);
        delete op;
    };
+    last_posted_op = op;
    cli->execute(op);
    return r;
 }
@@ -177,7 +186,7 @@ void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)
 }

 #define check_disconnected(cli, osd_num) {\
-    if (cli->msgr->osd_peers.find(osd_num) != cli->msgr->osd_peers.end())\
+    if (cli->msgr.osd_peers.find(osd_num) != cli->msgr.osd_peers.end())\
    {\
        printf("OSD %ju not disconnected as it ought to be\n", (uint64_t)osd_num);\
        assert(0);\
@@ -185,7 +194,7 @@ void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)
 }

 #define check_op_count(cli, osd_num, ops) {\
-    osd_client_t *cl = cli->msgr->osd_peers.at(osd_num);\
+    osd_client_t *cl = cli->msgr.osd_peers.at(osd_num);\
    int real_ops = cl->sent_ops.size();\
    if (real_ops != ops)\
    {\
@@ -196,7 +205,7 @@ void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)

 osd_op_t *find_op(cluster_client_t *cli, osd_num_t osd_num, uint64_t opcode, uint64_t offset, uint64_t len)
 {
-    osd_client_t *cl = cli->msgr->osd_peers.at(osd_num);
+    osd_client_t *cl = cli->msgr.osd_peers.at(osd_num);
    auto op_it = cl->sent_ops.begin();
    while (op_it != cl->sent_ops.end())
    {
@@ -225,7 +234,7 @@ void pretend_op_completed(cluster_client_t *cli, osd_op_t *op, int64_t retval)
        ? "sync" : (op->req.hdr.opcode == OSD_OP_WRITE ? "write" : "read"), op->req.rw.offset, op->req.rw.len);
    uint64_t op_id = op->req.hdr.id;
    uint64_t client_id = op->client_id;
-    cli->msgr->clients[client_id]->sent_ops.erase(op_id);
+    cli->msgr.clients[client_id]->sent_ops.erase(op_id);
    op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
    op->reply.hdr.id = op->req.hdr.id;
    op->reply.hdr.opcode = op->req.hdr.opcode;
@@ -308,7 +317,7 @@ void test1()
        uint64_t replay_start = UINT64_MAX;
        uint64_t replay_end = 0;
        std::vector<osd_op_t*> replay_ops;
-        auto osd_cl = cli->msgr->osd_peers.at(1);
+        auto osd_cl = cli->msgr.osd_peers.at(1);
        for (auto & op_p: osd_cl->sent_ops)
        {
            auto op = op_p.second;
@@ -739,7 +748,7 @@ void test_deoptimize_snapshot_read()
        };
        cli->execute(op);
        check_op_count(cli, 1, 1);
-        auto cl = cli->msgr->osd_peers.at(1);
+        auto cl = cli->msgr.osd_peers.at(1);
        osd_op_t *read_op = cl->sent_ops.begin()->second;
        assert(read_op->req.hdr.opcode == OSD_OP_READ);
        assert(read_op->req.rw.inode == child_inode);
@@ -781,7 +790,7 @@ void test_deoptimize_snapshot_read()
        };
        cli->execute(op);
        check_op_count(cli, 1, 1);
-        auto cl = cli->msgr->osd_peers.at(1);
+        auto cl = cli->msgr.osd_peers.at(1);
        osd_op_t *read_op = cl->sent_ops.begin()->second;
        assert(read_op->req.hdr.opcode == OSD_OP_READ);
        assert(read_op->req.rw.inode == child_inode);
@@ -815,7 +824,7 @@ void test_deoptimize_snapshot_read()
        };
        cli->execute(op);
        check_op_count(cli, 1, 1);
-        auto cl = cli->msgr->osd_peers.at(1);
+        auto cl = cli->msgr.osd_peers.at(1);
        osd_op_t *child_read = cl->sent_ops.begin()->second;
        assert(child_read->req.hdr.opcode == OSD_OP_READ);
        assert(child_read->req.rw.inode == child_inode);
@@ -1122,6 +1131,127 @@ void test_snapshot_over_nonexistent_pool()
    printf("[ok] test_snapshot_over_nonexistent_pool\n");
 }

+void test_cancel()
+{
+    json11::Json config;
+    timerfd_manager_t *tfd = new timerfd_manager_t([](int fd, bool wr, std::function<void(int, int)> callback){});
+    etcd_state_client_mock_t *mock = new etcd_state_client_mock_t();
+    mock->pause();
+    cluster_client_t *cli = new cluster_client_t(NULL, tfd, config, std::unique_ptr<etcd_state_client_t>(mock));
+
+    // Cancel an operation posted before the configuration is loaded
+    int *r1 = test_write(cli, 0, 4096, 0x55);
+    cluster_op_t *op1 = last_posted_op;
+    can_complete(r1);
+    assert(cli->cancel(op1));
+    assert(*r1 == -ECANCELED);
+    delete r1;
+
+    configure_single_pg_pool(mock);
+    mock->resume();
+    pretend_connected(cli, 1);
+
+    // Cancel an operation which is queued, but not sent because the OSD is down
+    pretend_disconnected(cli, 1);
+    r1 = test_write(cli, 0, 4096, 0x56);
+    op1 = last_posted_op;
+    can_complete(r1);
+    assert(cli->cancel(op1));
+    assert(*r1 == -ECANCELED);
+    delete r1;
+    // The canceled write must not be repeated when the OSD comes back
+    pretend_connected(cli, 1);
+    check_op_count(cli, 1, 0);
+
+    // Cancel an already sent operation - the messenger cancels it synchronously
+    r1 = test_write(cli, 0, 4096, 0x57);
+    op1 = last_posted_op;
+    check_op_count(cli, 1, 1);
+    can_complete(r1);
+    assert(cli->cancel(op1));
+    assert(*r1 == -ECANCELED);
+    delete r1;
+    check_op_count(cli, 1, 0);
+    // The connection must NOT be dropped because of a cancellation
+    assert(cli->msgr.osd_peers.find(1) != cli->msgr.osd_peers.end());
+
+    // Cancel an already sent operation - the messenger cancels it asynchronously
+    mock_msgr_async_cancel = true;
+    r1 = test_write(cli, 0, 4096, 0x58);
+    op1 = last_posted_op;
+    check_op_count(cli, 1, 1);
+    osd_op_t *part = find_op(cli, 1, OSD_OP_WRITE, 0, 4096);
+    assert(!cli->cancel(op1));
+    can_complete(r1);
+    pretend_op_completed(cli, part, -ECANCELED);
+    assert(*r1 == -ECANCELED);
+    delete r1;
+    assert(cli->msgr.osd_peers.find(1) != cli->msgr.osd_peers.end());
+    mock_msgr_async_cancel = false;
+
+    // Cancel a sync which waits for a previous write
+    r1 = test_write(cli, 0, 4096, 0x59);
+    check_op_count(cli, 1, 1);
+    int *r2 = test_sync(cli);
+    cluster_op_t *op2 = last_posted_op;
+    check_op_count(cli, 1, 1);
+    can_complete(r2);
+    assert(cli->cancel(op2));
+    assert(*r2 == -ECANCELED);
+    delete r2;
+    // The write must complete as usual
+    can_complete(r1);
+    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 4096), 0);
+    check_completed(r1);
+
+    // Cancel an already sent sync - the next sync must still sync the same OSD
+    r2 = test_sync(cli);
+    op2 = last_posted_op;
+    check_op_count(cli, 1, 1);
+    can_complete(r2);
+    assert(cli->cancel(op2));
+    assert(*r2 == -ECANCELED);
+    delete r2;
+    check_op_count(cli, 1, 0);
+    r2 = test_sync(cli);
+    check_op_count(cli, 1, 1);
+    can_complete(r2);
+    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
+    check_completed(r2);
+
+    // Cancel a CAS write
+    int *r3 = new int;
+    *r3 = -1000;
+    cluster_op_t *op3 = new cluster_op_t();
+    op3->opcode = OSD_OP_WRITE;
+    op3->inode = 0x1000000000001;
+    op3->offset = 0;
+    op3->len = 4096;
+    op3->version = 1;
+    op3->iov.push_back(malloc_or_die(4096), 4096);
+    memset(op3->iov.buf[0].iov_base, 0x5a, 4096);
+    op3->callback = [r3](cluster_op_t *op)
+    {
+        assert(*r3 != -1000);
+        *r3 = op->retval;
+        printf("Done CAS write %jx+%jx r=%d\n", op->offset, op->len, op->retval);
+        free(op->iov.buf[0].iov_base);
+        delete op;
+    };
+    printf("Post CAS write 0+1000\n");
+    cli->execute(op3);
+    check_op_count(cli, 1, 1);
+    can_complete(r3);
+    assert(cli->cancel(op3));
+    assert(*r3 == -ECANCELED);
+    delete r3;
+    check_op_count(cli, 1, 0);
+
+    delete cli;
+    delete tfd;
+    printf("[ok] test_cancel\n");
+}
+
 int main(int narg, char *args[])
 {
    test1();
@@ -1135,5 +1265,6 @@ int main(int narg, char *args[])
    cluster_client_test_t::test_vault();
    test_pg_config_disappears();
    test_snapshot_over_nonexistent_pool();
+    test_cancel();
    return 0;
 }
--- a/src/test/test_heap.cpp
+++ b/src/test/test_heap.cpp
@@ -1010,211 +1010,6 @@ void test_iterate_compaction()
        assert(small_writes == 1);
    }

-    {
-        blockstore_heap_t heap(&dsk, buffer_area.data());
-        heap.finish_recheck();
-
-        // Case: BIG_STABLE(v1 l1) SMALL_STABLE(v2 l2) BIG_STABLE(v3 l3) un-fsynced
-        // -> skip compaction of l2 into l1 EVEN WHEN under pressure,
-        // because add_compact rejects stable overwrites with EBUSY
-        _test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 4096, buffer_area.data());
-        _test_small_write(heap, dsk, 1, 0, 2, 0, 4096, 0, true, buffer_area.data(), false);
-
-        heap.use_data(INODE_WITH_POOL(1, 1), 128*1024);
-        uint32_t mblock = 999999;
-        res = _test_do_big_write(heap, dsk, 1, 0, 3, 128*1024, true, 0, 0, buffer_area.data(), &mblock);
-        assert(res == 0);
-        heap.start_block_write(mblock);
-
-        object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
-        auto obj = heap.read_entry(oid);
-        assert(obj->lsn == 3);
-        assert(obj->is_overwrite());
-        assert(heap.get_fsynced_lsn() == 2);
-
-        // When not under pressure: postpone compaction
-        int small_writes = 0;
-        auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
-        {
-            small_writes++;
-        });
-        assert(compact_info.compact_lsn == 0);
-        assert(compact_info.compact_version == 0);
-        assert(!compact_info.clean_wr);
-        assert(!compact_info.do_delete);
-        assert(small_writes == 0);
-
-        // When under pressure: MUST ALSO postpone compaction because obj is STABLE!
-        // Before the fix, this would incorrectly compact l2 (compact_lsn == 2)
-        compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), true, [&](heap_entry_t *wr)
-        {
-            small_writes++;
-        });
-        assert(compact_info.compact_lsn == 0);
-        assert(compact_info.compact_version == 0);
-        assert(!compact_info.clean_wr);
-        assert(!compact_info.do_delete);
-        assert(small_writes == 0);
-
-        // Verify that add_compact() rejects stable overwrites with EBUSY
-        uint32_t compact_mblock = 999999;
-        uint8_t ref_int_bitmap[dsk.clean_entry_bitmap_size];
-        memset(ref_int_bitmap, 0, dsk.clean_entry_bitmap_size);
-        res = heap.add_compact(obj, 2, 2, 0, false, &compact_mblock, ref_int_bitmap, ref_int_bitmap, NULL);
-        assert(res == EBUSY);
-
-        heap.complete_block_write(mblock);
-        assert(heap.get_fsynced_lsn() == 3);
-
-        // Once fsynced, l3 is durable and becomes the clean_wr base
-        compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
-        {
-            small_writes++;
-        });
-        assert(compact_info.compact_lsn == 0);
-        assert(compact_info.clean_wr && compact_info.clean_wr->lsn == 3);
-        assert(small_writes == 0);
-    }
-
-    {
-        blockstore_heap_t heap(&dsk, buffer_area.data());
-        heap.finish_recheck();
-
-        // Case: BIG_STABLE(v1 l1) SMALL_STABLE(v2 l2) BIG_INTENT(v3 l3) un-fsynced
-        // -> skip compaction of l2 into l1 both when not under pressure and under pressure
-        // Before the fix, BIG_INTENT was missing from the check and was never skipped.
-        _test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 4096, buffer_area.data());
-        _test_small_write(heap, dsk, 1, 0, 2, 0, 4096, 0, true, buffer_area.data(), false);
-
-        object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
-        auto obj = heap.read_entry(oid);
-        uint8_t ext_bitmap[dsk.clean_entry_bitmap_size];
-        memset(ext_bitmap, 0x8e, dsk.clean_entry_bitmap_size);
-        heap.use_data(INODE_WITH_POOL(1, 1), 128*1024);
-        uint32_t mblock = 999999;
-        res = heap.add_redirect_intent(oid, &obj, 3, 0, dsk.data_block_size, 128*1024, ext_bitmap, buffer_area.data(), &mblock);
-        assert(res == 0);
-        heap.start_block_write(mblock);
-
-        obj = heap.read_entry(oid);
-        assert(obj->lsn == 3);
-        assert(obj->type() == BS_HEAP_BIG_INTENT);
-        assert(obj->is_overwrite());
-        assert(heap.get_fsynced_lsn() == 2);
-
-        // When not under pressure: MUST postpone compaction!
-        // Before the fix, BIG_INTENT was not checked, so it incorrectly compacted l2 (compact_lsn == 2)
-        int small_writes = 0;
-        auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
-        {
-            small_writes++;
-        });
-        assert(compact_info.compact_lsn == 0);
-        assert(compact_info.compact_version == 0);
-        assert(!compact_info.clean_wr);
-        assert(!compact_info.do_delete);
-        assert(small_writes == 0);
-
-        // When under pressure: MUST ALSO postpone compaction!
-        compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), true, [&](heap_entry_t *wr)
-        {
-            small_writes++;
-        });
-        assert(compact_info.compact_lsn == 0);
-        assert(compact_info.compact_version == 0);
-        assert(!compact_info.clean_wr);
-        assert(!compact_info.do_delete);
-        assert(small_writes == 0);
-
-        // Verify add_compact() rejects stable overwrite with EBUSY
-        uint32_t compact_mblock = 999999;
-        uint8_t ref_int_bitmap[dsk.clean_entry_bitmap_size];
-        memset(ref_int_bitmap, 0, dsk.clean_entry_bitmap_size);
-        res = heap.add_compact(obj, 2, 2, 0, false, &compact_mblock, ref_int_bitmap, ref_int_bitmap, NULL);
-        assert(res == EBUSY);
-
-        // Complete block write
-        heap.complete_block_write(mblock);
-        assert(heap.get_fsynced_lsn() == 3);
-
-        // Once fsynced, BIG_INTENT is clean_wr base
-        compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
-        {
-            small_writes++;
-        });
-        assert(compact_info.compact_lsn == 0);
-        assert(compact_info.clean_wr && compact_info.clean_wr->lsn == 3);
-        assert(small_writes == 0);
-    }
-
-    {
-        blockstore_heap_t heap(&dsk, buffer_area.data());
-        heap.finish_recheck();
-
-        // Case: BIG_STABLE(v1 l1) SMALL_STABLE(v2 l2) DELETE_STABLE(l3) un-fsynced
-        // -> skip compaction of l2 into l1 EVEN WHEN under pressure,
-        // because add_compact rejects stable overwrites with EBUSY
-        _test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 4096, buffer_area.data());
-        _test_small_write(heap, dsk, 1, 0, 2, 0, 4096, 0, true, buffer_area.data(), false);
-
-        object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
-        auto obj = heap.read_entry(oid);
-        uint32_t mblock = 999999;
-        res = heap.add_delete(obj, &mblock);
-        assert(res == 0);
-        heap.start_block_write(mblock);
-
-        obj = heap.read_entry(oid);
-        assert(obj->lsn == 3);
-        assert(obj->type() == BS_HEAP_DELETE);
-        assert(obj->is_overwrite());
-        assert(heap.get_fsynced_lsn() == 2);
-
-        // When not under pressure: postpone compaction
-        int small_writes = 0;
-        auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
-        {
-            small_writes++;
-        });
-        assert(compact_info.compact_lsn == 0);
-        assert(compact_info.compact_version == 0);
-        assert(!compact_info.clean_wr);
-        assert(!compact_info.do_delete);
-        assert(small_writes == 0);
-
-        // When under pressure: MUST ALSO postpone compaction because DELETE is STABLE!
-        // Before the fix, this would incorrectly compact l2 (compact_lsn == 2)
-        compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), true, [&](heap_entry_t *wr)
-        {
-            small_writes++;
-        });
-        assert(compact_info.compact_lsn == 0);
-        assert(compact_info.compact_version == 0);
-        assert(!compact_info.clean_wr);
-        assert(!compact_info.do_delete);
-        assert(small_writes == 0);
-
-        // Verify add_compact() rejects stable overwrite with EBUSY
-        uint32_t compact_mblock = 999999;
-        uint8_t ref_int_bitmap[dsk.clean_entry_bitmap_size];
-        memset(ref_int_bitmap, 0, dsk.clean_entry_bitmap_size);
-        res = heap.add_compact(obj, 2, 2, 0, false, &compact_mblock, ref_int_bitmap, ref_int_bitmap, NULL);
-        assert(res == EBUSY);
-
-        // Complete block write
-        heap.complete_block_write(mblock);
-        assert(heap.get_fsynced_lsn() == 3);
-
-        // Once fsynced, DELETE is processed
-        compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
-        {
-            small_writes++;
-        });
-        assert(compact_info.do_delete);
-        assert(!compact_info.clean_wr);
-        assert(small_writes == 0);
-    }
-
    {
        blockstore_heap_t heap(&dsk, buffer_area.data());
        heap.finish_recheck();
--- a/src/test/test_messenger.cpp
+++ b/src/test/test_messenger.cpp
@@ -0,0 +1,302 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
+//
+// Tests for the messenger receive path, in particular for reading and
+// discarding replies to canceled operations (osd_op_t::ignore_reply)
+
+#undef NDEBUG
+#include <assert.h>
+#include <stdio.h>
+#include <vector>
+
+#include "messenger.h"
+#include "util/xxh_x86dispatch.h"
+
+#define BMP_LEN 8
+#define DATA_LEN 4096
+
+// The receive path is protected, so reach it through a subclass
+struct test_msgr_t: osd_messenger_t
+{
+    bool feed(osd_client_t *cl, uint8_t *buf, size_t len) { return handle_read_buffer(cl, buf, len); }
+    void drain() { handle_immediate_ops(); }
+    osd_op_t *make_stub(osd_op_t *op) { return copy_out_op_to_ignore(op); }
+    size_t serialize(osd_client_t *cl, uint8_t *dst, size_t len) { return copy_ops_to(cl, dst, len); }
+    size_t immediate_count() { return set_immediate_ops.size(); }
+};
+
+static void append_csum(std::vector<uint8_t> & buf, int csum_status, size_t payload)
+{
+    if (csum_status == MSGR_CSUM_FULL)
+    {
+        // The header is covered by the checksum too
+        uint64_t csum = XXH3_64bits(buf.data(), buf.size());
+        buf.resize(buf.size()+8);
+        memcpy(buf.data()+buf.size()-8, &csum, 8);
+    }
+    else if (csum_status == MSGR_CSUM_PAYLOAD)
+    {
+        uint64_t csum = XXH3_64bits(buf.data()+OSD_PACKET_SIZE, payload);
+        buf.resize(buf.size()+8);
+        memcpy(buf.data()+buf.size()-8, &csum, 8);
+    }
+}
+
+// SEC_READ reply: header + bitmap + data
+static std::vector<uint8_t> make_read_reply(uint64_t id, uint8_t fill, int csum_status = 0)
+{
+    std::vector<uint8_t> buf(OSD_PACKET_SIZE + BMP_LEN + DATA_LEN);
+    osd_any_reply_t *r = (osd_any_reply_t*)buf.data();
+    memset(buf.data(), 0, OSD_PACKET_SIZE);
+    r->hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
+    r->hdr.id = id;
+    r->hdr.opcode = OSD_OP_SEC_READ;
+    r->hdr.retval = DATA_LEN;
+    r->sec_rw.attr_len = BMP_LEN;
+    memset(buf.data()+OSD_PACKET_SIZE, 0xB0|fill, BMP_LEN);
+    memset(buf.data()+OSD_PACKET_SIZE+BMP_LEN, fill, DATA_LEN);
+    append_csum(buf, csum_status, BMP_LEN+DATA_LEN);
+    return buf;
+}
+
+// SEC_LIST reply - a single payload region, unlike reads
+static std::vector<uint8_t> make_list_reply(uint64_t id, uint64_t count, int csum_status = 0)
+{
+    size_t payload = sizeof(obj_ver_id)*count;
+    std::vector<uint8_t> buf(OSD_PACKET_SIZE + payload);
+    osd_any_reply_t *r = (osd_any_reply_t*)buf.data();
+    memset(buf.data(), 0, OSD_PACKET_SIZE);
+    r->hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
+    r->hdr.id = id;
+    r->hdr.opcode = OSD_OP_SEC_LIST;
+    r->hdr.retval = count;
+    memset(buf.data()+OSD_PACKET_SIZE, 0x5A, payload);
+    append_csum(buf, csum_status, payload);
+    return buf;
+}
+
+static osd_op_t *make_read_op(uint64_t id)
+{
+    osd_op_t *op = new osd_op_t;
+    op->op_type = OSD_OP_OUT;
+    op->req.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
+    op->req.hdr.id = id;
+    op->req.hdr.opcode = OSD_OP_SEC_READ;
+    op->req.sec_rw.len = DATA_LEN;
+    op->bitmap_len = BMP_LEN;
+    return op;
+}
+
+// A read operation with preallocated buffers, like cluster_client and the OSD make them
+static osd_op_t *make_real_read_op(uint64_t id, int *done, uint8_t **out_data)
+{
+    osd_op_t *op = make_read_op(id);
+    op->buf = malloc_or_die(BMP_LEN + DATA_LEN);
+    op->iov.push_back(op->buf, DATA_LEN);
+    op->bitmap = (uint8_t*)op->buf + DATA_LEN;
+    *out_data = (uint8_t*)op->buf;
+    op->callback = [done](osd_op_t *op) { *done = op->reply.hdr.retval; };
+    return op;
+}
+
+// Put a stub in place of a canceled operation, like cancel_out() does
+static void cancel_op(test_msgr_t *msgr, osd_client_t *cl, osd_op_t *op)
+{
+    osd_op_t *stub = msgr->make_stub(op);
+    assert(stub->ignore_reply);
+    assert(stub->op_type == OSD_OP_OUT);
+    // No data buffers are allocated for a discarded reply
+    assert(!stub->buf && !stub->iov.count && !stub->bitmap);
+    cl->sent_ops[op->req.hdr.id] = stub;
+    delete op;
+}
+
+static osd_client_t *add_client(test_msgr_t *msgr, int csum_status)
+{
+    auto cl = new osd_client_t();
+    cl->client_id = msgr->next_client_id++;
+    cl->peer_fd = -1;
+    cl->peer_state = PEER_CONNECTED;
+    cl->proto_csum_status = csum_status;
+    msgr->clients[cl->client_id] = cl;
+    return cl;
+}
+
+// A canceled reply is read and thrown away, and the stub is freed
+static void test_discard_one_chunk(test_msgr_t *msgr)
+{
+    auto cl = add_client(msgr, 0);
+    cancel_op(msgr, cl, make_read_op(1));
+    auto buf = make_read_reply(1, 0x11);
+    assert(msgr->feed(cl, buf.data(), buf.size()));
+    assert(cl->sent_ops.size() == 0);
+    assert(msgr->immediate_count() == 1);
+    msgr->drain();
+    msgr->stop_client(cl->client_id);
+    printf("[ok] discarded reply, single chunk\n");
+}
+
+// The same, but split into small buffers - checks that the skip resumes correctly
+static void test_discard_chunked(test_msgr_t *msgr)
+{
+    auto cl = add_client(msgr, 0);
+    cancel_op(msgr, cl, make_read_op(1));
+    auto buf = make_read_reply(1, 0x22);
+    size_t chunk = 100;
+    for (size_t pos = 0; pos < buf.size(); pos += chunk)
+    {
+        size_t n = buf.size()-pos < chunk ? buf.size()-pos : chunk;
+        assert(msgr->feed(cl, buf.data()+pos, n));
+    }
+    assert(cl->sent_ops.size() == 0);
+    msgr->drain();
+    msgr->stop_client(cl->client_id);
+    printf("[ok] discarded reply, %zu-byte chunks\n", chunk);
+}
+
+// The important one: if the discard consumes the wrong number of bytes,
+// the next reply in the same buffer is garbage
+static void test_stream_stays_in_sync(test_msgr_t *msgr, int csum_status)
+{
+    auto cl = add_client(msgr, csum_status);
+    cancel_op(msgr, cl, make_read_op(1));
+    int done = 0;
+    uint8_t *data = NULL;
+    osd_op_t *real = make_real_read_op(2, &done, &data);
+    cl->sent_ops[2] = real;
+    auto b1 = make_read_reply(1, 0x33, csum_status);
+    auto b2 = make_read_reply(2, 0x44, csum_status);
+    std::vector<uint8_t> both(b1);
+    both.insert(both.end(), b2.begin(), b2.end());
+    assert(msgr->feed(cl, both.data(), both.size()));
+    msgr->drain();
+    assert(!cl->io_error);
+    assert(cl->sent_ops.size() == 0);
+    assert(done == DATA_LEN);
+    for (int i = 0; i < DATA_LEN; i++)
+        assert(data[i] == 0x44);
+    assert(((uint8_t*)real->bitmap)[0] == (0xB0|0x44));
+    delete real;
+    msgr->stop_client(cl->client_id);
+    printf("[ok] stream stays in sync after a discarded reply, csum_status=%d\n", csum_status);
+}
+
+// SEC_LIST goes through the single-payload-region branch of the discard path
+static void test_discard_list(test_msgr_t *msgr, int csum_status)
+{
+    auto cl = add_client(msgr, csum_status);
+    osd_op_t *op = new osd_op_t;
+    op->op_type = OSD_OP_OUT;
+    op->req.hdr.id = 1;
+    op->req.hdr.opcode = OSD_OP_SEC_LIST;
+    cancel_op(msgr, cl, op);
+    int done = 0;
+    uint8_t *data = NULL;
+    osd_op_t *real = make_real_read_op(2, &done, &data);
+    cl->sent_ops[2] = real;
+    auto b1 = make_list_reply(1, 37, csum_status);
+    auto b2 = make_read_reply(2, 0x77, csum_status);
+    std::vector<uint8_t> both(b1);
+    both.insert(both.end(), b2.begin(), b2.end());
+    assert(msgr->feed(cl, both.data(), both.size()));
+    msgr->drain();
+    assert(!cl->io_error);
+    assert(cl->sent_ops.size() == 0);
+    assert(done == DATA_LEN);
+    for (int i = 0; i < DATA_LEN; i++)
+        assert(data[i] == 0x77);
+    delete real;
+    msgr->stop_client(cl->client_id);
+    printf("[ok] discarded SEC_LIST + real reply, csum_status=%d\n", csum_status);
+}
+
+// The data is thrown away, but it's still checksummed on the way
+static void test_discard_checks_csum(test_msgr_t *msgr, int csum_status)
+{
+    auto cl = add_client(msgr, csum_status);
+    uint64_t client_id = cl->client_id;
+    cancel_op(msgr, cl, make_read_op(1));
+    auto buf = make_read_reply(1, 0x88, csum_status);
+    buf[OSD_PACKET_SIZE+BMP_LEN+10] ^= 0xff;
+    printf("-- a checksum mismatch message is expected below:\n");
+    msgr->feed(cl, buf.data(), buf.size());
+    // A checksum mismatch stops the client, so it has to be gone by now
+    assert(msgr->clients.find(client_id) == msgr->clients.end());
+    printf("[ok] corruption inside a discarded reply is still detected, csum_status=%d\n", csum_status);
+}
+
+// Build a SEC_READ reply the way an OSD does, for the real send path
+static osd_op_t *make_reply_op(uint64_t id, uint8_t fill)
+{
+    osd_op_t *op = new osd_op_t;
+    op->op_type = OSD_OP_IN;
+    op->req.hdr.id = id;
+    op->req.hdr.opcode = OSD_OP_SEC_READ;
+    op->req.sec_rw.len = DATA_LEN;
+    op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
+    op->reply.hdr.id = id;
+    op->reply.hdr.opcode = OSD_OP_SEC_READ;
+    op->reply.hdr.retval = DATA_LEN;
+    op->reply.sec_rw.attr_len = BMP_LEN;
+    op->buf = malloc_or_die(BMP_LEN + DATA_LEN);
+    memset(op->buf, fill, DATA_LEN);
+    op->bitmap = (uint8_t*)op->buf + DATA_LEN;
+    memset(op->bitmap, 0xB0|fill, BMP_LEN);
+    op->iov.push_back(op->buf, DATA_LEN);
+    return op;
+}
+
+// Serialize two replies through the real send path and feed them into the real receive
+// path, discarding the first one. Unlike the tests above, the checksum is produced by
+// the messenger itself, so this also verifies that the sender and the receiver agree
+static void test_roundtrip(test_msgr_t *msgr, int csum_status)
+{
+    auto scl = add_client(msgr, csum_status);
+    scl->write_ops.push_back(make_reply_op(1, 0x11));
+    scl->write_ops.push_back(make_reply_op(2, 0x22));
+    std::vector<uint8_t> wire(2*(OSD_PACKET_SIZE + BMP_LEN + DATA_LEN + 8));
+    size_t n = msgr->serialize(scl, wire.data(), wire.size());
+    assert(!scl->write_op && !scl->write_ops.size());
+    wire.resize(n);
+
+    auto rcl = add_client(msgr, csum_status);
+    cancel_op(msgr, rcl, make_read_op(1));
+    int done = 0;
+    uint8_t *data = NULL;
+    osd_op_t *real = make_real_read_op(2, &done, &data);
+    rcl->sent_ops[2] = real;
+    assert(msgr->feed(rcl, wire.data(), wire.size()));
+    msgr->drain();
+    assert(!rcl->io_error);
+    assert(rcl->sent_ops.size() == 0);
+    assert(done == DATA_LEN);
+    for (int i = 0; i < DATA_LEN; i++)
+        assert(data[i] == 0x22);
+    assert(((uint8_t*)real->bitmap)[0] == (0xB0|0x22));
+    delete real;
+    msgr->stop_client(rcl->client_id);
+    msgr->stop_client(scl->client_id);
+    printf("[ok] send->receive round trip with a discarded reply, csum_status=%d\n", csum_status);
+}
+
+int main()
+{
+    setvbuf(stdout, NULL, _IONBF, 0);
+    auto msgr = new test_msgr_t();
+    test_discard_one_chunk(msgr);
+    test_discard_chunked(msgr);
+    test_stream_stays_in_sync(msgr, 0);
+    test_stream_stays_in_sync(msgr, MSGR_CSUM_PAYLOAD);
+    test_stream_stays_in_sync(msgr, MSGR_CSUM_FULL);
+    test_discard_list(msgr, 0);
+    test_discard_list(msgr, MSGR_CSUM_PAYLOAD);
+    test_discard_list(msgr, MSGR_CSUM_FULL);
+    test_discard_checks_csum(msgr, MSGR_CSUM_PAYLOAD);
+    test_discard_checks_csum(msgr, MSGR_CSUM_FULL);
+    test_roundtrip(msgr, 0);
+    test_roundtrip(msgr, MSGR_CSUM_PAYLOAD);
+    test_roundtrip(msgr, MSGR_CSUM_FULL);
+    delete msgr;
+    printf("All messenger tests passed\n");
+    return 0;
+}
--- a/src/test/test_msgr_kdf.cpp
+++ b/src/test/test_msgr_kdf.cpp
@@ -1,122 +0,0 @@
-// Copyright (c) Alexey Shabalin, 2026+
-// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
-
-// Known-answer test for msgr_handshake_ctx_i::derive_kdf (HKDF-SHA384).
-// Keys must be the same with any OpenSSL version, and the salt used for key rotation
-// must not affect the next derive_kdf call without salt.
-
-#include <stdio.h>
-#include <stdlib.h>
-#include <stdint.h>
-
-#include <string>
-#include <vector>
-
-#include "msgr_handshake.h"
-
-// Self-signed EC P-384 certificate, only used to initialize the handshake context
-static const char *test_key =
-    "-----BEGIN PRIVATE KEY-----\n"
-    "MIG2AgEAMBAGByqGSM49AgEGBSuBBAAiBIGeMIGbAgEBBDA0CIkBTlNDaxTDbpc3\n"
-    "2JBY4ofK7PfLDw4D2jEEQ/kTQRAXOobqFCjSqefm2Bbmz9yhZANiAAQZEaeZPB3z\n"
-    "60xhhxxGYn2/qUgRXcoyPkGBzBXT54esK/qcuumFo1/uoB2+PTNT9lRqYO9s4xhC\n"
-    "naDV653Ynp6J8O0AnyhzqDaxHIjG84yWlslTWZMyDvqAtp1ZsIPprSk=\n"
-    "-----END PRIVATE KEY-----\n";
-
-static const char *test_cert =
-    "-----BEGIN CERTIFICATE-----\n"
-    "MIIBwjCCAUegAwIBAgIUMTD1mUZ7KJZYC2m28V+S3Y3M2wkwCgYIKoZIzj0EAwIw\n"
-    "GDEWMBQGA1UEAwwNdml0YXN0b3ItdGVzdDAgFw0yNjA5MzAxODM3NDJaGA8yMTI2\n"
-    "MDkwNjE4Mzc0MlowGDEWMBQGA1UEAwwNdml0YXN0b3ItdGVzdDB2MBAGByqGSM49\n"
-    "AgEGBSuBBAAiA2IABBkRp5k8HfPrTGGHHEZifb+pSBFdyjI+QYHMFdPnh6wr+py6\n"
-    "6YWjX+6gHb49M1P2VGpg72zjGEKdoNXrndienonw7QCfKHOoNrEciMbzjJaWyVNZ\n"
-    "kzIO+oC2nVmwg+mtKaNQME4wHQYDVR0OBBYEFLkMxzaAAbRTajF+q6HbCWjJtDkP\n"
-    "MB8GA1UdIwQYMBaAFLkMxzaAAbRTajF+q6HbCWjJtDkPMAwGA1UdEwQFMAMBAf8w\n"
-    "CgYIKoZIzj0EAwIDaQAwZgIxAJ0uH7Y2LmXKnRHSgYLqmpeKFqsvmbph3oWEOErX\n"
-    "+NJK56FzMceivmjvMkt4WMa69wIxAO14TLCFZeESLltGwIzdM0hv/Iysiec88ip+\n"
-    "OlW2PuLyhw8f/4HHp7Ooyu0ki3AHEA==\n"
-    "-----END CERTIFICATE-----\n";
-
-static const char *hs_key_hex =
-    "f5ed89f2161b25f4ac4a097179c12008804f2c8cfe757a8d810e166720dc6f67"
-    "86f2852a19697bd869551b10";
-
-static const char *key_hex =
-    "2b4047bb889aa0e63c5ca73e2edfc8b29dca0887bed9058a390c341bfb8a677c"
-    "d632115d5342470a48c76c6094a1e27945dffead049d308c53e3bbaf9a4e71f1"
-    "5758e3b3414416f2c9f0ea59235f2a022ad836ff8d8b91ca926f5d5c07ac5dc8"
-    "57f3ed36b1cd085b26c2471aa55c314804efeb9da022fd7a25c035cfdb773eb2"
-    "75f10d6f2f1551bc53dca1d34b44ba602c3b2e975ddd068732fa5d40ac0302f9"
-    "408f357361dd8ba6de9cfb42114663ae3bec06b94fce87227283e28464f27468"
-    "1677d78682b142e17fcdfaf6ac24eca4c31b11f3fde6186d5e1850fda7ec6747"
-    "7fb2d1e6b1d3998b38373904";
-
-static const char *rotated_key_hex =
-    "33b3f43c75660d051ef7465b4e9f61e74c3715af5bb3730a0093931874b0c2f2"
-    "73363984f3ae5e7cc6023200bfd3bdff0ee55d6ee10feb78396b0f74663c59f6"
-    "394a861c9f1f86b29b5d5fd491e196f05f18576592c86c7c0713ec9fad9e8053"
-    "a9113ba72ad56b725e57bf887891abb56fcac346b8f5eb30cc58c255b904ee88"
-    "2e7db934efadd32a76e235be842c4a647bcfe4d96a1d11b68d793d7356261bc0"
-    "0b10f345a456674855cc6621440a94201dfef0ee724d0913c24f74c3e4f21faf"
-    "a263f16f512a84103aeb50507466a2df51c2f35f9c49a27c6697d031a80dcd27"
-    "78959c188de53c8a4d757260";
-
-static void fill(uint8_t *buf, size_t len, uint8_t start)
-{
-    for (size_t i = 0; i < len; i++)
-        buf[i] = (uint8_t)(start + i);
-}
-
-static std::string to_hex(const uint8_t *buf, size_t len)
-{
-    std::string r;
-    char tmp[3];
-    for (size_t i = 0; i < len; i++)
-    {
-        snprintf(tmp, sizeof(tmp), "%02x", buf[i]);
-        r += tmp;
-    }
-    return r;
-}
-
-static void check(msgr_handshake_ctx_i *ctx, const char *name, const uint8_t *secret, size_t secret_len,
-    const uint8_t *salt, size_t salt_len, const char *label, size_t out_len, const char *expected)
-{
-    std::vector<uint8_t> out(out_len);
-    if (!ctx->derive_kdf(secret, secret_len, salt, salt_len, label, out.data(), out_len))
-    {
-        printf("%s: derive_kdf failed\n", name);
-        exit(1);
-    }
-    std::string actual = to_hex(out.data(), out_len);
-    if (actual != expected)
-    {
-        printf("%s: key mismatch\nexpected: %s\nactual:   %s\n", name, expected, actual.c_str());
-        exit(1);
-    }
-}
-
-int main(int narg, char *args[])
-{
-    // Same sizes as in msgr_handshake_t::derive_hs_keys() and osd_messenger_t::derive_aes_keys()
-    const size_t hs_key_len = AES_256_GCM_KEY_SIZE + AES_256_GCM_IV_SIZE;
-    const size_t key_len = AES_256_GCM_KEY_SIZE + AES_256_GCM_IV_SIZE + 192 /* XXH_SECRET_DEFAULT_SIZE */;
-    uint8_t secret[48], old_key[key_len];
-    fill(secret, sizeof(secret), 0x00);
-    fill(old_key, sizeof(old_key), 0x80);
-    msgr_handshake_ctx_i *ctx = msgr_handshake_ctx_i::create_ctx();
-    if (!ctx->init(test_cert, test_key, test_cert, "", ""))
-    {
-        printf("handshake context init failed: %s\n", ctx->get_error().c_str());
-        exit(1);
-    }
-    check(ctx, "handshake key", secret, sizeof(secret), NULL, 0, "server hs key", hs_key_len, hs_key_hex);
-    check(ctx, "initial key", secret, sizeof(secret), NULL, 0, "client key", key_len, key_hex);
-    // Key rotation uses the old key as salt
-    check(ctx, "rotated key", secret, sizeof(secret), old_key, sizeof(old_key), "client key", key_len, rotated_key_hex);
-    // Salt of the previous call must not affect the next call without salt
-    check(ctx, "handshake key after rotation", secret, sizeof(secret), NULL, 0, "server hs key", hs_key_len, hs_key_hex);
-    delete ctx;
-    printf("OK\n");
-    return 0;
-}
Author	SHA1	Message	Date
Vitaliy Filippov	1d773779d4	Implement actual client operation cancel	2026-09-25 18:07:09 +03:00
Vitaliy Filippov	ee8b63d971	Add a test for operation cancel	2026-09-24 11:55:52 +03:00
Vitaliy Filippov	0d28c889d8	Throw replies away without allocating buffers(?)	2026-09-24 11:55:51 +03:00
Vitaliy Filippov	67131e5fbc	Support canceling outbound operations in messenger	2026-09-24 11:55:00 +03:00