Release 2.4.4

- Implement `vitastor-kv rescue` command which dumps all readable blocks from the
  database regardless of the tree structure and allows to recover a corrupted database.
- Simplify primary OSD selection - use just PG number instead of the hash. This allows
  to distribute primary OSDs more evenly and reach better read speeds.
- Fix vitastor-kv-stress hang on parallel list & reopen.
- Fix CAS writes without immediate_commit. Fix is sufficient for vitastor-kv to
  operate correctly in non-immediate_commit pools.
- Fix NFS ACCESS calls for broken inodes (allow root access to them).
- Fix broken vitastor-kv-stress --runtime parameter.
- Fix bad vitastor-kv dump output buffering leading to incorrect JSON in dumps.
- Fix writing garbage from uninitialized memory to the empty space in VitastorKV DB.
- Refactor some code in preparation to the release of the new log-structured metadata store.
- Replace assert(done != expected) error message, probably caused by kernel issues with
  zero-copy TCP send, with a more concise error description.

.gitea/workflows/test.yml

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

CMakeLists.txt

@@ -2,6 +2,6 @@ cmake_minimum_required(VERSION 2.8.12)
 
 project(vitastor)
 
-set(VITASTOR_VERSION "2.4.3")
+set(VITASTOR_VERSION "2.4.4")
 
 add_subdirectory(src)

csi/Makefile

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v2.4.3
+VITASTOR_VERSION ?= v2.4.4
 
 all: build push
 

csi/deploy/004-csi-nodeplugin.yaml

@@ -49,7 +49,7 @@ spec:
             capabilities:
               add: ["SYS_ADMIN"]
             allowPrivilegeEscalation: true
-          image: vitalif/vitastor-csi:v2.4.3
+          image: vitalif/vitastor-csi:v2.4.4
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/deploy/007-csi-provisioner.yaml

@@ -121,7 +121,7 @@ spec:
             privileged: true
             capabilities:
               add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v2.4.3
+          image: vitalif/vitastor-csi:v2.4.4
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/src/config.go

@@ -5,7 +5,7 @@ package vitastor
 
 const (
     vitastorCSIDriverName    = "csi.vitastor.io"
-    vitastorCSIDriverVersion = "2.4.3"
+    vitastorCSIDriverVersion = "2.4.4"
 )
 
 // Config struct fills the parameters of request or user input

debian/changelog

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

docker/Makefile

@@ -1,4 +1,4 @@
-VITASTOR_VERSION ?= v2.4.3
+VITASTOR_VERSION ?= v2.4.4
 
 all: build push
 

docker/etc/vitastor/docker.conf

@@ -4,7 +4,7 @@
 #
 
 # Desired Vitastor version
-VITASTOR_VERSION=v2.4.3
+VITASTOR_VERSION=v2.4.4
 
 # 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:v2.4.3`
+   `docker pull vitalif/vitastor:v2.4.4`
 2. Install scripts to the host system: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.3 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.4 install.sh`
 3. Reload udev rules: \
    `udevadm control --reload-rules`
 

docs/installation/docker.ru.md

@@ -25,9 +25,9 @@ Vitastor можно установить в Docker/Podman. При этом etcd,
 Инструкция по установке максимально простая.
 
 1. Скачайте Docker-образ желаемой версии: \
-   `docker pull vitalif/vitastor:v2.4.3`
+   `docker pull vitalif/vitastor:v2.4.4`
 2. Установите скрипты в хост-систему командой: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.3 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v2.4.4 install.sh`
 3. Перезагрузите правила udev: \
    `udevadm control --reload-rules`
 

mon/package.json

@@ -1,6 +1,6 @@
 {
   "name": "vitastor-mon",
-  "version": "2.4.3",
+  "version": "2.4.4",
   "description": "Vitastor SDS monitor service",
   "main": "mon-main.js",
   "scripts": {

mon/pg_gen.js

@@ -9,7 +9,6 @@ const LPOptimizer = require('./lp_optimizer/lp_optimizer.js');
 const { scale_pg_count } = require('./pg_utils.js');
 const { make_hier_tree, filter_osds_by_root_node,
     filter_osds_by_tags, filter_osds_by_block_layout, get_affinity_osds } = require('./osd_tree.js');
-const { select_murmur3 } = require('./lp_optimizer/murmur3.js');
 
 function pick_primary(pool_id, pg_num, pool_config, osd_set, up_osds, aff_osds)
 {
@@ -39,7 +38,7 @@ function pick_primary(pool_id, pg_num, pool_config, osd_set, up_osds, aff_osds)
     {
         return 0;
     }
-    return alive_set[select_murmur3(alive_set.length, osd_num => pool_id+'/'+pg_num+'/'+osd_num)];
+    return alive_set[pg_num % alive_set.length];
 }
 
 function recheck_primary(state, global_config, up_osds, osd_tree)

node-binding/package.json

@@ -1,6 +1,6 @@
 {
   "name": "vitastor",
-  "version": "2.4.3",
+  "version": "2.4.4",
   "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 = '2.4.3'
+VITASTOR_VERSION = '2.4.4'
 
 LOG = logging.getLogger(__name__)
 

rpm/vitastor-el7.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        2.4.3
+Version:        2.4.4
 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-2.4.3.el7.tar.gz
+Source0:        vitastor-2.4.4.el7.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  devtoolset-9-gcc-c++

rpm/vitastor-el8.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        2.4.3
+Version:        2.4.4
 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-2.4.3.el8.tar.gz
+Source0:        vitastor-2.4.4.el8.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-toolset-9-gcc-c++

rpm/vitastor-el9.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        2.4.3
+Version:        2.4.4
 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-2.4.3.el9.tar.gz
+Source0:        vitastor-2.4.4.el9.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-c++

src/CMakeLists.txt

@@ -20,7 +20,7 @@ if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 	set(CMAKE_INSTALL_RPATH "${CMAKE_INSTALL_PREFIX}/${CMAKE_INSTALL_LIBDIR}")
 endif()
 
-add_definitions(-DVITASTOR_VERSION="2.4.3")
+add_definitions(-DVITASTOR_VERSION="2.4.4")
 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/client/cluster_client.cpp

@@ -765,8 +765,13 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
     {
         return;
     }
-    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && enable_writeback && !(op->flags & OP_FLUSH_BUFFER) &&
-        !op->version /* no CAS writeback */)
+    // CAS writes are simplified: they're not cached, not resliced, not retried, and not part of the regular write queue at all
+    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version)
+    {
+        execute_cas(op);
+        return;
+    }
+    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && enable_writeback && !(op->flags & OP_FLUSH_BUFFER))
     {
         if (wb->writebacks_active >= client_max_writeback_iodepth)
         {
@@ -788,7 +793,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
     }
     if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && !(op->flags & OP_IMMEDIATE_COMMIT))
     {
-        if (!(op->flags & OP_FLUSH_BUFFER) && !op->version /* no CAS write-repeat */)
+        if (!(op->flags & OP_FLUSH_BUFFER))
         {
             uint64_t flush_id = ++wb->last_flush_id;
             wb->copy_write(op, CACHE_REPEATING, flush_id);
@@ -847,6 +852,72 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
     }
 }
 
+void cluster_client_t::execute_cas(cluster_op_t *op)
+{
+    slice_rw(op);
+    op->needs_reslice = false;
+    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1)
+    {
+        // Atomic writes to multiple stripes are unsupported
+        op->retval = -EINVAL;
+        auto cb = std::move(op->callback);
+        cb(op);
+        return;
+    }
+    int res = try_send(op, 0, [this, op](osd_op_t *part)
+    {
+        int expected = part->req.hdr.opcode == OSD_OP_DELETE ? 0 : part->req.rw.len;
+        op->retval = part->reply.hdr.retval;
+        op->retval = op->retval == expected ? 0 : (op->retval >= 0 ? -EIO : op->retval);
+        op->retval = op->retval == -EPIPE ? -EINTR : op->retval;
+        auto peer_it = msgr.osd_peer_fds.find(op->parts[0].osd_num);
+        if (op->retval != 0 || (op->flags & OP_IMMEDIATE_COMMIT))
+        {
+            auto cb = std::move(op->callback);
+            cb(op);
+        }
+        else if (peer_it == msgr.osd_peer_fds.end())
+        {
+            // Care must be taken to make sure that the client doesn't reconnect to the OSD
+            // before executing the previously completed operation callback (!)
+            op->retval = -EINTR;
+            auto cb = std::move(op->callback);
+            cb(op);
+        }
+        else
+        {
+            // CAS writes have a built-in sync
+            auto peer_fd = peer_it->second;
+            *part = (osd_op_t){
+                .op_type = OSD_OP_OUT,
+                .peer_fd = peer_fd,
+                .req = {
+                    .hdr = {
+                        .magic = SECONDARY_OSD_OP_MAGIC,
+                        .opcode = OSD_OP_SYNC,
+                    },
+                },
+                .callback = [this, op](osd_op_t *part)
+                {
+                    op->retval = part->reply.hdr.retval;
+                    op->retval = op->retval == -EPIPE ? -EINTR : op->retval;
+                    auto cb = std::move(op->callback);
+                    cb(op);
+                },
+            };
+            msgr.outbox_push(part);
+        }
+    });
+    if (res == TRY_SEND_CONNECTING || res == TRY_SEND_OFFLINE)
+    {
+        // In theory, CAS writes could wait for the PG to come up, but it's easier to just fail it
+        op->retval = -EINTR;
+        auto cb = std::move(op->callback);
+        cb(op);
+        return;
+    }
+}
+
 bool cluster_client_t::check_rw(cluster_op_t *op)
 {
     if (op->opcode == OSD_OP_SYNC)
@@ -956,13 +1027,6 @@ resume_0:
     // Slice the operation into parts
     slice_rw(op);
     op->needs_reslice = false;
-    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1)
-    {
-        // Atomic writes to multiple stripes are unsupported
-        op->retval = -EINVAL;
-        erase_op(op);
-        return 1;
-    }
 resume_1:
     // Send unsent parts, if they're not subject to change
     op->state = 2;
@@ -1307,7 +1371,7 @@ bool cluster_client_t::affects_osd(uint64_t inode, uint64_t offset, uint64_t len
     return false;
 }
 
-int cluster_client_t::try_send(cluster_op_t *op, int i)
+int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_op_t *op_part)> cb)
 {
     if (!msgr_initialized)
     {
@@ -1367,7 +1431,7 @@ int cluster_client_t::try_send(cluster_op_t *op, int i)
                     ? (uint8_t*)op->part_bitmaps + pg_bitmap_size*i : NULL),
                 .bitmap_len = (unsigned)(op->opcode == OSD_OP_READ || op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP
                     ? pg_bitmap_size : 0),
-                .callback = [this, part](osd_op_t *op_part)
+                .callback = cb ? cb : [this, part](osd_op_t *op_part)
                 {
                     handle_op_part(part);
                 },

src/client/cluster_client.h

@@ -175,12 +175,13 @@ protected:
     void on_change_node_placement_hook();
 
     void execute_internal(cluster_op_t *op);
+    void execute_cas(cluster_op_t *op);
     void unshift_op(cluster_op_t *op);
     int continue_rw(cluster_op_t *op);
     bool check_rw(cluster_op_t *op);
     void slice_rw(cluster_op_t *op);
     void reset_retry_timer(int new_duration);
-    int try_send(cluster_op_t *op, int i);
+    int try_send(cluster_op_t *op, int i, std::function<void(osd_op_t *op_part)> cb = nullptr);
     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);

src/client/osd_ops.h

@@ -262,7 +262,10 @@ struct __attribute__((__packed__)) osd_reply_del_t
     uint32_t left_on_dead_count;
 };
 
-// sync to the primary OSD
+// sync to the primary OSD. semantics:
+// 1) any non-synced write may disappear on OSD restart. even if it's a CAS write.
+// 2) sync only guarantees to commit completed writes, not in-progress ones.
+// 3) sync is a no-op when immediate_commit is active.
 struct __attribute__((__packed__)) osd_op_sync_t
 {
     osd_op_header_t header;

src/client/vitastor.pc.in

@@ -6,7 +6,7 @@ includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@
 
 Name: Vitastor
 Description: Vitastor client library
-Version: 2.4.3
+Version: 2.4.4
 Libs: -L${libdir} -lvitastor_client
 Cflags: -I${includedir}
 

src/kv/kv_cli.cpp

@@ -97,6 +97,7 @@ void kv_cli_t::parse_args(int narg, const char *args[])
                 "  dump [<start> [end]]\n"
                 "  dumpjson [<start> [end]]\n"
                 "  loadjson\n"
+                "  rescue\n"
                 "\n"
                 "<IMAGE> should be the name of Vitastor image with the DB.\n"
                 "Without <COMMAND>, you get an interactive DB shell.\n"
@@ -298,6 +299,50 @@ struct kv_cli_list_t
     int n = 0;
     std::function<void(int)> cb;
 
+    void handle_key(int res, const std::string & key, const std::string & value)
+    {
+        if (res < 0)
+        {
+            if (res != -ENOENT)
+                fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
+            if (format == 2)
+                write("\n}\n");
+            if (handle)
+                db->list_close(handle);
+            flush();
+            cb(res == -ENOENT ? 0 : res);
+            delete this;
+        }
+        else
+        {
+            if (format == 2)
+            {
+                write(n ? ",\n  " : "{\n  ");
+                write(addslashes(key));
+                write(": ");
+                write(addslashes(value));
+            }
+            else if (format == 1)
+            {
+                write("set ");
+                write(auto_addslashes(key));
+                write(" ");
+                write(value);
+                write("\n");
+            }
+            else
+            {
+                write(key);
+                write(" = ");
+                write(value);
+                write("\n");
+            }
+            n++;
+            if (handle)
+                db->list_next(handle, NULL);
+        }
+    }
+
     void write(const std::string & str)
     {
         if (buf.capacity() < KV_LIST_BUF_SIZE)
@@ -316,6 +361,7 @@ struct kv_cli_list_t
             if (res > 0)
                 done += res;
         }
+        buf.clear();
     }
 };
 
@@ -624,44 +670,18 @@ void kv_cli_t::handle_cmd(const std::vector<std::string> & cmd, std::function<vo
         lst->cb = std::move(cb);
         db->list_next(lst->handle, [lst](int res, const std::string & key, const std::string & value)
         {
-            if (res < 0)
-            {
-                if (res != -ENOENT)
-                    fprintf(stderr, "Error: %s (code %d)\n", strerror(-res), res);
-                if (lst->format == 2)
-                    lst->write("\n}\n");
-                lst->flush();
-                lst->db->list_close(lst->handle);
-                lst->cb(res == -ENOENT ? 0 : res);
-                delete lst;
-            }
-            else
-            {
-                if (lst->format == 2)
-                {
-                    lst->write(lst->n ? ",\n  " : "{\n  ");
-                    lst->write(addslashes(key));
-                    lst->write(": ");
-                    lst->write(addslashes(value));
-                }
-                else if (lst->format == 1)
-                {
-                    lst->write("set ");
-                    lst->write(auto_addslashes(key));
-                    lst->write(" ");
-                    lst->write(value);
-                    lst->write("\n");
-                }
-                else
-                {
-                    lst->write(key);
-                    lst->write(" = ");
-                    lst->write(value);
-                    lst->write("\n");
-                }
-                lst->n++;
-                lst->db->list_next(lst->handle, NULL);
-            }
+            lst->handle_key(res, key, value);
+        });
+    }
+    else if (opname == "rescue")
+    {
+        kv_cli_list_t *lst = new kv_cli_list_t;
+        lst->db = db;
+        lst->format = 2;
+        lst->cb = std::move(cb);
+        db->rescue([lst](int res, const std::string & key, const std::string & value)
+        {
+            lst->handle_key(res, key, value);
         });
     }
     else if (opname == "loadjson")

src/kv/kv_db.cpp

@@ -96,7 +96,7 @@ struct kv_block_t
 
     void set_data_size();
     static int kv_size(const std::string & key, const std::string & value);
-    int parse(uint64_t offset, uint8_t *data, int size);
+    int parse(uint64_t offset, uint8_t *data, int size, bool allow_empty = false);
     bool serialize(uint8_t *data, int size);
     void apply_change();
     void cancel_change();
@@ -139,7 +139,6 @@ struct kv_db_t
     uint64_t next_free = 0;
     uint32_t kv_block_size = 0;
     uint32_t ino_block_size = 0;
-    bool immediate_commit = false;
     uint64_t memory_limit = 128*1024*1024;
     uint64_t evict_unused_age = 1000;
     uint64_t evict_max_misses = 10;
@@ -173,6 +172,7 @@ struct kv_db_t
     void open(inode_t inode_id, json11::Json cfg, std::function<void(int)> cb);
     void set_config(json11::Json cfg);
     void close(std::function<void()> cb);
+    void rescue(std::function<void(int res, const std::string & key, const std::string & value)> cb);
 
     void find_size(uint64_t min, uint64_t max, int phase, std::function<void(int, uint64_t)> cb);
     void run_continue_update(uint64_t offset);
@@ -243,13 +243,13 @@ static std::string read_string(uint8_t *data, int size, int *pos)
     return key;
 }
 
-int kv_block_t::parse(uint64_t offset, uint8_t *data, int size)
+int kv_block_t::parse(uint64_t offset, uint8_t *data, int size, bool allow_empty)
 {
     kv_stored_block_t *blk = (kv_stored_block_t *)data;
     if (blk->magic == 0 || blk->type == KV_EMPTY)
     {
         // empty block
-        if (offset != 0)
+        if (!allow_empty)
             fprintf(stderr, "K/V: Block %ju is %s\n", offset, blk->magic == 0 ? "empty" : "cleared");
         return -ENOTBLK;
     }
@@ -389,6 +389,10 @@ bool kv_block_t::serialize(uint8_t *buf, int size)
             return false;
         blk->items++;
     }
+    if (pos < size)
+    {
+        memset(buf+pos, 0, size-pos);
+    }
     return true;
 }
 
@@ -524,7 +528,6 @@ void kv_db_t::open(inode_t inode_id, json11::Json cfg, std::function<void(int)>
         return;
     }
     this->inode_id = inode_id;
-    this->immediate_commit = cli->get_immediate_commit(inode_id);
     this->ino_block_size = pool_cfg.data_block_size * pg_data_size;
     this->kv_block_size = kv_block_size;
     this->next_free = 0;
@@ -542,6 +545,127 @@ void kv_db_t::open(inode_t inode_id, json11::Json cfg, std::function<void(int)>
     });
 }
 
+struct kv_rescue_t
+{
+    kv_db_t *db = NULL;
+    uint64_t size = 0;
+    uint64_t pos = 0;
+    uint64_t cur_size = 0;
+    uint64_t cur_offset = 0;
+    int state = 0;
+    kv_block_t blk;
+    std::vector<uint8_t> buf;
+    std::function<void(int res, const std::string & key, const std::string & value)> cb;
+
+    void finish(int retval)
+    {
+        auto cb = std::move(this->cb);
+        cb(retval, "", "");
+        delete this;
+    }
+
+    void send_read()
+    {
+        if (pos >= size)
+        {
+            finish(-ENOENT);
+            return;
+        }
+        if (!buf.size())
+        {
+            buf.resize(1048576);
+        }
+        cluster_op_t *op = new cluster_op_t;
+        op->opcode = OSD_OP_READ;
+        op->inode = db->inode_id;
+        op->offset = pos;
+        cur_size = op->len = pos+buf.size() < size ? buf.size() : size-pos;
+        op->iov.push_back(buf.data(), cur_size);
+        op->callback = [=](cluster_op_t *op)
+        {
+            if (op->retval != op->len)
+            {
+                // error
+                finish(op->retval >= 0 ? -EIO : op->retval);
+                return;
+            }
+            state = 2;
+            cur_offset = 0;
+            run();
+            delete op;
+        };
+        db->cli->execute(op);
+        state = 1;
+    }
+
+    void parse_block()
+    {
+        if (cur_offset < cur_size)
+        {
+            blk = {};
+            int err = blk.parse(pos+cur_offset, buf.data()+cur_offset, db->kv_block_size, true);
+            if (err != 0)
+            {
+            }
+            else if (blk.type == KV_LEAF || blk.type == KV_LEAF_SPLIT)
+            {
+                for (auto it = blk.data.begin(); it != blk.data.end(); it++)
+                {
+                    cb(0, it->first, it->second);
+                }
+            }
+            cur_offset += db->kv_block_size;
+        }
+        else
+        {
+            pos += cur_size;
+            state = 0;
+        }
+    }
+
+    void run()
+    {
+        while (true)
+        {
+            if (state == 0)
+            {
+                send_read();
+                return;
+            }
+            else if (state == 1)
+            {
+                return;
+            }
+            else if (state == 2)
+            {
+                parse_block();
+            }
+        }
+    }
+};
+
+void kv_db_t::rescue(std::function<void(int res, const std::string & key, const std::string & value)> cb)
+{
+    if (!inode_id || closing)
+    {
+        cb(-EINVAL, "", "");
+        return;
+    }
+    find_size(0, 0, 1, [=](int res, uint64_t size)
+    {
+        if (res < 0)
+        {
+            cb(res, "", "");
+            return;
+        }
+        kv_rescue_t *st = new kv_rescue_t();
+        st->db = this;
+        st->size = size;
+        st->cb = cb;
+        st->run();
+    });
+}
+
 void kv_db_t::set_config(json11::Json cfg)
 {
     this->memory_limit = cfg["kv_memory_limit"].is_null() ? 128*1024*1024 : cfg["kv_memory_limit"].uint64_value();
@@ -563,7 +687,6 @@ void kv_db_t::close(std::function<void()> cb)
         next_free = 0;
         kv_block_size = 0;
         ino_block_size = 0;
-        immediate_commit = false;
         block_cache.clear();
         known_versions.clear();
         cb();
@@ -939,7 +1062,7 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
                 del_block_level(db, blk);
                 *blk = {};
             }
-            int err = blk->parse(op->offset, (uint8_t*)op->iov.buf[0].iov_base, op->len);
+            int err = blk->parse(op->offset, (uint8_t*)op->iov.buf[0].iov_base, op->len, op->offset == 0);
             if (err == 0)
             {
                 blk->level = cur_level;
@@ -1002,6 +1125,8 @@ kv_op_t::~kv_op_t()
     {
         done = true;
         db->active_ops--;
+        if (!db->active_ops && db->closing)
+            db->close(db->on_close);
     }
 }
 
@@ -1272,22 +1397,7 @@ static void write_block(kv_db_t *db, kv_block_t *blk, std::function<void(int)> c
             }
         }
         delete op;
-        if (res < 0 || db->immediate_commit)
-        {
-            cb(res);
-        }
-        else
-        {
-            op = new cluster_op_t;
-            op->opcode = OSD_OP_SYNC;
-            op->callback = [cb](cluster_op_t *op)
-            {
-                auto res = op->retval;
-                delete op;
-                cb(res);
-            };
-            db->cli->execute(op);
-        }
+        cb(res);
     };
     db->cli->execute(op);
 }
@@ -1662,6 +1772,7 @@ void kv_op_t::update_block(int path_pos, bool is_delete, const std::string & key
         {
             if (res < 0)
             {
+                blk->cancel_change();
                 auto blk_offset = blk->offset;
                 del_block_level(db, blk);
                 db->block_cache.erase(blk_offset);
@@ -1794,6 +1905,7 @@ void kv_op_t::update_block(int path_pos, bool is_delete, const std::string & key
             {
                 if (write_res < 0)
                 {
+                    blk->cancel_change();
                     auto blk_offset = blk->offset;
                     del_block_level(db, blk);
                     db->block_cache.erase(blk_offset);
@@ -1981,6 +2093,11 @@ void vitastorkv_dbw_t::close(std::function<void()> cb)
     db->close(cb);
 }
 
+void vitastorkv_dbw_t::rescue(std::function<void(int res, const std::string & key, const std::string & value)> cb)
+{
+    db->rescue(cb);
+}
+
 void vitastorkv_dbw_t::get(const std::string & key, std::function<void(int res, const std::string & value)> cb, bool cached)
 {
     auto *op = new kv_op_t;

src/kv/kv_stress.cpp

@@ -82,6 +82,7 @@ public:
     uint64_t total_prob = 0;
     uint64_t ops_sent = 0, ops_done = 0;
     int stat_timer_id = -1;
+    int run_timer_id = -1;
     int in_progress = 0;
     bool reopening = false;
     std::set<kv_test_listing_t*> listings;
@@ -299,9 +300,11 @@ void kv_test_t::run(json11::Json cfg)
     ringloop->register_consumer(&consumer);
     if (print_stats_interval)
         stat_timer_id = epmgr->tfd->set_timer(print_stats_interval*1000, true, [this](int) { print_stats(prev_stat, prev_stat_time); });
+    if (runtime_sec)
+        run_timer_id = epmgr->tfd->set_timer(runtime_sec*1000, false, [this](int) { run_timer_id = -1; op_count = 0; });
     clock_gettime(CLOCK_REALTIME, &start_stat_time);
     prev_stat_time = start_stat_time;
-    while (!finished)
+    while (!finished || in_progress > 0)
     {
         ringloop->loop();
         if (!finished)
@@ -309,6 +312,8 @@ void kv_test_t::run(json11::Json cfg)
     }
     if (stat_timer_id >= 0)
         epmgr->tfd->clear_timer(stat_timer_id);
+    if (run_timer_id >= 0)
+        epmgr->tfd->clear_timer(run_timer_id);
     ringloop->unregister_consumer(&consumer);
     // Print total stats
     print_total_stats();

src/kv/vitastor_kv.h

@@ -28,6 +28,7 @@ struct __attribute__((visibility("default"))) vitastorkv_dbw_t
     void open(uint64_t inode_id, std::map<std::string, std::string> cfg, std::function<void(int)> cb);
     void set_config(std::map<std::string, std::string> cfg);
     void close(std::function<void()> cb);
+    void rescue(std::function<void(int res, const std::string & key, const std::string & value)> cb);
 
     uint64_t get_size();
 

src/nfs/nfs_kv_getattr.cpp

@@ -166,7 +166,20 @@ int kv_nfs3_access_proc(void *opaque, rpc_op_t *rop)
             fprintf(stderr, "[%d] ACCESS %ju -> %s\n", self->nfs_fd, ino, value.c_str());
         if (res < 0)
         {
-            *reply = (ACCESS3res){ .status = vitastor_nfs_map_err(-res) };
+            // Broken inode (non-existing), allow only root access
+            if (!rop->auth_sys.uid)
+            {
+                *reply = (ACCESS3res){
+                    .status = NFS3_OK,
+                    .resok = (ACCESS3resok){
+                        .access = args->access,
+                    },
+                };
+            }
+            else
+            {
+                *reply = (ACCESS3res){ .status = vitastor_nfs_map_err(-res) };
+            }
         }
         else
         {

src/osd/osd_primary_write.cpp

@@ -165,7 +165,6 @@ resume_3:
         return;
     }
     // Check CAS version
-    // FIXME: Handle CAS writes as "immediate" in non-immediate_commit pools, otherwise CAS doesn't make sense
     if (cur_op->req.rw.version && op_data->fact_ver != (cur_op->req.rw.version-1))
     {
         deref_object_state(pg, &op_data->object_state, true);

tests/run_tests.sh

@@ -48,6 +48,9 @@ SCHEME=ec ./test_snapshot_chain.sh
 ./test_snapshot_down.sh
 SCHEME=ec ./test_snapshot_down.sh
 
+./test_kv_stress.sh
+IMMEDIATE_COMMIT=1 ./test_kv_stress.sh
+
 ./test_splitbrain.sh
 
 ./test_rebalance_verify.sh

tests/test_kv_stress.sh

@@ -0,0 +1,8 @@
+#!/bin/bash -ex
+
+PG_COUNT=16
+. `dirname $0`/run_3osds.sh
+
+build/src/kv/vitastor-kv-stress --etcd_address $ETCD_URL --pool_id 1 --inode_id 1 --runtime 30
+
+format_green OK

tests/test_nfs.sh

@@ -1,5 +1,7 @@
 #!/bin/bash -ex
 
+GLOBAL_CONFIG=',"client_enable_writeback":false'
+IMMEDIATE_COMMIT=1
 PG_COUNT=16
 . `dirname $0`/run_3osds.sh
 
@@ -177,7 +179,7 @@ build/src/kv/vitastor-kv --etcd_address $ETCD_URL fsmeta set d11/settings.jsonLG
 sudo mount localhost:/ ./testdata/nfs -o port=2050,mountport=2050,nfsvers=3,soft,nolock,tcp
 ls -l ./testdata/nfs
 ls -l ./testdata/nfs/settings.jsonLGNmGn
-rm ./testdata/nfs/settings.jsonLGNmGn
+sudo rm ./testdata/nfs/settings.jsonLGNmGn
 build/src/kv/vitastor-kv --etcd_address $ETCD_URL fsmeta get d11/settings.jsonLGNmGn 2>&1 | grep '(code -2)'
 ls -l ./testdata/nfs
 
@@ -187,7 +189,7 @@ build/src/kv/vitastor-kv --etcd_address $ETCD_URL fsmeta set d11/settings.jsonLG
 sudo mount localhost:/ ./testdata/nfs -o port=2050,mountport=2050,nfsvers=3,soft,nolock,tcp
 ls -l ./testdata/nfs
 ls -l ./testdata/nfs/settings.jsonLGNmGn
-rm ./testdata/nfs/settings.jsonLGNmGn
+sudo rm ./testdata/nfs/settings.jsonLGNmGn
 build/src/kv/vitastor-kv --etcd_address $ETCD_URL fsmeta get d11/settings.jsonLGNmGn 2>&1 | grep '(code -2)'
 ls -l ./testdata/nfs