WIP Replace exit/abort in client with on_fatal_error_hook

Change cluster_client_t::msgr to unique_ptr
Fix QEMU driver build for idiotic RHEL backport into 9.1
2026-10-02 12:42:45 +03:00 · 2026-10-02 12:42:45 +03:00 · 2026-10-01 11:13:10 +03:00 · 2026-10-01 00:50:44 +03:00 · 2026-10-01 00:50:31 +03:00 · 2026-09-29 01:40:00 +03:00
44 changed files with 686 additions and 940 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.1")
+set(VITASTOR_VERSION "3.2.2")

 include(CTest)

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

 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.1
+          image: vitalif/vitastor-csi:v3.2.2
          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.1
+          image: k8s.gcr.io/sig-storage/csi-attacher:v3.2.2
          args:
            - "--v=5"
            - "--csi-address=$(ADDRESS)"
@@ -121,7 +121,7 @@ spec:
            privileged: true
            capabilities:
              add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v3.2.1
+          image: vitalif/vitastor-csi:v3.2.2
          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.1"
+    vitastorCSIDriverVersion = "3.2.2"
 )

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

  * Bugfixes

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

 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.1
+VITASTOR_VERSION=v3.2.2

 # 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.1`
+   `docker pull vitalif/vitastor:v3.2.2`
 2. Install scripts to the host system: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.1 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.2 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.1`
+   `docker pull vitalif/vitastor:v3.2.2`
 2. Установите скрипты в хост-систему командой: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.1 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.2 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.1",
+  "version": "3.2.2",
  "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.1",
+  "version": "3.2.2",
  "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.1'
+VITASTOR_VERSION = '3.2.2'

 LOG = logging.getLogger(__name__)

--- a/rpm/vitastor-el10.spec
+++ b/rpm/vitastor-el10.spec
@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.2.1
+Version:        3.2.2
 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.1.el10.tar.gz
+Source0:        vitastor-3.2.2.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.1
+Version:        3.2.2
 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.1.el7.tar.gz
+Source0:        vitastor-3.2.2.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.1
+Version:        3.2.2
 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.1.el8.tar.gz
+Source0:        vitastor-3.2.2.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.1
+Version:        3.2.2
 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.1.el9.tar.gz
+Source0:        vitastor-3.2.2.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.1")
+add_definitions(-DVITASTOR_VERSION="3.2.2")
 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,18 +213,25 @@ 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,7 +197,10 @@ uint32_t *heap_entry_t::get_checksum(blockstore_heap_t *heap)
 {
    if (type() == BS_HEAP_SMALL_WRITE || type() == BS_HEAP_INTENT_WRITE)
    {
-        if (heap->dsk->csum_block_size || small().len == 0)
+        // 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)
            return NULL;
        return (uint32_t*)((uint8_t*)this + sizeof(heap_small_write_t) + heap->dsk->clean_entry_bitmap_size);
    }
@@ -1263,6 +1266,12 @@ 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);
@@ -2436,9 +2445,11 @@ 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 && (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_DELETE))
+            if ((!under_pressure || stable) && (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_BIG_INTENT || 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,6 +14,11 @@

 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();
@@ -23,12 +28,13 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
    this->ringloop = ringloop;
    this->tfd = tfd;

-    msgr.osd_num = 0;
-    msgr.tfd = tfd;
-    msgr.ringloop = ringloop;
-    msgr.repeer_pgs = [this](osd_num_t peer_osd)
+    msgr = new osd_messenger_t();
+    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();
@@ -45,14 +51,16 @@ 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;
    };
-    msgr.parse_config(config, true);
+    auto exit_hook = msgr->on_fatal_error_hook;
+    msgr->on_fatal_error_hook = on_fatal_error_hook;
+    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); };
@@ -64,6 +72,7 @@ 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;
@@ -72,11 +81,30 @@ 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()
 {
-    vault_destroy();
+    aborted = true;
    if (retry_timeout_id >= 0)
    {
        tfd->clear_timer(retry_timeout_id);
@@ -89,13 +117,21 @@ 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()
@@ -150,13 +186,15 @@ 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++);
    }
 }
@@ -165,14 +203,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);
@@ -482,7 +520,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();
 }
@@ -643,9 +681,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();
    }
 }
@@ -761,7 +799,6 @@ 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;
@@ -861,115 +898,15 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
    }
 }

-// 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)
+void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
 {
-    if (op->canceled || (op->flags & OP_FLUSH_BUFFER))
+    if (aborted)
    {
-        // 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 true;
+        return;
    }
-    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
@@ -986,7 +923,14 @@ 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)
        {
-            execute_cas(op, true);
+            if (sync_op->retval != 0)
+            {
+                op->retval = sync_op->retval;
+                auto cb = std::move(op->callback);
+                cb(op);
+            }
+            else
+                execute_cas(op, true);
            delete sync_op;
        };
        execute_internal(sync_op);
@@ -1002,18 +946,25 @@ 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 (!)
@@ -1045,7 +996,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)
@@ -1176,33 +1127,25 @@ 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 (op->canceled)
-    {
-        if (op->inflight_count > 0)
-        {
-            // Wait for the remaining parts to be canceled
-            return 0;
-        }
-        finish_canceled(op);
-        return 1;
-    }
+    if (aborted)
+        return 0;
    if (op->state == 0)
        goto resume_0;
    else if (op->state == 1)
@@ -1627,8 +1570,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;
@@ -1672,12 +1615,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;
        }
    }
@@ -1686,16 +1629,8 @@ 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 (op->canceled)
-    {
-        if (op->inflight_count > 0)
-        {
-            // Wait for the remaining parts to be canceled
-            return 0;
-        }
-        finish_canceled(op);
-        return 1;
-    }
+    if (aborted)
+        return 0;
    if (op->state == 1)
        goto resume_1;
    if (!dirty_osds.size())
@@ -1709,8 +1644,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++;
@@ -1763,7 +1698,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){
@@ -1780,7 +1715,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)
@@ -1804,8 +1739,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
        if (ERR_PRIO(part->op.reply.hdr.retval) > ERR_PRIO(op->retval))
            op->retval = part->op.reply.hdr.retval;
        uint64_t stop_client_id = 0;
-        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC &&
-            op->retval != -EPERM && op->retval != -EAGAIN && op->retval != -ECANCELED)
+        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC && op->retval != -EPERM && op->retval != -EAGAIN)
        {
            stop_client_id = part->op.client_id;
            if (op->retval != -EPIPE || log_level > 0)
@@ -1836,7 +1770,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,7 +65,6 @@ 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 = {};
@@ -177,11 +176,12 @@ 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,6 +203,7 @@ 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);
@@ -234,7 +235,6 @@ 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)
        {
-            fprintf(stderr, "Failed to initialize HTTP context for Vault: %s\n", error.c_str());
-            exit(1);
+            fatal_error("Failed to initialize HTTP context for Vault: "+error);
+            return;
        }
        vault_http_cli = http_init(vault_http_ctx);
    }
--- a/src/client/etcd_state_client.cpp
+++ b/src/client/etcd_state_client.cpp
@@ -10,6 +10,20 @@
 #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,6 +144,7 @@ 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();
@@ -192,6 +193,7 @@ 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);
@@ -215,6 +217,7 @@ 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,7 +49,11 @@ void etcd_state_client_http_t::etcd_add_watch(json11::Json watch)

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

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

@@ -159,8 +170,12 @@ void etcd_state_client_http_t::pick_next_etcd(std::function<void()> cb)
 {
    if (!etcd_addresses.size() && !etcd_local.size())
    {
-        fprintf(stderr, "etcd_address is missing in Vitastor configuration\n");
-        exit(1);
+        fatal_error("etcd_address is missing in Vitastor configuration");
+        return;
+    }
+    if (!get_http_ctx())
+    {
+        return;
    }
    if (selected_etcd_url.addr != "")
    {
@@ -196,7 +211,7 @@ void etcd_state_client_http_t::pick_next_etcd(std::function<void()> cb)
    for (auto & url: etcd_name_urls)
    {
        resolve_count++;
-        http_resolve(get_http_ctx(), url.ssl, url.addr, [this, url](const std::string & error, const std::vector<std::string>& addresses)
+        http_resolve(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());
@@ -228,8 +243,8 @@ void etcd_state_client_http_t::pick_next_etcd_on_resolve()
 {
    if (!etcd_urls_to_try.size())
    {
-        fprintf(stderr, "None of etcd_address could be resolved\n");
-        exit(1);
+        fatal_error("None of etcd_address could be resolved");
+        return;
    }
    if (!rand_initialized)
    {
@@ -274,7 +289,11 @@ 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)
-        etcd_watch_ws = http_init(get_http_ctx());
+    {
+        if (!get_http_ctx())
+            return;
+        etcd_watch_ws = http_init(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 },
@@ -324,20 +343,17 @@ 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
                        {
-                            fprintf(stderr, "Revisions before %ju were compacted by etcd, exiting\n",
-                                data["result"]["compact_revision"].uint64_value());
-                            exit(1);
+                            fatal_error("Revisions before "+data["result"]["compact_revision"].as_string()+" were compacted by etcd, exiting");
                        }
                    }
                    else
                    {
-                        fprintf(stderr, "Watch canceled by etcd, reason: %s, exiting\n", data["result"]["cancel_reason"].string_value().c_str());
-                        exit(1);
+                        fatal_error("Watch canceled by etcd, reason: "+data["result"]["cancel_reason"].string_value()+", exiting");
                    }
+                    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"}}
@@ -492,9 +508,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();
@@ -502,7 +518,7 @@ void etcd_state_client_http_t::load_global_config()
            }
            else
            {
-                exit(1);
+                fatal_error("Error reading configuration from etcd: "+err);
            }
        }
    });
--- a/src/client/messenger.cpp
+++ b/src/client/messenger.cpp
@@ -17,6 +17,20 @@
 #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();
@@ -198,7 +212,7 @@ osd_messenger_t::~osd_messenger_t()
    destroy_tls();
 }

-static int parse_proto_checksums(const json11::Json & val, int default_value)
+int osd_messenger_t::parse_proto_checksums(const json11::Json & val, int default_value)
 {
    if (val.is_string())
    {
@@ -216,9 +230,9 @@ static int parse_proto_checksums(const json11::Json & val, int default_value)
    }
    else if (val.is_null())
        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);
+    fatal_error("proto_checksums should be \"full\", \"payload\", \"gcm\", \"none\""
+        ", \"\" or null (default), but it is: "+val.dump());
+    return default_value;
 }

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

 struct osd_client_t;

@@ -152,7 +151,6 @@ 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;
@@ -259,6 +257,7 @@ 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);
@@ -314,16 +313,17 @@ 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,6 +347,7 @@ 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);
@@ -355,7 +356,6 @@ 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,10 +597,9 @@ void osd_messenger_t::init_tls()
        if (client_cert.empty() || client_key.empty() || osd_ca.empty() || osd_num && client_ca.empty())
        {
            if (osd_num)
-                fprintf(stderr, "Vitastor transport encryption requires osd_cert, osd_pkey, osd_ca, client_ca options for OSDs\n");
+                fatal_error("Vitastor transport encryption requires osd_cert, osd_pkey, osd_ca, client_ca options for OSDs");
            else
-                fprintf(stderr, "Vitastor transport encryption requires cert, pkey and osd_ca options\n");
-            exit(1);
+                fatal_error("Vitastor transport encryption requires cert, pkey and osd_ca options\n");
        }
        else
        {
@@ -609,8 +608,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))
            {
-                fprintf(stderr, "Error: %s\n", hs_ctx->get_error().c_str());
-                exit(1);
+                gcm_enabled = false;
+                fatal_error("Error: "+hs_ctx->get_error());
            }
 #endif
        }
--- a/src/client/msgr_handshake.cpp
+++ b/src/client/msgr_handshake.cpp
@@ -205,6 +205,8 @@ 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;
 }

@@ -245,6 +247,9 @@ 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);
@@ -405,14 +410,20 @@ 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 (!ctx)
+    if (!enc_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: ");
@@ -438,7 +449,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 (!ctx)
+    if (!dec_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,17 +187,13 @@ 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(int retval);
+    void cancel();

    bool is_recovery_related();
    uint64_t calc_data_checksum();
--- a/src/client/msgr_receive.cpp
+++ b/src/client/msgr_receive.cpp
@@ -179,15 +179,14 @@ public:
        }
        else if (flags & RDR_GCM)
        {
-            // 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);
+            // Here, dst == NULL is not allowed
+            assert(dst != NULL);
 #ifdef WITH_ISAL_CRYPTO
-            int r = isal_aes_gcm_dec_256_update(&cl->peer_key_isal, cl->dec_ctx, out, curbuf+done, n);
+            int r = isal_aes_gcm_dec_256_update(&cl->peer_key_isal, cl->dec_ctx, dst+from, curbuf+done, n);
            assert(!r);
 #else
            int actual_out;
-            if (EVP_DecryptUpdate(cl->dec_ctx, out, &actual_out, curbuf+done, n) != 1)
+            if (EVP_DecryptUpdate(cl->dec_ctx, dst+from, &actual_out, curbuf+done, n) != 1)
            {
                fprintf(stderr, "DecryptUpdate error: ");
                ERR_print_errors_fp(stderr);
@@ -197,7 +196,7 @@ public:
 #endif
            if (cl->read_csum_state && !(flags & RDR_NO_CSUM))
            {
-                XXH3_64bits_update(cl->read_csum_state, out, n);
+                XXH3_64bits_update(cl->read_csum_state, dst+from, n);
            }
            done += n;
        }
@@ -308,8 +307,6 @@ 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)
@@ -400,8 +397,7 @@ void osd_messenger_t::read_requests()
            continue;
        }
        auto cl = cl_it->second;
-        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)
+        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)
        {
            get_op_reader_t rdr(this, cl);
            if (!op_read_from(cl, rdr))
@@ -574,15 +570,11 @@ void osd_messenger_t::handle_immediate_ops()
            else
                delete op;
        }
-        else if (op->callback)
+        else
        {
            // Copy lambda to be unaffected by `delete op`
            std::function<void(osd_op_t*)>(op->callback)(op);
        }
-        else
-        {
-            delete op;
-        }
    }
 }

@@ -867,12 +859,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->ignore_reply || op->bitmap);
+            assert(op->bitmap);
            cl->read_op_size += bmp_len;
        }
        if (op->reply.hdr.retval > 0)
        {
-            assert(op->ignore_reply || op->iov.count > 0);
+            assert(op->iov.count > 0);
            cl->read_op_size += op->reply.hdr.retval;
        }
    }
@@ -886,8 +878,7 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
                cl->client_id, op->reply.hdr.retval);
            return false;
        }
-        if (!op->ignore_reply)
-            op->buf = malloc_or_die(cl->read_op_size);
+        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)
    {
@@ -900,11 +891,8 @@ 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;
-        if (!op->ignore_reply)
-        {
-            free(op->buf);
-            op->buf = malloc_or_die(cl->read_op_size);
-        }
+        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)
    {
@@ -915,20 +903,14 @@ 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;
        }
-        if (!op->ignore_reply)
-        {
-            free(op->buf);
-            op->buf = malloc_or_die(op->reply.hdr.retval);
-        }
+        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;
-        if (!op->ignore_reply)
-        {
-            free(op->buf);
-            op->buf = malloc_or_die(op->reply.describe.result_bytes);
-        }
+        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)
@@ -992,37 +974,7 @@ 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->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.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(-EPIPE);
+        cur_op->cancel();
        return;
    }
    osd_client_t *cl = cl_it->second;
@@ -494,81 +494,6 @@ 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);
@@ -761,11 +686,7 @@ 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);
-        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 });
+        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) });
    }
    return wr.get_done();
 }
@@ -773,7 +694,6 @@ 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(-EPIPE);
+        op->cancel();
    }
 }

-void osd_op_t::cancel(int retval)
+void osd_op_t::cancel()
 {
    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 = retval;
+        reply.hdr.retval = -EPIPE;
        // Copy lambda to be unaffected by `delete this`
        (std::function<void(osd_op_t*)>(callback))(this);
    }
    else
    {
-        // It's fine to just free responses to incoming ops
+        // This function is only called in stop_client(), so it's fine to destroy the operation
        delete this;
    }
 }
@@ -57,8 +57,6 @@ 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);
        }
@@ -69,8 +67,6 @@ 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++;
@@ -208,21 +204,14 @@ 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(-EPIPE);
+        read_op->cancel();
        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
+#if QEMU_VERSION_MAJOR > 10 || QEMU_VERSION_MAJOR == 10 && QEMU_VERSION_MINOR >= 1 || defined BDRV_WANT_PRECISE
    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
+#if QEMU_VERSION_MAJOR > 10 || QEMU_VERSION_MAJOR == 10 && QEMU_VERSION_MINOR >= 1 || defined BDRV_WANT_PRECISE
    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.1
+Version: 3.2.2
 Libs: -L${libdir} -lvitastor_client
 Cflags: -I${includedir}

--- a/src/client/vitastor_c.cpp
+++ b/src/client/vitastor_c.cpp
@@ -30,6 +30,18 @@ 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" {
@@ -103,7 +115,16 @@ 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);
-    self->cli = cluster_client_t::create(NULL, self->tfd, cfg_json);
+    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;
+    }
    return self;
 }

@@ -111,47 +132,45 @@ 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);
-    self->ringloop = ringloop;
-    self->cli = cluster_client_t::create(self->ringloop, self->tfd, cfg_json);
-    ringloop->loop();
+    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;
+    }
    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;
-    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();
+    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;
+    }
    return self;
 }

@@ -166,15 +185,6 @@ 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)
    {
@@ -189,10 +199,18 @@ vitastor_c *vitastor_c_create_uring_json(const char **options, int options_len)
    }
    json11::Json cfg_json(cfg);
    vitastor_c *self = new vitastor_c;
-    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();
+    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;
+    }
    return self;
 }

@@ -205,8 +223,17 @@ vitastor_c *vitastor_c_create_epoll_json(const char **options, int options_len)
    }
    json11::Json cfg_json(cfg);
    vitastor_c *self = new vitastor_c;
-    self->epmgr = new epoll_manager_t(NULL);
-    self->cli = cluster_client_t::create(NULL, self->epmgr->tfd, cfg_json);
+    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;
+    }
    return self;
 }

@@ -217,13 +244,6 @@ 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,17 +27,6 @@ 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)
@@ -99,3 +88,9 @@ 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(-EPIPE);
+        cur_op->cancel();
        return;
    }
    auto cl = cl_it->second;
@@ -34,35 +34,6 @@ 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,9 +12,6 @@
 #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:
@@ -56,9 +53,6 @@ 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);
@@ -84,7 +78,6 @@ 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)
    {
@@ -120,7 +113,6 @@ 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;
 }
@@ -137,11 +129,10 @@ 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 : op->retval;
+        *r = op->retval == 0 ? 1 : 0;
        printf("Done sync r=%d\n", op->retval);
        delete op;
    };
-    last_posted_op = op;
    cli->execute(op);
    return r;
 }
@@ -186,7 +177,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);\
@@ -194,7 +185,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)\
    {\
@@ -205,7 +196,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())
    {
@@ -234,7 +225,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;
@@ -317,7 +308,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;
@@ -748,7 +739,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);
@@ -790,7 +781,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);
@@ -824,7 +815,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);
@@ -1131,127 +1122,6 @@ 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();
@@ -1265,6 +1135,5 @@ 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,6 +1010,211 @@ 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
@@ -1,302 +0,0 @@
-// 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
@@ -0,0 +1,122 @@
+// 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	0ea7b03aef	WIP Replace exit/abort in client with on_fatal_error_hook	2026-10-02 12:42:45 +03:00
Vitaliy Filippov	a43bf4924b	Change cluster_client_t::msgr to unique_ptr	2026-10-02 12:42:45 +03:00
Yury Luneff	1f2db0f95b	Fix QEMU driver build for idiotic RHEL backport into 9.1	2026-10-01 11:13:10 +03:00
Alexey Shabalin	c05f4769ff	Fix error handling in the transport encryption handshake - Check the result of EVP_KDF_CTX_new() in msgr_handshake_ctx_t::init(). - Check the created cipher context instead of msgr_handshake_t::ctx after EVP_CIPHER_CTX_new() in encrypt() and decrypt(). - Set the error message and the error state when the handshake signature is invalid, and clear the OpenSSL error queue so that its errors are not reported by the next on_error() call.	2026-10-01 00:50:44 +03:00
Alexey Shabalin	f1bf31f9db	Fix stale HKDF salt after transport encryption key rotation Some OpenSSL 3 versions ignore empty OSSL_KDF_PARAM_SALT in HKDF (checked: 3.0.0, 3.0.13, 3.0.14, 3.3.0, 3.3.3, 3.4.0, 3.5.0; 3.0.19 and 3.5.4 are fixed). So the shared kdf_ctx kept the salt set by the last key rotation (derive_aes_keys uses the old key as salt), and all following derive_kdf calls without salt, i.e. handshakes of all new connections, derived keys different from the peer's and failed with "Handshake decryption failed" until restart. Reset the context before each derive_kdf call. Add test_msgr_kdf with HKDF-SHA384 known-answer vectors, including a call without salt after a key rotation.	2026-10-01 00:50:31 +03:00
Vitaliy Filippov	f3e047fb05	Store v2: fill the checksum slot of zero-length small writes get_small_entry_size() always reserves a 4-byte data checksum slot when block checksums are off, and the entry-level checksum covers every byte of that size - but get_checksum() returned NULL for small/intent entries with len == 0, so calc_checksums() skipped the slot and the entry checksum was computed over whatever malloc leftover bytes happened to be there.	2026-09-29 01:40:00 +03:00
Vitaliy Filippov	df593badb5	Release 3.2.2 - New store (thanks Changwei Ye) - Fix rejected compaction when an object is already overwritten again - Reduce superblock I/O amplification introduced after previous correctness fixes - Fix possible `vitastor-cli rm` crash with `assert(rwo->todo == 1)` introduced in 3.2.1 - Fix replicated objects possibly incorrectly transitioning to the INCOMPLETE state after multiple failed partial writes - Fix possible "operation sequencing broken ... expected num 2, got 3, stopping client" possible when a cluster is overloaded by peering/OSD flaps	2026-09-27 14:29:43 +03:00
Changwei Ye	b51c05200d	Store v2: fix compaction skipping stable overwrites and BIG_INTENT type The original compaction condition only postponed compaction when not under pressure, and only for BIG_WRITE and DELETE types: if (!under_pressure && (BIG_WRITE \|\| DELETE)) This had two bugs: 1. BIG_INTENT overwrites were not included, causing add_compact() to be called on BIG_INTENT entries. add_compact() unconditionally rejects any stable overwrite with EBUSY, leading to unnecessary retry loops. 2. When under pressure, stable overwrites (BIG_WRITE, BIG_INTENT, DELETE with BS_HEAP_STABLE set) were not skipped. add_compact() rejects stable overwrites with EBUSY regardless of pressure, so attempting to compact them under pressure also always fails. Fix: change the condition to: if ((!under_pressure \|\| stable) && (BIG_WRITE \|\| BIG_INTENT \|\| DELETE)) When under pressure with an unstable overwrite, we now fall through to `continue`, skipping it and scanning older entries (SMALL_WRITE / INTENT_WRITE) that can actually be compacted to reclaim space.	2026-09-24 17:06:20 +08:00
Changwei Ye	6d1b9165a2	Store v2: decouple trim_lsn from force_fsync Commit `df9a2850` introduced force_fsync to resolve the placement contract conflict (EAGAIN) when a new metadata entry depends on a previous unsynced entry that is already in-flight. However, it mistakenly grouped the 'force_fsync > 0' condition together with the Superblock Checkpoint logic. As a result, every time the internal torn-write prevention mechanism requested a force_fsync, the system inadvertently executed an expensive trim_lsn() (updating the Superblock with RWF_DSYNC), causing unnecessary write latency and I/O amplification during congested object updates. This change minimally refactors the state machine to ensure trim_lsn() is strictly bound to the intended journal_trim_interval, while force_fsync only triggers the necessary fsync_buffer().	2026-09-24 11:08:15 +03:00