osd: account for key_index_bytes when calculating chained read bitmap_len

When OSD_OP_RETURN_CHAIN is set, the primary OSD appends chain info to
the returned bitmap buffer. The space required for chain info is
(len / bs_bitmap_granularity) * key_index_bytes, where key_index_bytes
depends on the snapshot chain depth (1 byte for <= 255 items, 2 bytes for
<= 65535, etc.).

Previously, cur_op->reply.rw.bitmap_len only added (len / bs_bitmap_granularity)
without multiplying by key_index_bytes. When the snapshot chain depth
exceeded 255 (key_index_bytes == 2), the reported bitmap_len was too short,
causing the client to receive truncated chain info.

Multiply by key_index_bytes when updating cur_op->reply.rw.bitmap_len, and
add test_chained_read_large_chain_info regression test in osd_test.

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)
 

csi/Makefile

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v3.2.1
+VITASTOR_VERSION ?= v3.2.2
 
 all: build push
 

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)"

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)"

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

debian/changelog

@@ -1,4 +1,4 @@
-vitastor (3.2.1-1) unstable; urgency=medium
+vitastor (3.2.2-1) unstable; urgency=medium
 
   * Bugfixes
 

docker/Makefile

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v3.2.1
+VITASTOR_VERSION ?= v3.2.2
 
 all: build push
 

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

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: \

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: \

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": {

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": [

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__)
 

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++

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++

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++

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++

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

src/blockstore/blockstore_flush.cpp

@@ -311,7 +311,8 @@ resume_1:
     }
     flusher->flushing[cur_oid] = compact_info.compact_lsn;
     flusher->active_flushers++;
-    if (!compact_info.do_delete && compact_info.clean_wr->type() == BS_HEAP_BIG_INTENT &&
+    if (!compact_info.do_delete && compact_info.clean_wr &&
+        compact_info.clean_wr->type() == BS_HEAP_BIG_INTENT &&
         (!bs->dsk.csum_block_size || bs->dsk.csum_block_size > bs->dsk.bitmap_granularity))
     {
         // Compaction merges newer writes into the data block in place, which changes bytes that
@@ -347,7 +348,13 @@ resume_28:
         {
             printf("Compacting %jx:%jx up to l%ju (delete)\n", cur_oid.inode, cur_oid.stripe, compact_info.compact_lsn);
         }
+        // Delete discards everything, so reset to the empty state
         clean_loc = UINT64_MAX;
+        read_vec.clear();
+        csum_copy.clear();
+        copy_count = 0;
+        memset(new_bmp, 0, bs->dsk.clean_entry_bitmap_size);
+        memset(new_ext_bmp, 0, bs->dsk.clean_entry_bitmap_size);
     }
     else
     {

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);
     }
@@ -300,6 +303,14 @@ blockstore_heap_t::~blockstore_heap_t()
 void blockstore_heap_t::start_load(uint64_t completed_lsn)
 {
     this->completed_lsn = completed_lsn;
+    // The persisted completed_lsn may be larger than the lsn of the last entry which survived:
+    // entries above it were still in flight when the power went away. New entries must never
+    // reuse an lsn which is at or below completed_lsn, or on the next load they'd look already
+    // fsynced and escape the rollback of unfinished writes
+    if (completed_lsn > next_lsn)
+    {
+        next_lsn = completed_lsn;
+    }
 }
 
 int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uint8_t *buf, bool allow_corrupted,
@@ -1263,6 +1274,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);
@@ -2058,7 +2075,7 @@ int blockstore_heap_t::add_commit(heap_entry_t *obj, uint64_t version, uint32_t
         {
             auto rollback_version = wr->version;
             wr = prev(wr);
-            while (wr->version > rollback_version)
+            while (wr && wr->version > rollback_version)
             {
                 assert(!(wr->entry_type & BS_HEAP_STABLE));
                 wr = prev(wr);
@@ -2334,6 +2351,17 @@ int blockstore_heap_t::get_next_compact(object_id & oid)
     return 0;
 }
 
+bool blockstore_heap_t::is_absent(heap_entry_t *obj)
+{
+    bool present = false;
+    iterate_with_stable(obj, UINT64_MAX, [&](heap_entry_t *wr, bool stable)
+    {
+        present = (wr->type() != BS_HEAP_DELETE);
+        return false;
+    });
+    return !present;
+}
+
 void blockstore_heap_t::iterate_with_stable(heap_entry_t *obj, uint64_t max_lsn, std::function<bool(heap_entry_t*, bool)> cb)
 {
     auto old_wr = obj;
@@ -2464,6 +2492,13 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
         {
             // Object is deleted
             assert(!has_small && stable); // unstable deletes are not supported
+            if (!res.do_delete && res.compact_lsn)
+            {
+                // A COMMIT above the delete reset do_delete, but all writes between
+                // them are lost (incorrect chain / no fsync). Restore do_delete.
+                res.do_delete = true;
+                res.compact_version = 0;
+            }
             return res;
         }
         assert(wr->type() == BS_HEAP_SMALL_WRITE || wr->type() == BS_HEAP_INTENT_WRITE);
@@ -2479,6 +2514,23 @@ heap_compact_t blockstore_heap_t::iterate_compaction(heap_entry_t *obj, uint64_t
         has_small = true;
         small_wr_cb(wr);
     }
+    if (!res.clean_wr && !res.do_delete && res.compact_lsn)
+    {
+        if (has_small)
+        {
+            // The chain holds no big entry surviving the rollbacks and no executed
+            // delete: neither a merge nor a delete is possible. The chain is invalid
+            // but we can't just remove it, so abort compaction and leave it as is.
+            res.compact_lsn = 0;
+        }
+        else
+        {
+            // Otherwise it means that there are no entries: it's either a bare COMMIT
+            // or a COMMIT over a ROLLBACK which masks all previous entries. Remove everything.
+            res.do_delete = true;
+        }
+        res.compact_version = 0;
+    }
     return res;
 }
 

src/blockstore/blockstore_heap.h

@@ -294,6 +294,9 @@ public:
     heap_entry_t *lock_and_read_entry(object_id oid);
     // read an object entry without locking it
     heap_entry_t *read_entry(object_id oid);
+    // true if a read of the chain would return ENOENT: the head is a DELETE entry
+    // or all entries are cut off by a ROLLBACK to version 0
+    bool is_absent(heap_entry_t *obj);
     // unlock an entry
     bool unlock_entry(object_id oid);
     // set or verify checksums in a write request

src/blockstore/blockstore_write.cpp

@@ -127,7 +127,8 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     if (op->opcode == BS_OP_DELETE)
     {
         // Delete
-        if (!obj || obj->type() == BS_HEAP_DELETE)
+        // An absent chain (deleted or fully rolled back) is already "not there"
+        if (!obj || heap->is_absent(obj))
         {
             // Already deleted
             op->retval = 0;
@@ -148,7 +149,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     }
     // FIXME: Allow to do initial writes as buffered, not redirected
     // FIXME: Allow to do direct writes over holes
-    else if (!obj || obj->type() == BS_HEAP_DELETE || op->offset == 0 && op->len == dsk.data_block_size)
+    else if (!obj || heap->is_absent(obj) || op->offset == 0 && op->len == dsk.data_block_size)
     {
         // Big (redirect) write
         PRIV(op)->write_type = dsk.disable_data_fsync || op->opcode != BS_OP_WRITE_STABLE ? BS_HEAP_BIG_WRITE : _REDIRECT_INTENT;

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: ");

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;

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}
 

src/osd/osd.h

@@ -231,20 +231,20 @@ class osd_t
     ring_consumer_t init_consumer;
 
     // op statistics
-    osd_op_stats_t prev_stats, prev_report_stats;
-    timespec report_stats_ts;
+    osd_op_stats_t prev_stats = {}, prev_report_stats = {};
+    timespec report_stats_ts = {};
     std::map<uint64_t, inode_stats_t> inode_stats;
     std::map<uint64_t, timespec> vanishing_inodes;
     const char* recovery_stat_names[2] = { "degraded", "misplaced" };
-    recovery_stat_t recovery_stat[2];
-    recovery_stat_t recovery_print_prev[2];
-    recovery_stat_t recovery_report_prev[2];
+    recovery_stat_t recovery_stat[2] = {};
+    recovery_stat_t recovery_print_prev[2] = {};
+    recovery_stat_t recovery_report_prev[2] = {};
 
     // recovery auto-tuning
     int rtune_timer_id = -1;
     uint64_t rtune_avg_lat = 0;
     double rtune_client_util = 0, rtune_target_util = 1;
-    osd_op_stats_t rtune_prev_stats, rtune_prev_recovery_stats;
+    osd_op_stats_t rtune_prev_stats = {}, rtune_prev_recovery_stats = {};
     std::vector<uint64_t> recovery_target_sleep_items;
     uint64_t recovery_target_sleep_us = 0;
     uint64_t recovery_target_sleep_total = 0;

src/osd/osd_cluster.cpp

@@ -294,7 +294,7 @@ void osd_t::report_statistics()
     last_stat = json11::Json::object();
     last_pool = 0;
     json11::Json::object inode_ops;
-    timespec tv_now;
+    timespec tv_now = { 0, 0 };
     for (auto st_it = inode_stats.begin(); st_it != inode_stats.end(); )
     {
         auto & kv = *st_it;
@@ -548,6 +548,8 @@ void osd_t::acquire_lease()
         }
         etcd_lease_id = data["ID"].string_value();
         create_osd_state();
+        // Flush PG state reports that were held back until the lease existed
+        report_pg_states();
     });
     printf(
         "[OSD %ju] reporting to etcd at %s every %d seconds (statistics every %d seconds)\n", this->osd_num,
@@ -1041,7 +1043,11 @@ struct reporting_pg_t
 
 void osd_t::report_pg_states()
 {
-    if (etcd_reporting_pg_state || !this->pg_state_dirty.size() || !st_cli->address_count())
+    // The lease is part of the state, so there's nothing to report before it's
+    // acquired. PG init may complete before acquire_lease() when blockstore init
+    // is asynchronous; reporting early would write lease 0 and then the lease
+    // check of every subsequent CAS on the state key would fail forever
+    if (etcd_lease_id == "" || etcd_reporting_pg_state || !this->pg_state_dirty.size() || !st_cli->address_count())
     {
         return;
     }

src/osd/osd_peering_pg.cpp

@@ -294,7 +294,12 @@ void pg_obj_state_check_t::finish_object()
             int j;
             for (j = 0; j < osd_set.size(); j++)
             {
-                if (osd_set[j].osd_num == list[i].osd_num)
+                // Only the same role at the same OSD subsumes this copy: an older
+                // version of the same piece on the same OSD is behind the target.
+                // The same OSD holding this piece at another role is a stale copy
+                // it must not be forgotten - deletes and cleanup have to reach it
+                if (osd_set[j].osd_num == list[i].osd_num &&
+                    osd_set[j].role == (list[i].oid.stripe & STRIPE_MASK))
                 {
                     break;
                 }
@@ -412,7 +417,10 @@ void pg_t::calc_object_states(int log_level)
     st.pg = this;
     st.replicated = (this->scheme == POOL_SCHEME_REPLICATED);
     auto ps = peering_state;
-    epoch = 0;
+    // Epoch must never go back: base it on the last epoch persisted in etcd, not on 0.
+    // Bumped epochs are reported to etcd before any write with a higher version starts,
+    // so a recreated object can never allocate versions below its own history
+    epoch = reported_epoch;
     for (auto it: ps->list_results)
     {
         auto nstab = it.second.stable_count;

src/osd/osd_primary.cpp

@@ -268,6 +268,19 @@ void osd_t::continue_primary_read(osd_op_t *cur_op)
 resume_0:
     cur_op->reply.rw.bitmap_len = 0;
     {
+        // Reset read ranges. On a retry we can come here with read ranges left on
+        // parity stripes by extend_missing_stripes() of the previous attempt -
+        // submit_primary_subops() would then send subops for them while
+        // alloc_read_buffer() only assigns buffers for ranges set below,
+        // making those subops write into a freed buffer.
+        for (int role = 0; role < op_data->stripe_count; role++)
+        {
+            op_data->stripes[role].read_start = 0;
+            op_data->stripes[role].read_end = 0;
+            op_data->stripes[role].missing = false;
+        }
+        // OP_DATA_DEGRADED may be left from a previous degraded attempt
+        op_data->flags &= ~(uint32_t)OP_DATA_DEGRADED;
         if (cur_op->req.rw.len == 0)
         {
             // len=0 => bitmap read
@@ -506,6 +519,32 @@ pg_osd_set_state_t *osd_t::mark_partial_write(pg_t & pg, osd_op_t *cur_op)
                 chunk.loc_bad &= ~LOC_OUTDATED;
             }
         }
+        // Add OSDs which took the new version, but weren't in the set (new locations of a misplaced object)
+        for (int i = 0; i < op_data->n_subops; i++)
+        {
+            if (op_data->subops[i].reply.hdr.retval != op_data->subops[i].req.sec_rw.len)
+            {
+                continue;
+            }
+            bool found = false;
+            for (auto & chunk: new_set)
+            {
+                if (chunk.osd_num == op_data->subops[i].osd_num)
+                {
+                    found = true;
+                    break;
+                }
+            }
+            if (!found)
+            {
+                changes++;
+                new_set.push_back((pg_obj_loc_t){
+                    .role = 0,
+                    .osd_num = op_data->subops[i].osd_num,
+                    .loc_bad = 0,
+                });
+            }
+        }
         return changes;
     });
 }
@@ -795,7 +834,11 @@ resume_1:
     {
         op_data->object_state->ref_count++;
     }
-    // Submit 1 read to determine the actual version number
+    // Submit 1 read to determine the actual version number. ENOENT is a legal answer:
+    // the object may be already deleted (a repeated delete - the client's remedy for a
+    // delete that reported left_on_dead). fact_ver 0 then makes the delete a no-op
+    // success instead of failing the repeated delete forever
+    op_data->flags |= OP_DATA_ENOENT_OK;
     submit_primary_subops(SUBMIT_RMW_READ, UINT64_MAX, op_data->prev_set, cur_op);
     op_data->prev_set = NULL;
 resume_2:
@@ -804,6 +847,7 @@ resume_2:
         op_data->st = 2;
         return;
     }
+    op_data->flags &= ~OP_DATA_ENOENT_OK;
 resume_3:
     if (op_data->errors > 0)
     {

src/osd/osd_primary_chain.cpp

@@ -586,9 +586,10 @@ void osd_t::send_chained_read_results(pg_t *pg, osd_op_t *cur_op)
     // Send bitmap
     cur_op->bitmap = op_data->stripes[0].bmp_buf;
     cur_op->reply.rw.bitmap_len = (pg ? pg->pg_data_size : 1) * clean_entry_bitmap_size;
+    size_t key_index_bytes = osd_op_rw_t::chain_info_bytes(op_data->chain_size);
     if (cur_op->req.rw.flags & OSD_OP_RETURN_CHAIN)
     {
-        cur_op->reply.rw.bitmap_len += (cur_op->req.rw.len / bs_bitmap_granularity);
+        cur_op->reply.rw.bitmap_len += key_index_bytes * (cur_op->req.rw.len / bs_bitmap_granularity);
     }
     // And finally compose the result
     uint64_t sent = 0;
@@ -597,7 +598,6 @@ void osd_t::send_chained_read_results(pg_t *pg, osd_op_t *cur_op)
     int prev = (cur_op->req.rw.offset - op_data->oid.stripe) / bs_bitmap_granularity;
     int end = prev + cur_op->req.rw.len/bs_bitmap_granularity;
     int cur = prev;
-    size_t key_index_bytes = osd_op_rw_t::chain_info_bytes(op_data->chain_size);
     while (cur <= end)
     {
         bool has_bit = false;

src/osd/osd_primary_write.cpp

@@ -304,7 +304,7 @@ resume_12:
             else
             {
                 pg.ver_override.erase(op_data->oid);
-                mark_partial_write(pg, cur_op);
+                op_data->object_state = mark_partial_write(pg, cur_op);
             }
         }
         if (op_data->subops)

src/test/CMakeLists.txt

@@ -88,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)

src/test/osd_test.cpp

@@ -455,7 +455,7 @@ void test_ec33_scrub_only_first_part()
         for (int i = 0; i < 6; i++)
             stripes[i].bmp_buf = bmp_buf.data() + i*4;
         split_stripes(3, 128*1024, 0, 128*1024, stripes);
-        void *rmw_buf = calc_rmw(data_buf.data(), stripes, fake_osd_set, 6, 3, 6, fake_osd_set, 128*1024, 4);
+        uint8_t *rmw_buf = (uint8_t*)calc_rmw(data_buf.data(), stripes, fake_osd_set, 6, 3, 6, fake_osd_set, 128*1024, 4);
         assert(rmw_buf);
         calc_rmw_parity_ec(stripes, 6, 3, fake_osd_set, fake_osd_set, 128*1024, 4);
         use_ec(6, 3, false);
@@ -1124,6 +1124,164 @@ void test_chained_read_eio_retry()
     printf("test_chained_read_eio_retry passed\n");
 }
 
+void test_chained_read_large_chain_info()
+{
+    printf("test_chained_read_large_chain_info\n");
+
+    osd_test_fixture_t f;
+    f.configure_replicated_pool(1, 2, 1, 1, { { 1, 2 } });
+
+    // Build 300-layer snapshot chain in pool 1:
+    // Inode 1 (root) <- Inode 2 <- ... <- Inode 300 (leaf)
+    const int NUM_SNAPS = 300;
+    for (int i = 1; i <= NUM_SNAPS; i++)
+    {
+        json11::Json::object obj;
+        if (i > 1)
+            obj["parent_id"] = i - 1;
+        f.st_cli->set("/vitastor/config/inode/1/" + std::to_string(i), obj);
+    }
+
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "none" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+    });
+    f.connect_peer(2);
+    f.complete_peering_empty();
+    assert(f.pg(1, 1).state == PG_ACTIVE);
+
+    uint64_t client_id = f.connect_client();
+
+    inode_t leaf_inode = INODE_WITH_POOL(1, NUM_SNAPS);
+    uint64_t meta_rev = f.st_cli->inode_config.at(leaf_inode).mod_revision;
+
+    auto *read_op = new osd_op_t();
+    read_op->op_type = OSD_OP_IN;
+    read_op->client_id = client_id;
+    read_op->req.rw.header.magic = SECONDARY_OSD_OP_MAGIC;
+    read_op->req.rw.header.id = 1;
+    read_op->req.rw.header.opcode = OSD_OP_READ;
+    read_op->req.rw.inode = leaf_inode;
+    read_op->req.rw.offset = 0;
+    read_op->req.rw.len = 128 * 1024;
+    read_op->req.rw.meta_revision = meta_rev;
+    read_op->req.rw.flags = OSD_OP_RETURN_CHAIN;
+
+    f.exec(read_op);
+
+    // Verify op completed and replied to client
+    assert(f.client(client_id)->sent_ops.count(read_op->req.hdr.id) == 1);
+    auto *reply_op = f.client(client_id)->sent_ops.at(read_op->req.hdr.id);
+    assert(reply_op == read_op);
+    assert(reply_op->reply.hdr.retval == (int64_t)read_op->req.rw.len);
+
+    // Replicated pool clean_entry_bitmap_size = 131072 / 4096 / 8 = 4 bytes
+    // For chain_size = 300, key_index_bytes = osd_op_rw_t::chain_info_bytes(300) = 2
+    // 32 blocks (128k / 4k) * 2 bytes = 64 bytes of chain_info
+    // Total bitmap_len must be 4 + 64 = 68 bytes.
+    // If osd_primary_chain.cpp missed multiplying key_index_bytes, bitmap_len would be 4 + 32 = 36.
+    assert(reply_op->reply.rw.bitmap_len == 68);
+
+    printf("test_chained_read_large_chain_info passed\n");
+}
+
+
+// A degraded EC read may leave its parity read ranges behind when the object state
+// changes under the operation (recovery brings a chunk back while subops are in flight)
+// and a failed subop retries the whole read. The retry then takes the fast path, which
+// only assigns buffers for data stripes - a parity subop sent on a leftover range would
+// write its reply into a freed buffer. The retry has to reset the read ranges.
+void test_ec_degraded_read_retry_resets_parity()
+{
+    printf("test_ec_degraded_read_retry_resets_parity\n");
+
+    osd_test_fixture_t f;
+    // EC 2+2, set {1,2,3,4}, primary = OSD 2. OSD 1 (role 0) holds nothing, so the
+    // object is degraded: it only exists at roles 1(local), 2 and 3
+    f.configure_ec_pool(2, 2, /*primary_osd*/ 2);
+    f.st_cli->set("/vitastor/pg/history/1/1", json11::Json::object {
+        { "osd_sets", json11::Json::array {
+            json11::Json::array{ 1, 2, 3, 4 },
+        } },
+    });
+    f.start(json11::Json::object {
+        { "osd_num", 2 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "none" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+        { "no_recovery", true },
+    });
+    inode_t inode = INODE_WITH_POOL(1, 1);
+
+    f.connect_peer(1);
+    f.connect_peer(3);
+    f.connect_peer(4);
+    // OSD 1 holds nothing -> the object is missing at role 0
+    f.reply_peer_list(1, {}, 0);
+    f.reply_peer_list(3, { { { inode, 2 }, 1 } }, 1);
+    f.reply_peer_list(4, { { { inode, 3 }, 1 } }, 1);
+    f.reply_local_list({ { { inode, 1 }, 1 } }, 1);
+    f.ringloop->loop();
+    assert(f.pg(1, 1).state == (PG_ACTIVE|PG_HAS_DEGRADED));
+
+    // Read 0..32k. It lies in role 0, which is missing, so the attempt is degraded:
+    // extend_missing_stripes() moves the read to role 1 (local) + role 2 (parity, OSD 3)
+    auto *read_op = new osd_op_t();
+    read_op->op_type = OSD_OP_IN;
+    read_op->client_id = 0;
+    read_op->req.rw.header.magic = SECONDARY_OSD_OP_MAGIC;
+    read_op->req.rw.header.id = 1;
+    read_op->req.rw.header.opcode = OSD_OP_READ;
+    read_op->req.rw.inode = inode;
+    read_op->req.rw.offset = 0;
+    read_op->req.rw.len = 32768;
+    int read_retval = -1;
+    read_op->callback = [&](osd_op_t *op) { read_retval = op->reply.hdr.retval; };
+    f.exec(read_op);
+
+    // Attempt 1: local read of role 1 and a parity read from OSD 3
+    auto *local_read = f.bs->take(BS_OP_READ);
+    assert(local_read->offset == 0 && local_read->len == 32768);
+    auto *parity_read = f.peer_take(3, OSD_OP_SEC_READ);
+    assert(parity_read->req.sec_rw.oid.stripe == 2);
+    assert(parity_read->req.sec_rw.offset == 0 && parity_read->req.sec_rw.len == 32768);
+
+    // While the subops are in flight a peering round recomputes the object state
+    // and finds it complete again: the PG goes to exactly PG_ACTIVE and the parity
+    // range of the degraded attempt becomes stale. Peering itself won't run while
+    // an op is inflight, but the object state can be replaced under a running op
+    // by any parallel event - drive the resulting state directly
+    f.pg(1, 1).state = PG_ACTIVE;
+
+    // The local read completes, then the parity read fails with -EIO. The error
+    // handler sees that the object state changed under the op and retries the read
+    local_read->version = 1;
+    local_read->retval = local_read->len;
+    local_read->callback(local_read);
+    f.peer_complete(parity_read, -EIO);
+
+    // The retry takes the fast path now. It must not re-send the leftover parity
+    // range - there is no read buffer for it anymore
+    assert(!f.peer_has_sent_op(3, OSD_OP_SEC_READ));
+    assert(!f.peer_has_sent_op(4, OSD_OP_SEC_READ));
+
+    // The retry reads the recovered chunk normally and the op completes.
+    // role 1 is not read this time - it was only pulled in to reconstruct role 0
+    auto *recovered_read = f.peer_take(1, OSD_OP_SEC_READ);
+    assert(recovered_read->req.sec_rw.oid.stripe == 0);
+    assert(recovered_read->req.sec_rw.len == 32768);
+    recovered_read->reply.sec_rw.version = 1;
+    f.peer_complete(recovered_read, 32768);
+    assert(read_retval == 32768);
+
+    delete read_op;
+    printf("test_ec_degraded_read_retry_resets_parity passed\n");
+}
+
 // Regression test for the bug in osd_flush.cpp where a remote
 // stabilize/rollback failure permanently hangs the PG in PG_REPEERING.
 bool test_flush_error_pg_repeer(int ctr)
@@ -2156,6 +2314,433 @@ void test_repeated_partial_write_unmarks_outdated()
     printf("test_repeated_partial_write_unmarks_outdated passed\n");
 }
 
+// A partially failed write to a misplaced object must record the OSDs that
+// took the new version, even if the object wasn't on them before.
+//
+// Layout: pool 1, replicated x3, PG target set 1, 2, 3, primary = OSD 1 (us).
+// The object is still on OSD 1, 2 and 4 from the previous set, so the write
+// goes to OSD 1, 2 and 3, and OSD 3 isn't in the object's location set yet.
+void test_partial_write_to_misplaced_object()
+{
+    printf("test_partial_write_to_misplaced_object\n");
+
+    osd_test_fixture_t f;
+    f.configure_replicated_pool(/*pool_id*/ 1, /*pg_size*/ 3, /*pg_minsize*/ 1, /*pg_count*/ 1,
+        { { 1, 2, 3 } });
+    // OSD 4 is in the PG history, so peering lists it, too
+    f.st_cli->set("/vitastor/pg/history/1/1", json11::Json::object {
+        { "osd_sets", json11::Json::array {
+            json11::Json::array{ 1, 2, 4 },
+        } },
+    });
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "all" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+        // Don't let rebalance move the object before the test writes to it
+        { "no_rebalance", true },
+        { "no_recovery", true },
+    });
+    f.connect_peer(2);
+    f.connect_peer(3);
+    f.connect_peer(4);
+
+    inode_t inode = INODE_WITH_POOL(1, 1);
+    object_id oid = { .inode = inode, .stripe = 0 };
+
+    // v1 is on OSD 1, 2 and 4, OSD 3 doesn't have the object yet
+    f.reply_local_list({ { oid, 1 } }, 1);
+    f.reply_peer_list(2, { { oid, 1 } }, 1);
+    f.reply_peer_list(3, {}, 0);
+    f.reply_peer_list(4, { { oid, 1 } }, 1);
+    f.pump();
+    assert(f.pg(1, 1).state == (PG_ACTIVE|PG_HAS_MISPLACED));
+    assert(f.pg(1, 1).misplaced_objects.count(oid));
+
+    // Does the object's location set have a current copy on <osd_num>?
+    auto current_on = [&](osd_num_t osd_num)
+    {
+        auto *st = find_object_state(f.pg(1, 1), oid);
+        assert(st);
+        for (auto & chunk: st->osd_set)
+            if (chunk.osd_num == osd_num)
+                return !(chunk.loc_bad & LOC_OUTDATED);
+        return false;
+    };
+
+    // ============ WRITE — ENOSPC on OSD 1 and 2, OK on OSD 3 ============
+
+    // Full-block write, so there's no read-modify-write
+    auto *write = make_write_op(inode, 0, 131072, 0xab);
+    int write_retval = -1;
+    write->callback = [&write_retval](osd_op_t *op) { write_retval = op->reply.hdr.retval; };
+    f.exec(write);
+
+    // Zero-length read resolving the object version
+    f.bs_zero_read_ok(1);
+
+    // The write goes to the target set, not to OSD 4
+    auto *local_w = f.bs->take(BS_OP_WRITE_STABLE);
+    // The new version includes the PG epoch, which the PG history has bumped
+    uint64_t new_ver = local_w->version;
+    assert(new_ver > 1);
+    local_w->retval = -ENOSPC;
+    local_w->callback(local_w);
+    f.peer_complete(f.peer_take(2, OSD_OP_SEC_WRITE_STABLE), -ENOSPC);
+    f.peer_write_ok(3, OSD_OP_SEC_WRITE_STABLE, new_ver);
+    assert(f.peer(4)->sent_ops.empty());
+    assert(write_retval == -ENOSPC);
+
+    // THE KEY ASSERTION — OSD 3 has the new version, so it's a current copy.
+    // Otherwise no current copy is left and the object becomes incomplete.
+    if (!current_on(3))
+        printf("BUG: OSD 3 took the new version, but isn't recorded as a current copy\n");
+    assert(current_on(3));
+    assert(!current_on(1));
+    assert(!current_on(2));
+    assert(!current_on(4));
+    assert(!(f.pg(1, 1).state & PG_HAS_INCOMPLETE));
+
+    // ============ READ — served by OSD 3 instead of failing with EIO ============
+
+    auto *read = make_self_rw(OSD_OP_READ, inode, 0, 4096);
+    int read_retval = -1;
+    read->callback = [&read_retval](osd_op_t *op) { read_retval = op->reply.hdr.retval; };
+    f.exec(read);
+    assert(read_retval == -1);
+    auto *peer3_read = f.peer_take(3, OSD_OP_SEC_READ);
+    assert(peer3_read->req.sec_rw.len == 4096);
+    peer3_read->reply.sec_rw.version = new_ver;
+    f.peer_complete(peer3_read, 4096);
+    assert(read_retval == 4096);
+
+    assert(f.pg(1, 1).inflight == 0);
+    assert(f.pg(1, 1).write_queue.empty());
+
+    delete read;
+    delete write;
+    printf("test_partial_write_to_misplaced_object passed\n");
+}
+
+// Regression test: continue_primary_write() ignored the object state returned by
+// mark_partial_write(). It leaked a reference to the new state and released the old
+// state twice, which is a use-after-free when mark_object() has already freed it
+void test_partial_write_object_state_refs()
+{
+    printf("test_partial_write_object_state_refs\n");
+
+    osd_test_fixture_t f;
+    f.configure_replicated_pool(/*pool_id*/ 1, /*pg_size*/ 3, /*pg_minsize*/ 1, /*pg_count*/ 1,
+        { { 1, 2, 3 } });
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "all" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+        { "no_recovery", true },
+    });
+    f.connect_peer(2);
+    f.connect_peer(3);
+
+    inode_t inode = INODE_WITH_POOL(1, 1);
+    object_id oid = { .inode = inode, .stripe = 0 };
+
+    // OSD 1 and 2 have the object, OSD 3 doesn't -> degraded after peering
+    f.reply_local_list({ { oid, 1 } }, 1);
+    f.reply_peer_list(2, { { oid, 1 } }, 1);
+    f.reply_peer_list(3, {}, 0);
+    f.ringloop->loop();
+    assert(f.pg(1, 1).state == (PG_ACTIVE|PG_HAS_DEGRADED));
+    auto *st = find_object_state(f.pg(1, 1), oid);
+    assert(st && st->object_count == 1 && st->ref_count == 0);
+
+    // ENOSPC locally, OK on OSD 2 and 3
+    auto *write = make_write_op(inode, 0, 131072, 0xcd);
+    int retval = -1;
+    write->callback = [&retval](osd_op_t *op) { retval = op->reply.hdr.retval; };
+    f.exec(write);
+    f.bs_zero_read_ok(1);
+    auto *local_w = f.bs->take(BS_OP_WRITE_STABLE);
+    uint64_t version = local_w->version;
+    local_w->retval = -ENOSPC;
+    local_w->callback(local_w);
+    f.peer_write_ok(2, OSD_OP_SEC_WRITE_STABLE, version);
+    f.peer_write_ok(3, OSD_OP_SEC_WRITE_STABLE, version);
+    assert(retval == -ENOSPC);
+    assert(f.pg(1, 1).inflight == 0);
+
+    // No operation is in flight, so nothing may hold a reference to the new state
+    st = find_object_state(f.pg(1, 1), oid);
+    assert(st && st->object_count == 1);
+    assert(st->ref_count == 0);
+    delete write;
+
+    // The same for a partially failed write to a clean object: ENOSPC on OSD 3 only
+    object_id oid2 = { .inode = inode, .stripe = 131072 };
+    auto *write2 = make_write_op(inode, 131072, 131072, 0xef);
+    int retval2 = -1;
+    write2->callback = [&retval2](osd_op_t *op) { retval2 = op->reply.hdr.retval; };
+    f.exec(write2);
+    f.bs_zero_read_ok(0);
+    auto *local_w2 = f.bs->take(BS_OP_WRITE_STABLE);
+    uint64_t version2 = local_w2->version;
+    local_w2->retval = local_w2->len;
+    local_w2->callback(local_w2);
+    f.peer_write_ok(2, OSD_OP_SEC_WRITE_STABLE, version2);
+    f.peer_complete(f.peer_take(3, OSD_OP_SEC_WRITE_STABLE), -ENOSPC);
+    assert(retval2 == -ENOSPC);
+    assert(f.pg(1, 1).inflight == 0);
+    st = find_object_state(f.pg(1, 1), oid2);
+    assert(st && st->object_count == 1);
+    assert(st->ref_count == 0);
+    delete write2;
+
+    printf("test_partial_write_object_state_refs passed\n");
+}
+
+// A repeated delete of an already deleted object must succeed as a no-op.
+// The fact-version read of the repeat sees ENOENT - that must resolve to
+// "nothing to delete", not to an error.
+void test_repeated_delete_is_noop()
+{
+    printf("test_repeated_delete_is_noop\n");
+
+    osd_test_fixture_t f;
+    f.configure_replicated_pool(/*pool_id*/ 1, /*pg_size*/ 2, /*pg_minsize*/ 1, /*pg_count*/ 1,
+        { { 1, 2 } });
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "all" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+    });
+    f.connect_peer(2);
+
+    inode_t inode = INODE_WITH_POOL(1, 1);
+    object_id oid = { .inode = inode, .stripe = 0 };
+    // v1 is present on both members
+    f.reply_local_list({ { oid, 1 } }, 1);
+    f.reply_peer_list(2, { { oid, 1 } }, 1);
+    f.pump();
+    assert(f.pg(1, 1).state == PG_ACTIVE);
+
+    auto make_delete_op = [&]()
+    {
+        auto *op = new osd_op_t();
+        op->op_type = OSD_OP_IN;
+        op->client_id = 0;
+        op->req.rw.header.magic = SECONDARY_OSD_OP_MAGIC;
+        op->req.rw.header.id = 1;
+        op->req.rw.header.opcode = OSD_OP_DELETE;
+        op->req.rw.inode = inode;
+        op->req.rw.offset = 0;
+        op->req.rw.len = 0;
+        return op;
+    };
+
+    // First delete: removes the chain locally and on the peer
+    int delete_retval = -1;
+    auto *del1 = make_delete_op();
+    del1->callback = [&delete_retval](osd_op_t *op) { delete_retval = op->reply.hdr.retval; };
+    f.exec(del1);
+    f.bs_zero_read_ok(1);
+    {
+        auto *bd = f.bs->take(BS_OP_DELETE);
+        bd->version = 1;
+        bd->retval = 0;
+        bd->callback(bd);
+    }
+    f.peer_write_ok(2, OSD_OP_SEC_DELETE, 1);
+    assert(delete_retval == 0);
+    delete del1;
+
+    // Repeated delete: nothing left anywhere, must ack cleanly
+    delete_retval = -1;
+    auto *del2 = make_delete_op();
+    del2->callback = [&delete_retval](osd_op_t *op) { delete_retval = op->reply.hdr.retval; };
+    f.exec(del2);
+    // The local chain is gone: the fact read answers ENOENT
+    f.bs_zero_read_ok(0);
+    f.pump();
+    // THE KEY ASSERTION - the repeat is a no-op success, not an error
+    assert(delete_retval == 0);
+    // And it really deleted nothing: no subops went out
+    assert(f.peer(2)->sent_ops.empty());
+    delete del2;
+
+    printf("test_repeated_delete_is_noop passed\n");
+}
+
+// Peering must not forget an outdated copy just because its OSD already holds
+// the object at another role. A stale piece kept by a set member after a set
+// change looks, by OSD number alone, like a copy that the target set already
+// covers - but the target version at that role is held by a different OSD, so
+// the stale copy has to land in the object's OSD set as LOC_OUTDATED, or
+// deletes and cleanup would never reach it.
+//
+// Layout: EC 2+2 pool, PG 1/1, cur_set = role 0..3 => OSD 1..4. Object
+// INODE_WITH_POOL(1,1) is fully present at version 2 (role R at OSD R+1), and
+// OSD 2 additionally keeps the role-3 piece of the older version 1.
+void test_peering_outdated_copy_other_role()
+{
+    printf("test_peering_outdated_copy_other_role\n");
+
+    inode_t inode = INODE_WITH_POOL(1, 1);
+
+    // One pg_t per scenario; the peer LIST results are fed straight into
+    // calc_object_states, the way peering hands them over. calc_object_states
+    // frees the buffers, so they are malloc'ed
+    auto make_pg = []()
+    {
+        pg_t *pg = new pg_t();
+        pg->pool_id = 1;
+        pg->pg_num = 1;
+        pg->scheme = POOL_SCHEME_EC;
+        pg->pg_size = 4;
+        pg->pg_cursize = 4;
+        pg->pg_minsize = 2;
+        pg->pg_data_size = 2;
+        pg->target_set = { 1, 2, 3, 4 };
+        pg->cur_set = { 1, 2, 3, 4 };
+        pg->all_peers = { 1, 2, 3, 4 };
+        pg->cur_peers = { 1, 2, 3, 4 };
+        pg->peering_state = new pg_peering_state_t();
+        return pg;
+    };
+    // A peer list of (stripe, version) entries, all stable
+    auto add_list = [](pg_t *pg, osd_num_t osd_num, std::vector<std::pair<uint64_t, uint64_t>> items)
+    {
+        obj_ver_id *buf = (obj_ver_id*)malloc(sizeof(obj_ver_id)*items.size());
+        for (int i = 0; i < items.size(); i++)
+            buf[i] = { .oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = items[i].first }, .version = items[i].second };
+        pg->peering_state->list_results[osd_num] = (pg_list_result_t){
+            .buf = buf, .total_count = items.size(), .stable_count = items.size(),
+        };
+    };
+
+    // ============ A: stale role-3 copy on OSD 2, which holds role 1 at target ============
+
+    pg_t *pg = make_pg();
+    add_list(pg, 1, { { 0, 2 } });
+    add_list(pg, 2, { { 1, 2 }, { 3, 1 } });
+    add_list(pg, 3, { { 2, 2 } });
+    add_list(pg, 4, { { 3, 2 } });
+    pg->calc_object_states(0);
+
+    object_id oid = { .inode = inode, .stripe = 0 };
+    assert(pg->state & PG_HAS_MISPLACED);
+    assert(pg->total_count == 1 && pg->clean_count == 0);
+    pg_osd_set_state_t *st = find_object_state(*pg, oid);
+    if (!st)
+        printf("BUG: the stale copy on OSD 2 was forgotten by peering\n");
+    assert(st);
+    bool outdated = false;
+    for (auto & chunk: st->osd_set)
+        if (chunk.role == 3 && chunk.osd_num == 2)
+            outdated = (chunk.loc_bad & LOC_OUTDATED) != 0;
+    assert(outdated);
+    // Reads must still target the good copy, not the outdated one
+    assert(st->read_target.size() == 4 && st->read_target[3] == 4);
+    delete pg->peering_state;
+    delete pg;
+    printf("test_peering_outdated_copy_other_role: stale copy on OSD 2 marked outdated\n");
+
+    // ============ B: the same role on the same OSD stays subsumed ============
+
+    // The OSD's own history: an older version of the very piece it holds at the
+    // target version is not a stray copy - it must not appear as outdated
+    pg = make_pg();
+    add_list(pg, 1, { { 0, 2 } });
+    add_list(pg, 2, { { 1, 2 }, { 1, 1 } });
+    add_list(pg, 3, { { 2, 2 } });
+    add_list(pg, 4, { { 3, 2 } });
+    pg->calc_object_states(0);
+    assert(!(pg->state & PG_HAS_MISPLACED));
+    assert(pg->total_count == 1 && pg->clean_count == 1);
+    assert(!find_object_state(*pg, oid));
+    delete pg->peering_state;
+    delete pg;
+
+    printf("test_peering_outdated_copy_other_role passed\n");
+}
+
+// PG epoch must be based on the value persisted in etcd, not on the version
+// epochs of LIST copies. Copies of the epoch-7 chain may be fully deleted while
+// etcd still remembers epoch 7; if peering recomputed the epoch from surviving
+// older copies it would regress, and writes to recreated objects would reuse
+// version numbers that copies missed by a delete may still hold - peering would
+// then treat those stale copies as up-to-date.
+void test_pg_epoch_base_is_reported_epoch()
+{
+    printf("test_pg_epoch_base_is_reported_epoch\n");
+
+    osd_test_fixture_t f;
+    f.configure_replicated_pool(/*pool_id*/ 1, /*pg_size*/ 2, /*pg_minsize*/ 1, /*pg_count*/ 1,
+        { { 1, 2 } });
+    // etcd persisted epoch 7; the LIST will only contain epoch-4 copies
+    f.st_cli->set("/vitastor/pg/history/1/1", json11::Json::object {
+        { "osd_sets", json11::Json::array { json11::Json::array { 1, 2 } } },
+        { "all_peers", json11::Json::array { 1, 2 } },
+        { "epoch", 7 },
+    });
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "none" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+        { "no_recovery", true },
+    });
+    assert(f.st_cli->pool_config[1].pg_config[1].epoch == 7);
+    f.connect_peer(2);
+    f.reply_local_list({ { { INODE_WITH_POOL(1, 1), 0 }, 0x40001 } }, 1);
+    f.reply_peer_list(2, { { { INODE_WITH_POOL(1, 1), 0 }, 0x40001 } }, 1);
+    f.ringloop->loop();
+    // A clean activation must not bump the epoch, but it also must not drag it
+    // down to the 4 seen in copies
+    assert(f.pg(1, 1).state == PG_ACTIVE);
+    assert(f.pg(1, 1).epoch == 7);
+
+    printf("test_pg_epoch_base_is_reported_epoch passed\n");
+}
+
+// A non-clean activation must bump the epoch above the persisted base, not above
+// the version epochs of surviving copies
+void test_pg_epoch_bump_above_reported_epoch()
+{
+    printf("test_pg_epoch_bump_above_reported_epoch\n");
+
+    osd_test_fixture_t f;
+    f.configure_replicated_pool(/*pool_id*/ 1, /*pg_size*/ 2, /*pg_minsize*/ 1, /*pg_count*/ 1,
+        { { 1, 2 } });
+    f.st_cli->set("/vitastor/pg/history/1/1", json11::Json::object {
+        { "osd_sets", json11::Json::array { json11::Json::array { 1, 2 } } },
+        { "all_peers", json11::Json::array { 1, 2 } },
+        { "epoch", 7 },
+    });
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "none" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+        { "no_recovery", true },
+    });
+    f.connect_peer(2);
+    // Only one copy comes back -> the PG is not purely ACTIVE
+    f.reply_local_list({ { { INODE_WITH_POOL(1, 1), 0 }, 0x40001 } }, 1);
+    f.reply_peer_list(2, {}, 0);
+    f.ringloop->loop();
+    assert(f.pg(1, 1).state != PG_ACTIVE);
+    assert(f.pg(1, 1).epoch == 8);
+
+    printf("test_pg_epoch_bump_above_reported_epoch passed\n");
+}
+
 int main(int narg, char *args[])
 {
     test_load_global_config();
@@ -2167,6 +2752,7 @@ int main(int narg, char *args[])
     test_ec42_write_parityless();
     test_ec33_chain_read_phantom_bitmap_source();
     test_chained_read_eio_retry();
+    test_ec_degraded_read_retry_resets_parity();
     for (int i = 0; test_flush_error_pg_repeer(i); i++) {}
     test_sync_nothing_to_sync();
     test_sync_happy_path_replicated();
@@ -2180,5 +2766,12 @@ int main(int narg, char *args[])
     test_pg_epoch_bump_blocks_write_until_reported();
     test_ec42_inconsistent_read();
     test_repeated_partial_write_unmarks_outdated();
+    test_partial_write_to_misplaced_object();
+    test_partial_write_object_state_refs();
+    test_repeated_delete_is_noop();
+    test_peering_outdated_copy_other_role();
+    test_pg_epoch_base_is_reported_epoch();
+    test_pg_epoch_bump_above_reported_epoch();
+    test_chained_read_large_chain_info();
     return 0;
 }

src/test/osd_test_fixture.h

@@ -425,4 +425,15 @@ struct osd_test_fixture_t
     {
         return osd->syncs_in_progress.size();
     }
+
+    bool peer_has_sent_op(osd_num_t osd_num, uint64_t opcode)
+    {
+        auto *cl = osd->msgr.osd_peers.at(osd_num);
+        for (auto & kv: cl->sent_ops)
+        {
+            if (kv.second->req.hdr.opcode == opcode)
+                return true;
+        }
+        return false;
+    }
 };

src/test/test_blockstore.cpp

@@ -1714,6 +1714,94 @@ static void test_compact_over_rolled_back_big_write()
     free(buf);
 }
 
+// A write over a chain rolled back to 0 must behave like a write over a new object.
+static void test_zero_write_over_rollback_to_zero()
+{
+    printf("\n-- test_zero_write_over_rollback_to_zero\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["immediate_commit"] = "none";
+    test.init();
+
+    object_id oid = { .inode = 1, .stripe = 0 };
+    uint32_t block_size = test.bs->dsk.data_block_size;
+    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, block_size);
+
+    auto stabilize = [&](object_id o, uint64_t version, bool rollback)
+    {
+        blockstore_op_t s;
+        s.opcode = rollback ? BS_OP_ROLLBACK : BS_OP_STABLE;
+        s.len = 1;
+        s.buf = (uint8_t*)malloc_or_die(sizeof(obj_ver_id));
+        *((obj_ver_id*)s.buf) = (obj_ver_id){ .oid = o, .version = version };
+        test.exec_op(&s);
+        assert(s.retval == 0);
+        free(s.buf);
+    };
+
+    printf("write v1 (big, unstable) and roll it back to 0\n");
+    blockstore_op_t op;
+    op.oid = oid;
+    op.buf = buf;
+    op.opcode = BS_OP_WRITE;
+    op.version = 1;
+    op.offset = 0;
+    op.len = block_size;
+    memset(buf, 0xAA, block_size);
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+    stabilize(oid, 0, true);
+
+    printf("the rolled back object must read as ENOENT\n");
+    blockstore_op_t rd;
+    rd.opcode = BS_OP_READ;
+    rd.oid = oid;
+    rd.version = UINT64_MAX;
+    rd.offset = 0;
+    rd.len = block_size;
+    rd.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, rd.len);
+    test.exec_op(&rd);
+    assert(rd.retval == -ENOENT);
+
+    printf("zero-length marker write v2 over the rolled back chain\n");
+    op.opcode = BS_OP_WRITE;
+    op.version = 2;
+    op.offset = 0;
+    op.len = 0;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+    stabilize(oid, 2, false);
+
+    printf("v2 must be readable and contain zeros\n");
+    memset(rd.buf, 0xCC, rd.len);
+    rd.version = UINT64_MAX;
+    test.exec_op(&rd);
+    assert(rd.retval == (int)rd.len);
+    assert(rd.version == 2);
+    assert(memcheck(rd.buf, 0, rd.len));
+
+    printf("a partial write over the marker must merge into the materialized block\n");
+    op.opcode = BS_OP_WRITE;
+    op.version = 3;
+    op.offset = 8192;
+    op.len = 4096;
+    memset(buf, 0xBB, 4096);
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+    memset(rd.buf, 0xCC, rd.len);
+    rd.version = UINT64_MAX;
+    test.exec_op(&rd);
+    assert(rd.retval == (int)rd.len);
+    assert(rd.version == 3);
+    assert(memcheck(rd.buf, 0, 8192));
+    assert(memcheck(rd.buf + 8192, 0xBB, 4096));
+    assert(memcheck(rd.buf + 12288, 0, rd.len - 12288));
+
+    free(rd.buf);
+    free(buf);
+}
+
 // With a checksum block larger than the write granularity, compaction puts the intent writes it
 // merges into read_vec as well - not to write them (their data is already in place inside the
 // data block) but because recalculating the checksum of a block they only partially cover needs
@@ -1942,6 +2030,7 @@ int main(int narg, char *args[])
     test_compact_big_intent_survives_crash(0);
     test_recheck_under_commit();
     test_compact_over_rolled_back_big_write();
+    test_zero_write_over_rollback_to_zero();
     test_compact_intent_write_in_partial_csum_block();
     test_no_intent_over_unwritten_compaction();
     return 0;

src/test/test_blockstore_v1.cpp

@@ -2804,6 +2804,90 @@ static void test_sync_waits_for_delete_taking_over_unsynced()
     free(buf);
 }
 
+// The same scenario as test_zero_write_over_rollback_to_zero in the new store tests:
+// a chain rolled back to version 0 is absent, and the next write (a zero-length EC
+// "version marker") must behave like a write over a new object and read back as zeros.
+static void test_zero_write_over_rollback_to_zero()
+{
+    printf("\n-- test_zero_write_over_rollback_to_zero\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["disable_data_fsync"] = "0";
+    test.config["immediate_commit"] = "none";
+    test.init();
+
+    object_id oid = { .inode = 1, .stripe = 0 };
+    uint32_t block_size = test.dsk().data_block_size;
+    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, block_size);
+
+    blockstore_op_t op;
+    op.oid = oid;
+    op.buf = buf;
+
+    printf("write v1 (unstable) and sync it\n");
+    op.opcode = BS_OP_WRITE;
+    op.version = 1;
+    op.offset = 0;
+    op.len = block_size;
+    memset(buf, 0xAA, block_size);
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+    op.opcode = BS_OP_SYNC;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+
+    printf("roll back to v0\n");
+    blockstore_op_t rb;
+    rb.opcode = BS_OP_ROLLBACK;
+    rb.len = 1;
+    rb.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, sizeof(obj_ver_id));
+    *((obj_ver_id*)rb.buf) = (obj_ver_id){ .oid = oid, .version = 0 };
+    test.exec_op(&rb);
+    assert(rb.retval == 0);
+    free(rb.buf);
+
+    printf("the object must be gone\n");
+    blockstore_op_t rd;
+    rd.opcode = BS_OP_READ;
+    rd.oid = oid;
+    rd.version = UINT64_MAX;
+    rd.offset = 0;
+    rd.len = block_size;
+    rd.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, rd.len);
+    test.exec_op(&rd);
+    assert(rd.retval == -ENOENT);
+
+    printf("zero-length marker write v2 over the rolled back chain\n");
+    op.opcode = BS_OP_WRITE;
+    op.version = 2;
+    op.offset = 0;
+    op.len = 0;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+
+    printf("stabilize v2\n");
+    blockstore_op_t st;
+    st.opcode = BS_OP_STABLE;
+    st.len = 1;
+    st.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, sizeof(obj_ver_id));
+    *((obj_ver_id*)st.buf) = (obj_ver_id){ .oid = oid, .version = 2 };
+    test.exec_op(&st);
+    assert(st.retval == 0);
+    free(st.buf);
+
+    printf("v2 must be readable and contain zeros\n");
+    rd.version = UINT64_MAX;
+    memset(rd.buf, 0xCC, rd.len);
+    test.exec_op(&rd);
+    assert(rd.retval == (int)rd.len);
+    assert(rd.version == 2);
+    assert(memcheck(rd.buf, 0, rd.len));
+
+    free(rd.buf);
+    free(buf);
+}
+
 int main(int narg, char *args[])
 {
     test_preserve_corruption();
@@ -2836,5 +2920,6 @@ int main(int narg, char *args[])
     test_recovered_state_is_made_durable();
     test_rollback_over_stable_write();
     test_sync_waits_for_delete_taking_over_unsynced();
+    test_zero_write_over_rollback_to_zero();
     return 0;
 }

src/test/test_heap.cpp

@@ -865,6 +865,26 @@ void test_iterate_compaction()
         assert(small_writes == 0);
     }
 
+    {
+        blockstore_heap_t heap(&dsk, buffer_area.data());
+        heap.finish_recheck();
+
+        // Case: BIG_UNSTABLE(v1 l1) ROLLBACK(v0 l2): a delayed commit of the
+        // rolled back version must return ENOENT, not crash - the rollback
+        // covers the whole chain, there is no entry with version <= 0 to
+        // stop the skip loop on
+        uint32_t mblock = 0;
+        _test_big_write(heap, dsk, 1, 0, 1, 0, false, 0, 4096, buffer_area.data());
+        object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
+        auto obj = heap.read_entry(oid);
+        res = heap.add_rollback(obj, 0, &mblock);
+        assert(res == 0);
+        heap.start_block_write(mblock);
+        heap.complete_block_write(mblock);
+        obj = heap.read_entry(oid);
+        assert(heap.add_commit(obj, 1, &mblock) == ENOENT);
+    }
+
     {
         blockstore_heap_t heap(&dsk, buffer_area.data());
         heap.finish_recheck();
@@ -1328,6 +1348,162 @@ void test_iterate_compaction()
     printf("OK test_iterate_compaction\n");
 }
 
+// Test for compaction with incorrect on-disk chains with COMMIT over nothing,
+// or for a COMMIT over no BIG_WRITE.
+void test_iterate_compaction_commit_over_nothing()
+{
+    printf("test_iterate_compaction_commit_over_nothing\n");
+    blockstore_disk_t dsk;
+    _test_init(dsk, false);
+    std::vector<uint8_t> buffer_area(dsk.journal_device_size);
+    std::vector<uint8_t> tmp(dsk.meta_block_size);
+
+    auto add_big = [&](blockstore_heap_t & heap, size_t & total_size, uint64_t lsn, uint64_t version)
+    {
+        auto wr1 = (heap_entry_t*)(tmp.data() + total_size);
+        wr1->size = heap.get_big_entry_size();
+        wr1->entry_type = BS_HEAP_BIG_WRITE;
+        wr1->lsn = lsn;
+        wr1->inode = INODE_WITH_POOL(1, 1);
+        wr1->stripe = 0;
+        wr1->version = version;
+        wr1->set_big_location(&heap, 0);
+        memset(wr1->get_ext_bitmap(&heap), 0xff, dsk.clean_entry_bitmap_size);
+        wr1->checksum = wr1->calc_checksum(&heap);
+        total_size += wr1->size;
+    };
+    auto add_small = [&](blockstore_heap_t & heap, size_t & total_size, uint64_t lsn, uint64_t version)
+    {
+        auto wr1 = (heap_entry_t*)(tmp.data() + total_size);
+        wr1->size = heap.get_small_entry_size(0, 4096);
+        wr1->entry_type = BS_HEAP_SMALL_WRITE;
+        wr1->lsn = lsn;
+        wr1->inode = INODE_WITH_POOL(1, 1);
+        wr1->stripe = 0;
+        wr1->version = version;
+        wr1->small().offset = 0;
+        wr1->small().len = 4096;
+        wr1->small().location = 0;
+        memset(wr1->get_ext_bitmap(&heap), 0xff, dsk.clean_entry_bitmap_size);
+        *((uint32_t*)wr1->get_checksum(&heap)) = crc32c(0, buffer_area.data(), 4096);
+        wr1->checksum = wr1->calc_checksum(&heap);
+        total_size += wr1->size;
+    };
+    auto add_simple = [&](blockstore_heap_t & heap, size_t & total_size, uint64_t type, uint64_t lsn, uint64_t version)
+    {
+        auto wr3 = (heap_entry_t*)(tmp.data() + total_size);
+        wr3->size = heap.get_simple_entry_size();
+        wr3->entry_type = type;
+        wr3->lsn = lsn;
+        wr3->inode = INODE_WITH_POOL(1, 1);
+        wr3->stripe = 0;
+        wr3->version = version;
+        wr3->checksum = wr3->calc_checksum(&heap);
+        total_size += wr3->size;
+    };
+    auto load = [&](blockstore_heap_t & heap, size_t & total_size)
+    {
+        *(uint16_t*)(tmp.data() + total_size) = dsk.meta_block_size - total_size;
+        *(uint16_t*)(tmp.data() + total_size + 2) = BS_HEAP_FREE_SPACE;
+        uint64_t entries_loaded;
+        heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded);
+        heap.finish_load();
+        bool done = heap.recheck_small_writes([&](bool, uint64_t, uint64_t, std::function<void(uint8_t*)> cb) {}, 1);
+        assert(done);
+        heap.finish_recheck();
+        auto mod = heap.get_recheck_modified_blocks();
+        assert(mod.size() == 0);
+    };
+
+    object_id oid = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
+
+    // Scenario A: commit over a rollback that erased everything
+    {
+        blockstore_heap_t heap(&dsk, buffer_area.data(), 10);
+
+        size_t total_size = 0;
+        add_big(heap, total_size, 1, 1);
+        add_simple(heap, total_size, BS_HEAP_ROLLBACK, 2, 0);
+        add_simple(heap, total_size, BS_HEAP_COMMIT, 3, 1);
+        load(heap, total_size);
+
+        auto obj = heap.read_entry(oid);
+        assert(obj && obj->type() == BS_HEAP_COMMIT);
+        int small_writes = 0;
+        auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr)
+        {
+            small_writes++;
+        });
+        // Nothing survived: the compaction is a delete, stamped at the commit's lsn
+        assert(compact_info.compact_lsn == 3);
+        assert(compact_info.compact_version == 0);
+        assert(compact_info.clean_wr == NULL);
+        assert(compact_info.do_delete);
+        assert(small_writes == 0);
+        // Apply the compaction the way the journal flusher would: the DELETE entry
+        // lands over the commit and the object is gone
+        uint32_t mblock = 999999;
+        int res = heap.add_compact(obj, compact_info.compact_version, compact_info.compact_lsn,
+            0, true, &mblock, NULL, NULL, NULL);
+        assert(res == 0);
+        heap.start_block_write(mblock);
+        heap.complete_block_write(mblock);
+        obj = heap.read_entry(oid);
+        assert(obj->type() == BS_HEAP_DELETE && obj->version == 0 && obj->lsn == 3);
+        compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr) {});
+        assert(compact_info.compact_lsn == 0);  // nothing left to compact
+    }
+
+    // Scenario B: commit over a rolled-back delete
+    {
+        blockstore_heap_t heap(&dsk, buffer_area.data(), 10);
+
+        size_t total_size = 0;
+        add_simple(heap, total_size, BS_HEAP_DELETE|BS_HEAP_STABLE, 1, 0);
+        add_big(heap, total_size, 2, 2);
+        add_simple(heap, total_size, BS_HEAP_ROLLBACK, 3, 1);
+        add_simple(heap, total_size, BS_HEAP_COMMIT, 4, 2);
+        load(heap, total_size);
+
+        auto obj = heap.read_entry(oid);
+        assert(obj && obj->type() == BS_HEAP_COMMIT);
+        auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr) {});
+        assert(compact_info.compact_lsn == 4);
+        assert(compact_info.compact_version == 0);
+        assert(compact_info.clean_wr == NULL);
+        assert(compact_info.do_delete);
+        uint32_t mblock = 999999;
+        assert(heap.add_compact(obj, compact_info.compact_version, compact_info.compact_lsn,
+            0, true, &mblock, NULL, NULL, NULL) == 0);
+        heap.start_block_write(mblock);
+        heap.complete_block_write(mblock);
+        obj = heap.read_entry(oid);
+        assert(obj->type() == BS_HEAP_DELETE && obj->version == 0 && obj->lsn == 4);
+        compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr) {});
+        assert(compact_info.compact_lsn == 0);
+    }
+
+    // Scenario C: commit over a dangling small_write without a big_write
+    {
+        blockstore_heap_t heap(&dsk, buffer_area.data(), 10);
+
+        size_t total_size = 0;
+        add_small(heap, total_size, 1, 1);
+        add_simple(heap, total_size, BS_HEAP_COMMIT, 2, 2);
+        load(heap, total_size);
+
+        auto obj = heap.read_entry(oid);
+        assert(obj && obj->type() == BS_HEAP_COMMIT);
+        auto compact_info = heap.iterate_compaction(obj, heap.get_fsynced_lsn(), false, [&](heap_entry_t *wr) {});
+        assert(compact_info.compact_lsn == 0);
+        assert(compact_info.compact_version == 0);
+        assert(compact_info.clean_wr == NULL);
+        assert(!compact_info.do_delete);
+    }
+
+    printf("test_iterate_compaction_commit_over_nothing passed\n");
+}
+
 void test_modify_bitmap()
 {
     blockstore_disk_t dsk;
@@ -2046,6 +2222,54 @@ void test_recheck_shadow_of_dropped_entry(bool reverse)
     printf("OK test_recheck_shadow_of_dropped_entry\n");
 }
 
+// The persisted completed_lsn may be larger than the lsn of the last entry which survived:
+// the entries above it were already covered by an fsynced journal record, but their metadata
+// never reached the disk. A new entry must never reuse such an lsn - on the next start it
+// would look already fsynced and escape the rollback of unfinished writes
+void test_start_lsn_below_completed()
+{
+    printf("test_start_lsn_below_completed\n");
+
+    blockstore_disk_t dsk;
+    _test_init(dsk, false);
+    std::vector<uint8_t> buffer_area(dsk.journal_device_size);
+    std::vector<uint8_t> tmp;
+
+    // Two stable big writes. The store then fsynced two more journal records (lsns up to 7)
+    // whose entries were lost
+    {
+        blockstore_heap_t heap(&dsk, buffer_area.data());
+        heap.finish_recheck();
+        _test_big_write(heap, dsk, 1, 0, 1, 0, true, 0, 8192, buffer_area.data());
+        _test_big_write(heap, dsk, 2, 0, 1, 0x20000, true, 0, 8192, buffer_area.data());
+        tmp.resize(dsk.meta_block_size);
+        heap.get_meta_block(0, tmp.data());
+    }
+
+    // Reload the way blockstore_init does it: start_load first, then read the meta blocks
+    {
+        blockstore_heap_t heap(&dsk, buffer_area.data());
+        heap.start_load(7);
+        uint64_t entries_loaded;
+        heap.load_blocks(0, dsk.meta_block_size, tmp.data(), false, entries_loaded);
+        heap.finish_load();
+        bool done = heap.recheck_small_writes([&](bool, uint64_t, uint64_t, std::function<void(uint8_t*)> cb) {}, 1);
+        assert(done);
+        heap.finish_recheck();
+        // Finish of the recheck must not roll completed_lsn back to the last surviving entry
+        assert(heap.get_completed_lsn() == 7);
+
+        // A new entry has to land above completed_lsn, otherwise it would look fsynced
+        _test_big_write(heap, dsk, 3, 0, 1, 0x40000, true, 0, 8192, buffer_area.data());
+        object_id oid = { .inode = INODE_WITH_POOL(1, 3), .stripe = 0 };
+        auto obj = heap.read_entry(oid);
+        assert(obj);
+        assert(obj->lsn == 8);
+    }
+
+    printf("...OK\n");
+}
+
 void test_corruption()
 {
     blockstore_disk_t dsk;
@@ -3597,6 +3821,7 @@ int main(int narg, char *args[])
     test_compact(false, true);
     test_compact(false, false);
     test_iterate_compaction();
+    test_iterate_compaction_commit_over_nothing();
     test_modify_bitmap();
     test_recheck(false, true);
     test_recheck(false, false);
@@ -3612,6 +3837,7 @@ int main(int narg, char *args[])
     test_recheck_shadow_of_dropped_entry(true);
     test_recheck_under_dropped_intent(false);
     test_recheck_under_dropped_intent(true);
+    test_start_lsn_below_completed();
     test_entry_placement_contract();
     test_corruption();
     test_full_overwrite(true);

src/test/test_msgr_kdf.cpp

@@ -0,0 +1,123 @@
+// Copyright (c) Alexey Shabalin, 2026+
+// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
+// I accept Vitastor CLA: see CLA-en.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;
+}