Repeer PG on immediate stabilize/rollback errors

...to prevent possible missed events in send_replies or read_requests on stop_client

csi/src/controllerserver.go

@@ -27,7 +27,6 @@ const (
     MB int64 = 1024 * KB
     GB int64 = 1024 * MB
     TB int64 = 1024 * GB
-    ETCD_TIMEOUT time.Duration = 15*time.Second
 )
 
 type InodeIndex struct
@@ -687,7 +686,7 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 
     return &csi.ControllerExpandVolumeResponse{
         CapacityBytes: int64(inodeCfg[0].Size),
-        NodeExpansionRequired: false,
+        NodeExpansionRequired: true,
     }, nil
 }
 

csi/src/nodeserver.go

@@ -1030,10 +1030,131 @@ func (ns *NodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVo
     return nil, status.Error(codes.Unimplemented, "")
 }
 
-// NodeExpandVolume expanding the file system on the node
+// NodeExpandVolume expands the block device and, for filesystem volumes, the filesystem on the node
 func (ns *NodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error)
 {
-    return nil, status.Error(codes.Unimplemented, "")
+    klog.Infof("received node expand volume request %+v", protosanitizer.StripSecrets(req))
+    if (req == nil)
+    {
+        return nil, status.Error(codes.InvalidArgument, "request cannot be empty")
+    }
+    if (req.VolumeId == "")
+    {
+        return nil, status.Error(codes.InvalidArgument, "VolumeId is required")
+    }
+
+    ctxVars := make(map[string]string)
+    err := json.Unmarshal([]byte(req.VolumeId), &ctxVars)
+    if (err != nil)
+    {
+        return nil, status.Error(codes.Internal, "volume ID not in JSON format")
+    }
+    volName := ctxVars["name"]
+
+    if (ctxVars["vitastorfs"] != "")
+    {
+        // VitastorFS CSI volumes are NFS-backed and currently don't have per-volume quotas to resize here.
+        return &csi.NodeExpandVolumeResponse{CapacityBytes: req.GetCapacityRange().GetRequiredBytes()}, nil
+    }
+
+    ns.lockVolume(ctxVars["configPath"]+":block:"+volName)
+    defer ns.unlockVolume(ctxVars["configPath"]+":block:"+volName)
+
+    targetPath := req.GetStagingTargetPath()
+    if (targetPath == "")
+    {
+        targetPath = req.GetVolumePath()
+    }
+    if (targetPath == "")
+    {
+        return nil, status.Error(codes.InvalidArgument, "VolumePath or StagingTargetPath is required")
+    }
+
+    devicePath, err := GetDeviceNameFromMount(targetPath)
+    if (err != nil)
+    {
+        if (os.IsNotExist(err))
+        {
+            return nil, status.Error(codes.NotFound, "Target path not found")
+        }
+        return nil, err
+    }
+    if (devicePath == "")
+    {
+        return nil, status.Errorf(codes.NotFound, "%s is not a mountpoint", targetPath)
+    }
+    if (devicePath[0] != '/')
+    {
+        return nil, status.Errorf(codes.Internal, "failed to resolve device for %s: %s", targetPath, devicePath)
+    }
+    if (len(devicePath) < 5 || devicePath[0:5] != "/dev/")
+    {
+        // NodeExpandVolume may be called with the published target path. For block volumes it is
+        // usually a bind mount of the staging target, so resolve one more mount layer if needed.
+        var realDevicePath string
+        realDevicePath, err = GetDeviceNameFromMount(devicePath)
+        if (err == nil && realDevicePath != "")
+        {
+            devicePath = realDevicePath
+        }
+    }
+
+    // All newest Vitastor driver versions (nbd, ublk and vduse) support online resize.
+    // So all we need to do is just wait a bit until the device is resized by the driver.
+    requiredBytes := req.GetCapacityRange().GetRequiredBytes()
+    if (requiredBytes > 0)
+    {
+        timeout := 30
+        var actualSize uint64
+        var err error
+        for i := 0; i < timeout && actualSize < requiredBytes; i++
+        {
+            actualSize, err = GetBlockDeviceSize(devicePath)
+            if (err != nil)
+            {
+                return nil, err
+            }
+            time.Sleep(1 * time.Second)
+        }
+        if (actualSize < requiredBytes)
+        {
+            return nil, status.Error(codes.NotFound, "Failed to wait for online resize of %s to %v bytes: actual size is still %v after %v seconds",
+                devicePath, requiredBytes, actualSize, timeout)
+        }
+    }
+
+    isBlock := req.GetVolumeCapability().GetBlock() != nil
+    if (isBlock)
+    {
+        return &csi.NodeExpandVolumeResponse{CapacityBytes: requiredBytes}, nil
+    }
+
+    diskMounter := &mount.SafeFormatAndMount{Interface: ns.mounter, Exec: utilexec.New()}
+    fsType := req.GetVolumeCapability().GetMount().GetFsType()
+    if (fsType == "")
+    {
+        fsType, err = diskMounter.GetDiskFormat(devicePath)
+        if (err != nil)
+        {
+            return nil, err
+        }
+    }
+
+    switch (fsType)
+    {
+        case "ext4":
+            _, err = systemCombined("resize2fs", devicePath)
+        case "xfs":
+            _, err = systemCombined("xfs_growfs", targetPath)
+        default:
+            return nil, status.Errorf(codes.InvalidArgument, "unsupported filesystem for resize: %s", fsType)
+    }
+    if (err != nil)
+    {
+        return nil, err
+    }
+
+    return &csi.NodeExpandVolumeResponse{CapacityBytes: requiredBytes}, nil
 }
 
 // NodeGetCapabilities returns the supported capabilities of the node server
@@ -1048,6 +1169,13 @@ func (ns *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetC
                     },
                 },
             },
+            &csi.NodeServiceCapability{
+                Type: &csi.NodeServiceCapability_Rpc{
+                    Rpc: &csi.NodeServiceCapability_RPC{
+                        Type: csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
+                    },
+                },
+            },
         },
     }, nil
 }

csi/src/utils.go

@@ -20,6 +20,7 @@ import (
 
     "google.golang.org/grpc/codes"
     "google.golang.org/grpc/status"
+    "golang.org/x/sys/unix"
 )
 
 type MountMethod int
@@ -519,3 +520,22 @@ func GetDeviceNameFromMount(mountPath string) (string, error)
 
     return device, nil
 }
+
+func GetBlockDeviceSize(devicePath string) (uint64, error)
+{
+    file, err := os.Open(devicePath)
+    if (err != nil)
+    {
+        return 0, err
+    }
+    defer file.Close()
+
+    // BLKGETSIZE64 returns the size in bytes as a uint64/int64
+    size, err := unix.IoctlGetInt(int(file.Fd()), unix.BLKGETSIZE64)
+    if (err != nil)
+    {
+        return 0, fmt.Errorf("ioctl BLKGETSIZE64 failed: %w", err)
+    }
+
+    return uint64(size), nil
+}

mon/etcd_schema.js

@@ -65,6 +65,8 @@ const etcd_tree = {
             placement_levels: { datacenter: 1, rack: 2, host: 3, osd: 4, ... },
             use_old_pg_combinator: false,
             osd_backfillfull_ratio: 0.99,
+            mon_metrics_path: '/prometheus',
+            mon_healthcheck_path: '/healthcheck',
             // client and osd
             tcp_header_buffer_size: 65536,
             use_sync_send_recv: false,

mon/mon.js

@@ -87,7 +87,7 @@ class Mon
             this.http = await create_http_server(this.config, (req, res) =>
             {
                 const u = new URL(req.url, 'http://'+(req.headers.host || 'localhost'));
-                if (u.pathname.replace(/\/+$/, '') == (this.config.prometheus_path||'/metrics'))
+                if (u.pathname.replace(/\/+$/, '') == (this.config.mon_metrics_path||'/metrics'))
                 {
                     if (!this.watcher_active)
                     {
@@ -100,6 +100,19 @@ class Mon
                         res.write(export_prometheus_metrics(this.state));
                     }
                 }
+                else if (u.pathname.replace(/\/+$/, '') == (this.config.mon_healthcheck_path||'/healthcheck'))
+                {
+                    if (!this.watcher_active)
+                    {
+                        res.writeHead(503);
+                        res.write('Monitor is in standby mode. Please retrieve metrics from master monitor instance\n');
+                    }
+                    else
+                    {
+                        res.setHeader('Content-Type', 'text/plain; version=0.0.4; charset=utf-8');
+                        res.write('Monitor is active');
+                    }
+                }
                 else
                 {
                     res.writeHead(404);

src/CMakeLists.txt

@@ -11,6 +11,7 @@ set(WITH_QEMU false CACHE BOOL "Build QEMU driver inside Vitastor source tree")
 set(WITH_FIO true CACHE BOOL "Build FIO driver")
 set(QEMU_PLUGINDIR qemu CACHE STRING "QEMU plugin directory suffix (qemu-kvm on RHEL)")
 set(WITH_ASAN false CACHE BOOL "Build with AddressSanitizer")
+set(WITH_UBSAN false CACHE BOOL "Build with UBSanitizer")
 set(WITH_SYSTEM_LIBURING false CACHE BOOL "Use system liburing")
 if("${CMAKE_INSTALL_PREFIX}" MATCHES "^/usr/local/?$")
 	if(EXISTS "/etc/debian_version")
@@ -27,6 +28,10 @@ if (${WITH_ASAN})
 	add_definitions(-fsanitize=address)
 	add_link_options(-fsanitize=address -fno-omit-frame-pointer)
 endif (${WITH_ASAN})
+if (${WITH_UBSAN})
+	add_definitions(-fsanitize=undefined -fno-sanitize=alignment)
+	add_link_options(-fsanitize=undefined -fno-sanitize=alignment)
+endif (${WITH_UBSAN})
 set(CMAKE_CXX_FLAGS_RELEASE "${CMAKE_CXX_FLAGS_RELEASE} -fvisibility-inlines-hidden")
 set(CMAKE_CXX_FLAGS_MINSIZEREL "${CMAKE_CXX_FLAGS_MINSIZEREL} -fvisibility-inlines-hidden")
 set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "${CMAKE_CXX_FLAGS_RELWITHDEBINFO} -fvisibility-inlines-hidden")

src/blockstore/blockstore.h

@@ -77,8 +77,11 @@ Output:
 - retval = number of bytes actually read/written or negative error number
   -EINVAL = invalid input parameters
   -ENOENT = requested object/version does not exist for reads
+  -EEXIST = requested version to write already exists
   -ENOSPC = no space left in the store for writes
-  -EDOM = checksum error.
+  -EAGAIN = temporary failure, operation should be retried (notably, writes may be retried
+    after stabilizing some journaled entries and freeing up journal space)
+  -EDOM = checksum error
 - version = the version actually read or written
 
 ## BS_OP_DELETE
@@ -90,7 +93,7 @@ Input:
 - version = requested version. Treated the same as with BS_OP_WRITE
 
 Output:
-- retval = 0 or negative error number (-EINVAL)
+- retval = 0 or negative error number (-EINVAL, -EAGAIN)
 - version = the version actually written (delete is initially written as an object version)
 
 ## BS_OP_SYNC
@@ -99,7 +102,7 @@ Make sure all previously issued modifications reach physical media.
 
 Input: Nothing except opcode
 Output:
-- retval = 0 or negative error number (-EINVAL)
+- retval = 0 or negative error number (-EINVAL, -EAGAIN)
 
 ## BS_OP_STABLE / BS_OP_ROLLBACK
 
@@ -129,7 +132,9 @@ Input:
 - max_oid = max inode/stripe or 0 to list all objects
 
 Output:
-- retval = total obj_ver_id count
+- retval = total obj_ver_id count or negative error number
+  -ENOMEM = could not allocate memory for the listing
+  -EAGAIN = pool sharding settings are incorrect, please reshard and retry
 - version = stable obj_ver_id count
 - buf = obj_ver_id array allocated by the blockstore. Stable versions come first.
   You must free it yourself after usage with free().

src/blockstore/blockstore_disk.cpp

@@ -210,6 +210,32 @@ void blockstore_disk_t::parse_config(std::map<std::string, std::string> & config
     }
 }
 
+// A metadata entry must fit into one metadata block; a large block_size with a small
+// csum_block_size otherwise inflates the entry past it and makes the callers divide by zero.
+static uint64_t entries_per_meta_block(uint64_t entry_size, uint64_t meta_block_size,
+    uint64_t data_block_size, uint32_t bitmap_granularity, uint32_t csum_block_size)
+{
+    if (entry_size && entry_size <= meta_block_size)
+    {
+        return meta_block_size / entry_size;
+    }
+    std::string msg = "Metadata entry size ("+std::to_string(entry_size)+" bytes) exceeds meta_block_size ("+
+        std::to_string(meta_block_size)+" bytes) with block_size="+std::to_string(data_block_size)+
+        ", bitmap_granularity="+std::to_string(bitmap_granularity);
+    if (csum_block_size)
+    {
+        msg += ", csum_block_size="+std::to_string(csum_block_size)+" ("+
+            std::to_string(data_block_size/csum_block_size)+" checksums per block)";
+    }
+    msg += ". Decrease block_size";
+    if (csum_block_size)
+    {
+        msg += ", increase csum_block_size";
+    }
+    msg += " or increase meta_block_size";
+    throw std::runtime_error(msg);
+}
+
 void blockstore_disk_t::calc_lengths(bool skip_meta_check)
 {
     // data
@@ -259,17 +285,20 @@ void blockstore_disk_t::calc_lengths(bool skip_meta_check)
     clean_entry_bitmap_size = (data_block_size / bitmap_granularity + 7) / 8;
     clean_dyn_size = clean_entry_bitmap_size*2 + (csum_block_size
         ? data_block_size/csum_block_size*(data_csum_type & 0xFF) : 0);
+    uint64_t entries_per_block = 0;
 recalc:
     if (meta_format == BLOCKSTORE_META_FORMAT_HEAP)
     {
-        uint32_t entries_per_block = meta_block_size / (sizeof(heap_big_write_t) + clean_dyn_size);
+        entries_per_block = entries_per_meta_block(sizeof(heap_big_write_t) + clean_dyn_size,
+            meta_block_size, data_block_size, bitmap_granularity, csum_block_size);
         min_meta_len = (block_count+entries_per_block-1) / entries_per_block * meta_block_size;
     }
     else if (meta_format == BLOCKSTORE_META_FORMAT_V1)
     {
         clean_entry_size = 24 /*sizeof(clean_disk_entry)*/ + 2*clean_entry_bitmap_size;
-        min_meta_len = (1 + (block_count - 1 + meta_block_size / clean_entry_size)
-            / (meta_block_size / clean_entry_size)) * meta_block_size;
+        entries_per_block = entries_per_meta_block(clean_entry_size, meta_block_size,
+            data_block_size, bitmap_granularity, 0);
+        min_meta_len = (1 + (block_count - 1 + entries_per_block) / entries_per_block) * meta_block_size;
         if (!skip_meta_check && meta_area_size < min_meta_len)
         {
 too_small:
@@ -281,7 +310,9 @@ too_small:
     {
         meta_format = BLOCKSTORE_META_FORMAT_V2;
         clean_entry_size = 24 /*sizeof(clean_disk_entry)*/ + clean_dyn_size + 4 /*entry_csum*/;
-        min_meta_len = (1 + (block_count - 1 + meta_block_size / clean_entry_size) / (meta_block_size / clean_entry_size)) * meta_block_size;
+        entries_per_block = entries_per_meta_block(clean_entry_size, meta_block_size,
+            data_block_size, bitmap_granularity, csum_block_size);
+        min_meta_len = (1 + (block_count - 1 + entries_per_block) / entries_per_block) * meta_block_size;
         if (!skip_meta_check && meta_area_size < min_meta_len)
         {
             if (!data_csum_type)

src/blockstore/blockstore_read.cpp

@@ -134,7 +134,8 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *op)
         }
         else if ((vec.copy_flags & COPY_BUF_JOURNAL) && dsk.inmemory_journal)
         {
-            memcpy(op->buf + vec.offset - op->offset, buffer_area + vec.disk_loc + vec.disk_offset, vec.len);
+            // disk_loc may be negative - make ubsan happy
+            memcpy(op->buf + vec.offset - op->offset, buffer_area + (vec.disk_loc + vec.disk_offset), vec.len);
         }
         else
         {

src/blockstore/blockstore_write.cpp

@@ -243,6 +243,12 @@ enospc:
         uint64_t loc = !op->len ? 0 : heap->find_free_buffer_area(op->len);
         if (loc == UINT64_MAX)
         {
+            if (!heap->get_to_compact_count())
+            {
+                op->retval = -EAGAIN;
+                FINISH_OP(op);
+                return 2;
+            }
             PRIV(op)->wait_for = WAIT_COMPACTION;
             PRIV(op)->wait_detail = heap->get_compacted_count();
             flusher->request_trim();

src/blockstore/v1/flush.cpp

@@ -76,6 +76,16 @@ size_t journal_flusher_t::get_queue_size()
     return flush_queue.size();
 }
 
+bool journal_flusher_t::may_advance()
+{
+    return flush_queue.size() || active_flushers > 0 || trim_possible;
+}
+
+bool journal_flusher_t::may_advance_except_one()
+{
+    return flush_queue.size() || active_flushers > 1 || trim_possible;
+}
+
 void journal_flusher_t::loop()
 {
     target_flusher_count = bs->write_iodepth*2;
@@ -190,6 +200,7 @@ void journal_flusher_t::request_trim()
 
 void journal_flusher_t::mark_trim_possible()
 {
+    trim_possible = true;
     if (trim_wanted > 0)
     {
         dequeuing = true;
@@ -652,9 +663,10 @@ resume_2:
         flusher->sync_to_repeat.erase(repeat_it);
     trim_journal:
         // Clear unused part of the journal every <journal_trim_interval> flushes
-        if (bs->journal_trim_interval && !((++flusher->journal_trim_counter) % bs->journal_trim_interval) ||
-            flusher->trim_wanted > 0)
+        if ((bs->journal_trim_interval && !((++flusher->journal_trim_counter) % bs->journal_trim_interval) ||
+            flusher->trim_wanted > 0) && flusher->trim_possible)
         {
+            flusher->trim_possible = false;
     resume_30:
     resume_31:
     resume_32:
@@ -1496,10 +1508,19 @@ bool journal_flusher_co::trim_journal(int wait_base)
                 printf("Journal flushed\n");
                 exit(0);
             }
+            bs->wakeup_wait_journal();
+        }
+        else if (!flusher->may_advance_except_one())
+        {
+            bs->wakeup_wait_journal();
         }
         flusher->journal_trim_counter = 0;
         flusher->trimming = false;
     }
+    else if (!flusher->may_advance_except_one())
+    {
+        bs->wakeup_wait_journal();
+    }
     return true;
 }
 

src/blockstore/v1/flush.h

@@ -94,6 +94,7 @@ public:
 class journal_flusher_t
 {
     int trim_wanted = 0;
+    bool trim_possible = false;
     bool dequeuing;
     int min_flusher_count, max_flusher_count, cur_flusher_count, target_flusher_count;
     int flusher_start_threshold;
@@ -117,12 +118,14 @@ class journal_flusher_t
 
     bool try_find_older(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
     bool try_find_other(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
+    bool may_advance_except_one();
 
 public:
     journal_flusher_t(blockstore_impl_t *bs);
     ~journal_flusher_t();
     void loop();
     bool is_trim_wanted() { return trim_wanted; }
+    bool may_advance();
     bool is_active();
     size_t get_queue_size();
     void mark_trim_possible();

src/blockstore/v1/impl.cpp

@@ -24,6 +24,7 @@ blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i *
         calc_lengths();
         alloc_dyn_data = dsk.clean_dyn_size > sizeof(void*) || dsk.csum_block_size > 0;
         zero_object = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.data_block_size);
+        memset(zero_object, 0, dsk.data_block_size);
         data_alloc = new allocator_t(dsk.block_count);
     }
     catch (std::exception & e)
@@ -211,10 +212,6 @@ void blockstore_impl_t::loop()
                     // ring is full, stop submission
                     break;
                 }
-                else if (PRIV(op)->wait_for == WAIT_JOURNAL)
-                {
-                    PRIV(op)->wait_detail2 = (unstable_writes.size()+unstable_unsynced);
-                }
             }
         }
         if (op_idx != new_idx)
@@ -279,6 +276,12 @@ bool blockstore_impl_t::is_safe_to_stop()
     return true;
 }
 
+void blockstore_impl_t::wakeup_wait_journal()
+{
+    wait_journal_counter++;
+    ringloop->wakeup();
+}
+
 void blockstore_impl_t::check_wait(blockstore_op_t *op)
 {
     if (PRIV(op)->wait_for == WAIT_SQE)
@@ -295,12 +298,11 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
     }
     else if (PRIV(op)->wait_for == WAIT_JOURNAL)
     {
-        if (journal.used_start == PRIV(op)->wait_detail &&
-            (unstable_writes.size()+unstable_unsynced) == PRIV(op)->wait_detail2)
+        if (wait_journal_counter == PRIV(op)->wait_detail)
         {
             // do not submit
 #ifdef BLOCKSTORE_DEBUG
-            printf("Still waiting to flush journal offset %08jx\n", PRIV(op)->wait_detail);
+            printf("Still waiting for flush journal counter %ju\n", PRIV(op)->wait_detail);
 #endif
             return;
         }

src/blockstore/v1/impl.h

@@ -93,7 +93,7 @@ struct blockstore_op_private_t
 {
     // Wait status
     int wait_for;
-    uint64_t wait_detail, wait_detail2;
+    uint64_t wait_detail;
     int pending_ops;
     int op_state;
 
@@ -215,6 +215,8 @@ class blockstore_impl_t: public blockstore_i
     blockstore_init_meta* metadata_init_reader;
     blockstore_init_journal* journal_init_reader;
 
+    uint64_t wait_journal_counter = 0;
+    void wakeup_wait_journal();
     void check_wait(blockstore_op_t *op);
     void init_op(blockstore_op_t *op);
 
@@ -238,7 +240,7 @@ class blockstore_impl_t: public blockstore_i
         uint64_t offset, uint64_t submit_len, uint64_t & blk_begin, uint64_t & blk_end, uint8_t* & blk_buf);
     bool read_range_fulfilled(std::vector<copy_buffer_t> & rv, uint64_t & fulfilled, uint8_t *read_buf,
         uint8_t *clean_entry_bitmap, uint32_t item_start, uint32_t item_end);
-    bool read_checksum_block(blockstore_op_t *op, int rv_pos, uint64_t &fulfilled, uint64_t clean_loc);
+    bool read_checksum_block(blockstore_op_t *op, int rv_pos, uint64_t &fulfilled, uint64_t clean_loc, bool padded);
     uint8_t* read_clean_meta_block(blockstore_op_t *read_op, uint64_t clean_loc, int rv_pos);
     bool verify_padded_checksums(uint8_t *clean_entry_bitmap, uint8_t *csum_buf, uint32_t offset,
         iovec *iov, int n_iov, std::function<void(uint32_t, uint32_t, uint32_t)> bad_block_cb);

src/blockstore/v1/init.cpp

@@ -1000,7 +1000,7 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                         memcpy((uint8_t*)dyn+sizeof(int), dyn_from, dyn_size);
                     }
                     bs->dirty_db.emplace(ov, (dirty_entry){
-                        .state = (BS_ST_SMALL_WRITE | BS_ST_SYNCED),
+                        .state = (BS_ST_SMALL_WRITE | BS_ST_SYNCED | (je->type == JE_SMALL_WRITE_INSTANT ? BS_ST_INSTANT : 0)),
                         .flags = 0,
                         .location = location,
                         .offset = je->small_write.offset,
@@ -1084,7 +1084,7 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                         memcpy((uint8_t*)dyn+sizeof(int), dyn_from, dyn_size);
                     }
                     auto dirty_it = bs->dirty_db.emplace(ov, (dirty_entry){
-                        .state = (BS_ST_BIG_WRITE | BS_ST_SYNCED),
+                        .state = (BS_ST_BIG_WRITE | BS_ST_SYNCED | (je->type == JE_BIG_WRITE_INSTANT ? BS_ST_INSTANT : 0)),
                         .flags = 0,
                         .location = je->big_write.location,
                         .offset = je->big_write.offset,
@@ -1171,6 +1171,12 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                     clean_it->second.version < je->del.version);
                 if (!clean_exists && dirty_exists)
                 {
+                    // Stabilize previous entries to fix inode_space_stats and used_blocks (FIXME - remove it)
+                    obj_ver_id ov = {
+                        .oid = je->del.oid,
+                        .version = dirty_it->first.version,
+                    };
+                    bs->mark_stable(ov, true);
                     // Clean entry doesn't exist. This means that the delete is already flushed.
                     // So we must not flush this object anymore.
                     erase_dirty_object(dirty_it);

src/blockstore/v1/journal.cpp

@@ -15,6 +15,7 @@ blockstore_journal_check_t::blockstore_journal_check_t(blockstore_impl_t *bs)
     first_sector = -1;
     next_in_pos = bs->journal.in_sector_pos;
     right_dir = next_pos >= bs->journal.used_start;
+    give_up = false;
 }
 
 // Check if we can write <required> entries of <size> bytes and <data_after> data bytes after them to the journal
@@ -119,13 +120,20 @@ int blockstore_journal_check_t::check_available(blockstore_op_t *op, int entries
         if (bs->log_level > 5)
         {
             printf(
-                "Ran out of journal space (used_start=%08jx, next_free=%08jx, dirty_start=%08jx)\n",
-                bs->journal.used_start, bs->journal.next_free, bs->journal.dirty_start
+                "Ran out of journal space (used_start=%08jx, next_free=%08jx, dirty_start=%08jx, queued %zu)\n",
+                bs->journal.used_start, bs->journal.next_free, bs->journal.dirty_start, bs->flusher->get_queue_size()
             );
         }
-        PRIV(op)->wait_for = WAIT_JOURNAL;
-        bs->flusher->request_trim();
-        PRIV(op)->wait_detail = bs->journal.used_start;
+        if (!bs->flusher->may_advance())
+        {
+            give_up = true;
+        }
+        else
+        {
+            PRIV(op)->wait_for = WAIT_JOURNAL;
+            PRIV(op)->wait_detail = bs->wait_journal_counter;
+            bs->flusher->request_trim();
+        }
         return 0;
     }
     return 1;

src/blockstore/v1/journal.h

@@ -70,6 +70,7 @@ struct blockstore_journal_check_t
     uint64_t next_pos, next_sector, next_in_pos;
     int sectors_to_write, first_sector;
     bool right_dir; // writing to the end or the beginning of the ring buffer
+    bool give_up;
 
     blockstore_journal_check_t(blockstore_impl_t *bs);
     int check_available(blockstore_op_t *op, int required, int size, int data_after);

src/blockstore/v1/read.cpp

@@ -110,7 +110,7 @@ int blockstore_impl_t::fulfill_read(blockstore_op_t *read_op,
                     (*dyn_data)++;
                 }
                 // Submit the journal checksum block read
-                if (!read_checksum_block(read_op, 1, fulfilled, item_location - item_start))
+                if (!read_checksum_block(read_op, 1, fulfilled, item_location - item_start, false))
                 {
                     r = 0;
                 }
@@ -228,10 +228,12 @@ int blockstore_impl_t::fill_partial_checksum_blocks(std::vector<copy_buffer_t> &
             }
             end_block++;
             // OK, mark this range as required
+            uint32_t rv_start = start_block*dsk.csum_block_size < item_start ? item_start : start_block*dsk.csum_block_size;
+            uint32_t rv_end = end_block*dsk.csum_block_size > item_end ? item_end : end_block*dsk.csum_block_size;
             rv.push_back((copy_buffer_t){
                 .copy_flags = COPY_BUF_CSUM_FILL | (from_journal ? COPY_BUF_JOURNALED_BIG : 0),
-                .offset = start_block*dsk.csum_block_size,
-                .len = (end_block-start_block)*dsk.csum_block_size,
+                .offset = rv_start,
+                .len = rv_end - rv_start,
                 // save checksum reference if we're reading clean data from the journal
                 .csum_buf = from_journal
                     ? clean_entry_bitmap + dsk.clean_entry_bitmap_size + (start_block-item_start_block)*(dsk.data_csum_type & 0xFF)
@@ -300,7 +302,7 @@ bool blockstore_impl_t::read_range_fulfilled(std::vector<copy_buffer_t> & rv, ui
     return all_done;
 }
 
-bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uint64_t &fulfilled, uint64_t clean_loc)
+bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uint64_t &fulfilled, uint64_t clean_loc, bool padded)
 {
     auto & rv = PRIV(op)->read_vec;
     auto *vi = &rv[rv.size()-rv_pos];
@@ -335,10 +337,22 @@ bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uin
         }
         return 0;
     });
+    if (padded && (item_start % dsk.csum_block_size))
+    {
+        n_iov++;
+    }
+    if (padded && (item_end % dsk.csum_block_size))
+    {
+        n_iov++;
+    }
     void *buf = memalign_or_die(MEM_ALIGNMENT, fill_size + n_iov*sizeof(struct iovec));
     iovec *iov = (struct iovec*)((uint8_t*)buf+fill_size);
     n_iov = 0;
     fill_size = 0;
+    if (padded && (item_start % dsk.csum_block_size))
+    {
+        iov[n_iov++] = (struct iovec){ zero_object, item_start % dsk.csum_block_size };
+    }
     find_holes(rv, item_start, item_end, [&](int pos, bool alloc, uint32_t cur_start, uint32_t cur_end)
     {
         int res = 0;
@@ -376,18 +390,31 @@ bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uin
         }
         return res;
     });
+    if (padded && (item_end % dsk.csum_block_size))
+    {
+        iov[n_iov++] = (struct iovec){ zero_object, dsk.csum_block_size - (item_end % dsk.csum_block_size) };
+    }
     vi = &rv[rv.size()-rv_pos];
     // Save buf into read_vec too but in a creepy way
     // FIXME: Shit, something else should be invented %)
     *vi = (copy_buffer_t){
         .copy_flags = vi->copy_flags,
-        .offset = vi->offset,
+        .offset = (padded ? vi->offset - vi->offset % dsk.csum_block_size : vi->offset),
         .len = ((uint64_t)n_iov << 32) | fill_size,
-        .disk_offset = clean_loc + item_start,
+        .disk_offset = clean_loc + (padded ? vi->offset - vi->offset % dsk.csum_block_size : vi->offset),
         .buf = (uint8_t*)buf,
         .csum_buf = vi->csum_buf,
         .dyn_data = vi->dyn_data,
     };
+    if (padded && (item_start % dsk.csum_block_size))
+    {
+        iov++;
+        n_iov--;
+    }
+    if (padded && (item_end % dsk.csum_block_size))
+    {
+        n_iov--;
+    }
     int submit_fd = (vi->copy_flags & COPY_BUF_JOURNAL ? dsk.journal_fd : dsk.data_fd);
     uint64_t submit_offset = (vi->copy_flags & COPY_BUF_JOURNAL ? journal.offset : dsk.data_offset);
     uint32_t d_pos = 0;
@@ -675,7 +702,7 @@ bool blockstore_impl_t::fulfill_clean_read(blockstore_op_t *read_op, uint64_t &
         }
         for (int i = req; i > 0; i--)
         {
-            if (!read_checksum_block(read_op, i, fulfilled, clean_loc))
+            if (!read_checksum_block(read_op, i, fulfilled, clean_loc, true))
             {
                 return false;
             }
@@ -1090,6 +1117,14 @@ int blockstore_impl_t::read_bitmap(object_id oid, uint64_t target_version, void
             // Condition has to be the same as in dequeue_read()
             if (!IS_IN_FLIGHT(dirty_it->second.state) && target_version >= dirty_it->first.version)
             {
+                if (IS_DELETE(dirty_it->second.state))
+                {
+                    if (result_version)
+                        *result_version = 0;
+                    if (bitmap)
+                        memset(bitmap, 0, dsk.clean_entry_bitmap_size);
+                    return -ENOENT;
+                }
                 if (result_version)
                     *result_version = dirty_it->first.version;
                 if (bitmap)

src/blockstore/v1/rollback.cpp

@@ -63,6 +63,12 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, op->len, sizeof(journal_entry_rollback), 0))
     {
+        if (space_check.give_up)
+        {
+            op->retval = -EAGAIN;
+            FINISH_OP(op);
+            return 2;
+        }
         return 0;
     }
     // There is sufficient space. Check SQEs
@@ -162,7 +168,10 @@ void blockstore_impl_t::mark_rolled_back(const obj_ver_id & ov)
             if (unstab_it != unstable_writes.end())
             {
                 if (max_unstable == 0)
+                {
                     unstable_writes.erase(unstab_it);
+                    wakeup_wait_journal();
+                }
                 else
                     unstab_it->second = max_unstable;
             }

src/blockstore/v1/stable.cpp

@@ -362,6 +362,12 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, op->len, sizeof(journal_entry_stable), 0))
     {
+        if (space_check.give_up)
+        {
+            op->retval = -EAGAIN;
+            FINISH_OP(op);
+            return 2;
+        }
         return 0;
     }
     // There is sufficient space. Check SQEs
@@ -463,6 +469,7 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
                 v.version = dirty_it->first.version;
             }
         }
+        auto erase_end = dirty_db.end();
         while (1)
         {
             bool was_stable = IS_STABLE(dirty_it->second.state);
@@ -470,6 +477,7 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
             {
                 dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK) | BS_ST_STABLE;
                 // Allocations and deletions are counted when they're stabilized
+                // FIXME: Count-on-stabilize logic is actually really strange, reconsider it
                 if (IS_BIG_WRITE(dirty_it->second.state))
                 {
                     int exists = -1;
@@ -523,25 +531,10 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
                 exit(1);
             }
             if (forget_dirty && (IS_BIG_WRITE(dirty_it->second.state) ||
-                IS_DELETE(dirty_it->second.state)))
+                IS_DELETE(dirty_it->second.state)) && erase_end == dirty_db.end())
             {
                 // Big write overrides all previous dirty entries
-                auto erase_end = dirty_it;
-                while (dirty_it != dirty_db.begin())
-                {
-                    dirty_it--;
-                    if (dirty_it->first.oid != v.oid)
-                    {
-                        dirty_it++;
-                        break;
-                    }
-                }
-                auto & clean_db = clean_db_shard(v.oid);
-                auto clean_it = clean_db.find(v.oid);
-                uint64_t clean_loc = clean_it != clean_db.end()
-                    ? clean_it->second.location : UINT64_MAX;
-                erase_dirty(dirty_it, erase_end, clean_loc);
-                break;
+                erase_end = dirty_it;
             }
             if (was_stable || dirty_it == dirty_db.begin())
             {
@@ -550,16 +543,27 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
             dirty_it--;
             if (dirty_it->first.oid != v.oid)
             {
+                dirty_it++;
                 break;
             }
         }
         flusher->enqueue_flush(v);
+        if (erase_end != dirty_db.end())
+        {
+            // Erase only after marking previous entries stable to not skip used_blocks++
+            auto & clean_db = clean_db_shard(v.oid);
+            auto clean_it = clean_db.find(v.oid);
+            uint64_t clean_loc = clean_it != clean_db.end()
+                ? clean_it->second.location : UINT64_MAX;
+            erase_dirty(dirty_it, erase_end, clean_loc);
+        }
     }
     auto unstab_it = unstable_writes.find(v.oid);
     if (unstab_it != unstable_writes.end() &&
         unstab_it->second <= v.version)
     {
         unstable_writes.erase(unstab_it);
+        wakeup_wait_journal();
     }
 }
 

src/blockstore/v1/sync.cpp

@@ -79,6 +79,7 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
         // 2nd step: Data device is synced, prepare & write journal entries
         // Check space in the journal and journal memory buffers
         blockstore_journal_check_t space_check(this);
+        bool space_ok = true;
         if (dsk.csum_block_size)
         {
             // More complex check because all journal entries have different lengths
@@ -90,13 +91,34 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
                 uint64_t dyn_size = dsk.dirty_dyn_size(dirty_entry.offset, dirty_entry.len);
                 if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size, 0))
                 {
-                    return 0;
+                    space_ok = false;
+                    break;
                 }
             }
         }
         else if (!space_check.check_available(op, PRIV(op)->sync_big_writes.size(),
             sizeof(journal_entry_big_write) + dsk.clean_entry_bitmap_size, 0))
         {
+            space_ok = false;
+        }
+        if (!space_ok)
+        {
+            if (space_check.give_up)
+            {
+                // Return everything back
+                unsynced_big_write_count += PRIV(op)->sync_big_writes.size();
+                PRIV(op)->sync_big_writes.insert(PRIV(op)->sync_big_writes.end(),
+                    unsynced_big_writes.begin(), unsynced_big_writes.end());
+                PRIV(op)->sync_big_writes.swap(unsynced_big_writes);
+                PRIV(op)->sync_big_writes.clear();
+                PRIV(op)->sync_small_writes.insert(PRIV(op)->sync_small_writes.end(),
+                    unsynced_small_writes.begin(), unsynced_small_writes.end());
+                PRIV(op)->sync_small_writes.swap(unsynced_small_writes);
+                PRIV(op)->sync_small_writes.clear();
+                op->retval = -EAGAIN;
+                FINISH_OP(op);
+                return 2;
+            }
             return 0;
         }
         // Check SQEs. Don't bother about merging, submit each journal sector as a separate request
@@ -193,6 +215,7 @@ void blockstore_impl_t::ack_sync(blockstore_op_t *op)
         {
             unstable_unsynced--;
             assert(unstable_unsynced >= 0);
+            wakeup_wait_journal();
         }
         dirty_it++;
         while (dirty_it != dirty_db.end() && dirty_it->first.oid == it->oid)
@@ -228,6 +251,7 @@ void blockstore_impl_t::ack_sync(blockstore_op_t *op)
             {
                 unstable_unsynced--;
                 assert(unstable_unsynced >= 0);
+                wakeup_wait_journal();
             }
         }
     }

src/blockstore/v1/write.cpp

@@ -12,19 +12,23 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
     bool found = false, deleted = false, unsynced = false, is_del = (op->opcode == BS_OP_DELETE);
     bool wait_big = false, wait_del = false;
     void *dyn = NULL;
+    uint8_t *dyn_ptr = NULL;
     if (is_del)
     {
         op->len = 0;
     }
-    size_t dyn_size = dsk.dirty_dyn_size(op->offset, op->len);
-    if (!is_del && alloc_dyn_data)
+    else
     {
-        // FIXME: Working with `dyn_data` has to be refactored somehow but I first have to decide how :)
-        // +sizeof(int) = refcount
-        dyn = calloc_or_die(1, dyn_size+sizeof(int));
-        *((int*)dyn) = 1;
+        if (alloc_dyn_data)
+        {
+            // FIXME: Working with `dyn_data` has to be refactored somehow but I first have to decide how :)
+            // +sizeof(int) = refcount
+            size_t dyn_size = dsk.dirty_dyn_size(op->offset, op->len);
+            dyn = calloc_or_die(1, dyn_size+sizeof(int));
+            *((int*)dyn) = 1;
+        }
+        dyn_ptr = (alloc_dyn_data ? (uint8_t*)dyn+sizeof(int) : (uint8_t*)&dyn);
     }
-    uint8_t *dyn_ptr = (alloc_dyn_data ? (uint8_t*)dyn+sizeof(int) : (uint8_t*)&dyn);
     uint64_t version = 1;
     if (dirty_db.size() > 0)
     {
@@ -35,6 +39,11 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
         dirty_it--; // segfaults when dirty_db is empty
         if (dirty_it != dirty_db.end() && dirty_it->first.oid == op->oid)
         {
+            if (op->opcode == BS_OP_WRITE_STABLE && !IS_INSTANT(dirty_it->second.state))
+            {
+                op->retval = -EINVAL;
+                return false;
+            }
             found = true;
             version = dirty_it->first.version + 1;
             deleted = IS_DELETE(dirty_it->second.state);
@@ -308,6 +317,11 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             sizeof(journal_entry_big_write) + dsk.clean_dyn_size,
             (unstable_writes.size()+unstable_unsynced+((dirty_it->second.state & BS_ST_INSTANT) ? 0 : 1))*journal.block_size))
         {
+            if (space_check.give_up)
+            {
+                cancel_all_writes(op, dirty_it, -EAGAIN);
+                return 2;
+            }
             return 0;
         }
         // Big (redirect) write
@@ -400,6 +414,11 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
                 sizeof(journal_entry_small_write) + dyn_size,
                 op->len + (unstable_writes.size()+unstable_unsynced+((dirty_it->second.state & BS_ST_INSTANT) ? 0 : 1))*journal.block_size))
         {
+            if (space_check.give_up)
+            {
+                cancel_all_writes(op, dirty_it, -EAGAIN);
+                return 2;
+            }
             return 0;
         }
         // There is sufficient space. Check SQE(s)
@@ -555,6 +574,25 @@ resume_2:
         if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size,
             (unstable_writes.size()+unstable_unsynced+((dirty_it->second.state & BS_ST_INSTANT) ? 0 : 1))*journal.block_size))
         {
+            if (space_check.give_up)
+            {
+                auto uo_it = used_clean_objects.find(dirty_it->second.location);
+                if (uo_it != used_clean_objects.end())
+                    uo_it->second.was_freed = true;
+                else
+                    data_alloc->set(dirty_it->second.location / dsk.data_block_size, false);
+                if (!(dirty_it->second.state & BS_ST_INSTANT))
+                {
+                    unstable_unsynced--;
+                    assert(unstable_unsynced >= 0);
+                    wakeup_wait_journal();
+                }
+                if (immediate_commit != IMMEDIATE_ALL)
+                    unsynced_big_write_count--;
+                write_iodepth--;
+                cancel_all_writes(op, dirty_it, -EAGAIN);
+                return 2;
+            }
             return 0;
         }
         BS_SUBMIT_CHECK_SQES(1);
@@ -618,6 +656,7 @@ resume_4:
             {
                 unstable_unsynced--;
                 assert(unstable_unsynced >= 0);
+                wakeup_wait_journal();
             }
         }
         dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK)
@@ -770,6 +809,11 @@ int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, 1, sizeof(journal_entry_del), (unstable_writes.size()+unstable_unsynced)*journal.block_size))
     {
+        if (space_check.give_up)
+        {
+            cancel_all_writes(op, dirty_it, -EAGAIN);
+            return 2;
+        }
         return 0;
     }
     // Write current journal sector only if it's dirty and full, or in the immediate_commit mode

src/client/cluster_client.cpp

@@ -1250,9 +1250,11 @@ resume_2:
         return 1;
     }
     else if (op->retval != 0 && op->opcode != OSD_OP_SYNC && !(op->flags & OP_FLUSH_BUFFER) &&
-        op->retval != -EPIPE && (op->retval != -EIO || !client_eio_retry_interval) && (op->retval != -ENOSPC || !client_retry_enospc))
+        op->retval != -EPIPE && op->retval != -EAGAIN &&
+        (op->retval != -EIO || !client_eio_retry_interval) &&
+        (op->retval != -ENOSPC || !client_retry_enospc))
     {
-        // Fatal error (neither -EPIPE, -EIO nor -ENOSPC)
+        // Fatal error (neither -EPIPE, -EAGAIN, -EIO nor -ENOSPC)
         erase_op(op);
         return 1;
     }
@@ -1611,7 +1613,7 @@ resume_1:
     if (op->retval != 0)
     {
         wb->fsync_error();
-        if (op->retval == -EPIPE || op->retval == -EIO || op->retval == -ENOSPC)
+        if (op->retval == -EPIPE || op->retval == -EAGAIN || op->retval == -EIO || op->retval == -ENOSPC)
         {
             // Retry later
             op->parts.clear();
@@ -1662,9 +1664,6 @@ static inline void mem_or(void *res, const void *r2, unsigned int len)
     }
 }
 
-// Error priority: others > EPERM > EIO > ENOSPC > ETIMEDOUT > EPIPE
-#define ERR_PRIO(e) (((e) == -EPERM ? 5 : ((e) == -EIO ? 4 : ((e) == -ENOSPC ? 3 : ((e) == -ETIMEDOUT ? 2 : ((e) == -EPIPE ? 1 : (!(e) ? 0 : 10)))))))
-
 void cluster_client_t::handle_op_part(cluster_op_part_t *part)
 {
     cluster_op_t *op = part->parent;
@@ -1676,7 +1675,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
         if (ERR_PRIO(part->op.reply.hdr.retval) > ERR_PRIO(op->retval))
             op->retval = part->op.reply.hdr.retval;
         uint64_t stop_client_id = 0;
-        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC && op->retval != -EPERM)
+        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC && op->retval != -EPERM && op->retval != -EAGAIN)
         {
             stop_client_id = part->op.client_id;
             if (op->retval != -EPIPE || log_level > 0)
@@ -1698,11 +1697,11 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
         // So do all these things after modifying operation state, otherwise we may hit reenterability bugs
         // FIXME postpone such things to set_immediate here to avoid bugs
         // Set op->retry_after to retry operation after a short pause (not immediately)
-        if (!op->retry_after && (op->retval == -EPIPE ||
+        if (!op->retry_after && (op->retval == -EPIPE || op->retval == -EAGAIN ||
             op->retval == -EIO && client_eio_retry_interval ||
             op->retval == -ENOSPC && client_retry_enospc))
         {
-            op->retry_after = op->retval != -EPIPE ? client_eio_retry_interval : client_retry_interval;
+            op->retry_after = op->retval != -EPIPE && op->retval != -EAGAIN ? client_eio_retry_interval : client_retry_interval;
         }
         reset_retry_timer(op->retry_after);
         if (stop_client_id)

src/client/cluster_client_list.cpp

@@ -370,8 +370,7 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
             fprintf(stderr, "Failed to get PG %u/%u object list from OSD %ju (retval=%jd), skipping\n",
                 cur_list->pg->lst->pool_id, cur_list->pg->pg_num, cur_list->osd_num, op->reply.hdr.retval);
             if (!cur_list->pg->errcode ||
-                cur_list->pg->errcode == -EPIPE ||
-                op->reply.hdr.retval != -EPIPE)
+                ERR_PRIO(op->reply.hdr.retval) > ERR_PRIO(cur_list->pg->errcode))
             {
                 cur_list->pg->errcode = op->reply.hdr.retval;
             }
@@ -434,7 +433,7 @@ void cluster_client_t::finish_list_pg(inode_list_pg_t *pg, bool retry_epipe)
     auto lst = pg->lst;
     if (pg->inflight_ops == 0)
     {
-        if (pg->errcode == -EPIPE && retry_epipe)
+        if ((pg->errcode == -EPIPE || pg->errcode == -EAGAIN) && retry_epipe)
         {
             // Retry listing after <client_retry_interval> ms on EPIPE
             pg->state = LIST_PG_WAIT_RETRY;

src/client/http_client.cpp

@@ -746,6 +746,10 @@ void http_co_t::handle_events()
             {
                 handle_connect_result();
             }
+            else
+            {
+                break;
+            }
         }
         else
         {
@@ -1301,8 +1305,8 @@ static bool ws_parse_frame(std::string & buf, uint8_t & type, std::string & res)
     {
         return false;
     }
-    type = buf[0] & ~0x80;
-    bool mask = !!(buf[1] & 0x80);
+    type = (uint8_t)buf[0] & ~0x80;
+    bool mask = !!((uint8_t)buf[1] & 0x80);
     hdr += mask ? 4 : 0;
     uint64_t len = ((uint8_t)buf[1] & ~0x80);
     if (len == 126)

src/client/http_client.h

@@ -39,7 +39,7 @@ struct http_message_t
     std::map<std::string, std::string> headers;
     std::string tls_cn;
     int tls_ca_idx = -1;
-    uint8_t ws_msg_type = -1;
+    uint8_t ws_msg_type = 0;
     std::string body;
 
     void parse_json_response(std::string & error, json11::Json & r) const;

src/client/messenger.cpp

@@ -634,8 +634,16 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
         op->iov.push_back(op->buf, payload_str.size());
         memcpy(op->buf, payload_str.c_str(), payload_str.size());
     }
-    op->callback = [this, cl](osd_op_t *op)
+    op->callback = [this, client_id = cl->client_id, peer_osd = cl->osd_num](osd_op_t *op)
     {
+        auto cl_it = clients.find(client_id);
+        if (cl_it == clients.end())
+        {
+            on_connect_peer(peer_osd, -EPIPE, 0);
+            delete op;
+            return;
+        }
+        auto cl = cl_it->second;
         std::string json_err;
         json11::Json config;
         bool err = false;
@@ -693,7 +701,6 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
         }
         if (err)
         {
-            osd_num_t peer_osd = cl->osd_num;
             stop_client(op->client_id);
             on_connect_peer(peer_osd, -EINVAL, 0);
             delete op;

src/client/msgr_rdmacm.cpp

@@ -60,7 +60,10 @@ rdma_cm_id *osd_messenger_t::rdmacm_listen(const std::string & bind_address, int
     }
     return listener;
 fail:
-    rdma_destroy_id(listener);
+    if (listener)
+    {
+        rdma_destroy_id(listener);
+    }
     return NULL;
 }
 

src/client/msgr_send.cpp

@@ -432,7 +432,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     auto cl_it = clients.find(cur_op->client_id);
     if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
     {
-        delete cur_op;
+        cur_op->cancel();
         return;
     }
     osd_client_t *cl = cl_it->second;
@@ -530,6 +530,13 @@ copy_ops:
         cl->write_state = 0;
         return true;
     }
+    // get_op_writer_t already guarantees that it's under IOV_MAX
+    assert(cl->send_list.size() <= IOV_MAX);
+    cl->send_list_size = 0;
+    for (auto & iov: cl->send_list)
+    {
+        cl->send_list_size += iov.iov_len;
+    }
     if (!use_sync_send_recv)
     {
         auto iothread = iothreads.size() ? iothreads[cl->peer_fd % iothreads.size()] : NULL;
@@ -542,13 +549,8 @@ copy_ops:
             data_local = {};
         }
         assert(sqe);
-        cl->send_list_size = 0;
-        for (auto & iov: cl->send_list)
-        {
-            cl->send_list_size += iov.iov_len;
-        }
         cl->write_msg.msg_iov = cl->send_list.data();
-        cl->write_msg.msg_iovlen = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
+        cl->write_msg.msg_iovlen = cl->send_list.size();
         cl->refs++;
         ring_data_t* data = ((ring_data_t*)sqe->user_data);
         data->callback = [this, cl](ring_data_t *data) { handle_send(data->res, data->prev, data->more, cl); };
@@ -574,7 +576,7 @@ copy_ops:
     else
     {
         cl->write_msg.msg_iov = cl->send_list.data();
-        cl->write_msg.msg_iovlen = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
+        cl->write_msg.msg_iovlen = cl->send_list.size();
         cl->refs++;
         int result = sendmsg(cl->peer_fd, &cl->write_msg, MSG_NOSIGNAL);
         if (result < 0)

src/client/msgr_stop.cpp

@@ -158,22 +158,6 @@ void osd_messenger_t::destroy_client(osd_client_t *cl)
     if (cl->peer_fd >= 0)
     {
         tfd->set_fd_handler(cl->peer_fd, false, NULL);
-        for (auto rit = read_ready_clients.begin(); rit != read_ready_clients.end(); rit++)
-        {
-            if (*rit == cl->client_id)
-            {
-                read_ready_clients.erase(rit);
-                break;
-            }
-        }
-        for (auto wit = write_ready_clients.begin(); wit != write_ready_clients.end(); wit++)
-        {
-            if (*wit == cl->client_id)
-            {
-                write_ready_clients.erase(wit);
-                break;
-            }
-        }
         clients_by_fd.erase(cl->peer_fd);
     }
 #ifdef WITH_RDMA

src/client/nbd_proxy.cpp

@@ -361,6 +361,26 @@ protected:
 public:
     ~nbd_proxy()
     {
+        if (change_thread)
+        {
+            change_thread->join();
+            change_thread.reset();
+        }
+        if (cli)
+        {
+            delete cli;
+            cli = NULL;
+        }
+        if (epmgr)
+        {
+            delete epmgr;
+            epmgr = NULL;
+        }
+        if (ringloop)
+        {
+            delete ringloop;
+            ringloop = NULL;
+        }
         if (recv_buf)
         {
             free(recv_buf);
@@ -403,6 +423,11 @@ public:
         return cfg;
     }
 
+    void request_stop()
+    {
+        stop = true;
+    }
+
     void exec(json11::Json cfg)
     {
         if (cfg["help"].bool_value())
@@ -525,10 +550,10 @@ help:
         if (!inode)
         {
             // Load image metadata
-            while (!cli->is_ready())
+            while (!cli->is_ready() && !stop)
             {
                 ringloop->loop();
-                if (cli->is_ready())
+                if (cli->is_ready() || stop)
                     break;
                 ringloop->wait();
             }
@@ -542,6 +567,10 @@ help:
             }
             watch->callback = [this](inode_watch_t*) { on_change_watch(); };
         }
+        if (stop)
+        {
+            return;
+        }
 
         // cli->config contains merged config
         if (cli->config.find("nbd_max_devices") != cli->config.end())
@@ -708,17 +737,6 @@ help:
             ringloop->wait();
         }
         cli->flush();
-        if (change_thread)
-        {
-            change_thread->join();
-            change_thread.reset();
-        }
-        delete cli;
-        delete epmgr;
-        delete ringloop;
-        cli = NULL;
-        epmgr = NULL;
-        ringloop = NULL;
     }
 
     void on_change_watch()
@@ -1271,13 +1289,25 @@ protected:
     }
 };
 
+nbd_proxy *srv = NULL;
+
+static void handle_sigint(int sig)
+{
+    if (srv)
+    {
+        srv->request_stop();
+    }
+}
+
 int main(int narg, const char *args[])
 {
     setvbuf(stdout, NULL, _IONBF, 0);
     setvbuf(stderr, NULL, _IONBF, 0);
     exe_name = args[0];
-    nbd_proxy *p = new nbd_proxy();
-    p->exec(nbd_proxy::parse_args(narg, args));
-    delete p;
+    srv = new nbd_proxy();
+    signal(SIGINT, handle_sigint);
+    signal(SIGTERM, handle_sigint);
+    srv->exec(nbd_proxy::parse_args(narg, args));
+    delete srv;
     return 0;
 }

src/client/object_id.h

@@ -111,4 +111,15 @@ namespace std
             return seed;
         }
     };
+
+    template<> struct hash<pool_pg_num_t>
+    {
+        inline size_t operator()(const pool_pg_num_t &s) const
+        {
+            size_t seed = 0;
+            seed ^= ((size_t)s.pool_id + 0xc6a4a7935bd1e995 + (seed << 6) + (seed >> 2));
+            seed ^= ((size_t)s.pg_num + 0xc6a4a7935bd1e995 + (seed << 6) + (seed >> 2));
+            return seed;
+        }
+    };
 }

src/client/osd_ops.h

@@ -57,6 +57,9 @@
 #define OSD_SEC_LOCK_PG 1
 #define OSD_SEC_UNLOCK_PG 2
 
+// Error priority: others > EPERM > EIO > EDOM > ENOSPC > EPIPE > EAGAIN
+#define ERR_PRIO(e) (((e) == -EPERM ? 6 : ((e) == -EIO ? 5 : ((e) == -EDOM ? 4 : ((e) == -ENOSPC ? 3 : ((e) == -EPIPE ? 2 : ((e) == -EAGAIN ? 1 : (!(e) ? 0 : 10))))))))
+
 // common request and reply headers
 struct __attribute__((__packed__)) osd_op_header_t
 {

src/client/ublk_server.cpp

@@ -99,6 +99,21 @@ public:
 
     ~ublk_server()
     {
+        if (cli)
+        {
+            delete cli;
+            cli = NULL;
+        }
+        if (epmgr)
+        {
+            delete epmgr;
+            epmgr = NULL;
+        }
+        if (ringloop)
+        {
+            delete ringloop;
+            ringloop = NULL;
+        }
         if (ctrl_fd >= 0)
         {
             close(ctrl_fd);
@@ -114,11 +129,6 @@ public:
             free(buf);
         }
         buffers.clear();
-        if (ringloop)
-        {
-            delete ringloop;
-            ringloop = NULL;
-        }
     }
 
     static json11::Json::object parse_args(int narg, const char *args[])
@@ -159,6 +169,11 @@ public:
         return cfg;
     }
 
+    void request_stop()
+    {
+        stop = true;
+    }
+
     void exec(json11::Json cfg)
     {
         if (cfg["help"].bool_value())
@@ -272,13 +287,17 @@ help:
         }
 
         // Load image metadata
-        while (!cli->is_ready())
+        while (!cli->is_ready() && !stop)
         {
             ringloop->loop();
-            if (cli->is_ready())
+            if (cli->is_ready() || stop)
                 break;
             ringloop->wait();
         }
+        if (stop)
+        {
+            return;
+        }
         if (!inode)
         {
             watch = cli->st_cli->watch_inode(image_name);
@@ -367,10 +386,6 @@ help:
         }
         cli->flush();
         ringloop->unregister_consumer(&consumer);
-        delete cli;
-        delete epmgr;
-        cli = NULL;
-        epmgr = NULL;
     }
 
     void load_module()
@@ -980,13 +995,25 @@ protected:
     }
 };
 
+ublk_server *srv = NULL;
+
+static void handle_sigint(int sig)
+{
+    if (srv)
+    {
+        srv->request_stop();
+    }
+}
+
 int main(int narg, const char *args[])
 {
     setvbuf(stdout, NULL, _IONBF, 0);
     setvbuf(stderr, NULL, _IONBF, 0);
     exe_name = args[0];
-    ublk_server *p = new ublk_server();
-    p->exec(ublk_server::parse_args(narg, args));
-    delete p;
+    srv = new ublk_server();
+    signal(SIGINT, handle_sigint);
+    signal(SIGTERM, handle_sigint);
+    srv->exec(ublk_server::parse_args(narg, args));
+    delete srv;
     return 0;
 }

src/disk_tool/disk_simple_offsets.cpp

@@ -130,6 +130,19 @@ void disk_tool_simple_offsets(json11::Json cfg, bool json_output)
     uint64_t clean_entry_bitmap_size = data_block_size/bitmap_granularity/8;
     uint64_t object_count = ((device_size-meta_offset)/data_block_size);
     uint64_t meta_size;
+    // A metadata entry must fit into a single metadata block, otherwise entries_per_block below is 0
+    auto check_entry_size = [&](uint64_t entry_size)
+    {
+        if (entry_size && entry_size <= device_block_size)
+            return;
+        fprintf(
+            stderr, "Metadata entry size (%ju bytes) exceeds meta_block_size (%ju bytes) with block_size=%ju,"
+            " bitmap_granularity=%ju, csum_block_size=%u. Decrease block_size, increase csum_block_size"
+            " or increase device_block_size\n",
+            entry_size, device_block_size, data_block_size, bitmap_granularity, csum_block_size
+        );
+        exit(1);
+    };
     if (meta_format == BLOCKSTORE_META_FORMAT_HEAP)
     {
         uint32_t min_object_size = sizeof(heap_big_intent_t) + (data_csum_size ? data_csum_size : 4) + 2*clean_entry_bitmap_size;
@@ -138,18 +151,21 @@ void disk_tool_simple_offsets(json11::Json cfg, bool json_output)
             meta_reserve = 1.5;
         else if (meta_reserve < 1)
             meta_reserve = 1;
+        check_entry_size(min_object_size);
         uint32_t entries_per_block = device_block_size / min_object_size;
         meta_size = device_block_size * (uint64_t)((object_count+entries_per_block-1) / entries_per_block * meta_reserve);
     }
     else if (meta_format == BLOCKSTORE_META_FORMAT_V2)
     {
         uint64_t clean_entry_size = 24 /*sizeof(clean_disk_entry)*/ + 2*clean_entry_bitmap_size + data_csum_size + 4 /*entry_csum*/;
+        check_entry_size(clean_entry_size);
         uint64_t entries_per_block = device_block_size / clean_entry_size;
         meta_size = (1 + (object_count+entries_per_block-1)/entries_per_block) * device_block_size;
     }
     else if (meta_format == BLOCKSTORE_META_FORMAT_V1)
     {
         uint64_t clean_entry_size = 24 /*sizeof(clean_disk_entry)*/ + 2*clean_entry_bitmap_size;
+        check_entry_size(clean_entry_size);
         uint64_t entries_per_block = device_block_size / clean_entry_size;
         meta_size = (1 + (object_count+entries_per_block-1)/entries_per_block) * device_block_size;
     }

src/nfs/nfs_proxy.cpp

@@ -39,23 +39,43 @@ nfs_proxy_t::~nfs_proxy_t()
 #ifdef WITH_RDMACM
     destroy_rdma();
 #endif
+    // Destroy the client
     if (kvfs)
+    {
         delete kvfs;
+        kvfs = NULL;
+    }
     if (blockfs)
+    {
         delete blockfs;
+        blockfs = NULL;
+    }
     if (db)
+    {
         delete db;
+        db = NULL;
+    }
     if (cmd)
+    {
         delete cmd;
+        cmd = NULL;
+    }
     if (cli)
     {
         cli->flush();
         delete cli;
+        cli = NULL;
     }
     if (epmgr)
+    {
         delete epmgr;
+        epmgr = NULL;
+    }
     if (ringloop)
+    {
         delete ringloop;
+        ringloop = NULL;
+    }
 }
 
 static const char* help_text =
@@ -340,13 +360,17 @@ void nfs_proxy_t::run(json11::Json cfg)
         blockfs->init(this, cfg);
     }
     // Load image metadata
-    while (!cli->is_ready())
+    while (!cli->is_ready() && !finished)
     {
         ringloop->loop();
-        if (cli->is_ready())
+        if (cli->is_ready() || finished)
             break;
         ringloop->wait();
     }
+    if (finished)
+    {
+        return;
+    }
     // Check default pool
     check_default_pool();
     // Daemonize before initializing messenger and RDMA because otherwise RDMA doesn't survive fork()
@@ -392,29 +416,11 @@ void nfs_proxy_t::run(json11::Json cfg)
         ringloop->loop();
         ringloop->wait();
     }
-    // Destroy the client
-    cli->flush();
-    if (kvfs)
-    {
-        delete kvfs;
-        kvfs = NULL;
-    }
-    if (blockfs)
-    {
-        delete blockfs;
-        blockfs = NULL;
-    }
-    if (db)
-    {
-        delete db;
-        db = NULL;
-    }
-    delete cli;
-    delete epmgr;
-    delete ringloop;
-    cli = NULL;
-    epmgr = NULL;
-    ringloop = NULL;
+}
+
+void nfs_proxy_t::request_stop()
+{
+    finished = true;
 }
 
 void nfs_proxy_t::run_server(json11::Json cfg)
@@ -1551,14 +1557,26 @@ void nfs_proxy_t::check_exit()
     finished = true;
 }
 
+nfs_proxy_t *srv = NULL;
+
+static void handle_sigint(int sig)
+{
+    if (srv)
+    {
+        srv->request_stop();
+    }
+}
+
 int main(int narg, const char *args[])
 {
     setvbuf(stdout, NULL, _IONBF, 0);
     setvbuf(stderr, NULL, _IONBF, 0);
     exe_name = strrchr(args[0], '/');
     exe_name = exe_name ? exe_name+1 : args[0];
-    nfs_proxy_t *p = new nfs_proxy_t();
-    p->run(nfs_proxy_t::parse_args(narg, args));
-    delete p;
+    srv = new nfs_proxy_t();
+    signal(SIGINT, handle_sigint);
+    signal(SIGTERM, handle_sigint);
+    srv->run(nfs_proxy_t::parse_args(narg, args));
+    delete srv;
     return 0;
 }

src/nfs/nfs_proxy.h

@@ -79,6 +79,7 @@ public:
 
     static json11::Json::object parse_args(int narg, const char *args[]);
     void run(json11::Json cfg);
+    void request_stop();
     void run_server(json11::Json cfg);
     void watch_stats();
     void parse_stats(etcd_kv_t & kv);

src/osd/osd_cluster.cpp

@@ -623,7 +623,7 @@ void osd_t::create_osd_state()
             {
                 printf("  listening at: %s:%jd\n", addr.string_value().c_str(), port);
             }
-            force_stop(0);
+            force_stop(1);
             return;
         }
         if (run_primary)

src/osd/osd_flush.cpp

@@ -81,18 +81,9 @@ void osd_t::handle_flush_op(bool rollback, pool_id_t pool_id, pg_num_t pg_num, p
     if (retval != 0)
     {
         fb->flush_errors++;
-        if (peer_osd == this->osd_num)
+        printf("Error while doing flush on OSD %ju: %d (%s)\n", peer_osd, retval, strerror(-retval));
+        if (peer_osd != this->osd_num && retval != -EAGAIN)
         {
-            throw std::runtime_error(
-                std::string(rollback
-                    ? "Error while doing local rollback operation: "
-                    : "Error while doing local stabilize operation: "
-                ) + strerror(-retval)
-            );
-        }
-        else
-        {
-            printf("Error while doing flush on OSD %ju: %d (%s)\n", osd_num, retval, strerror(-retval));
             auto peer_it = msgr.osd_peers.find(peer_osd);
             if (peer_it != msgr.osd_peers.end())
             {

src/osd/osd_primary_subops.cpp

@@ -101,15 +101,7 @@ void osd_t::finish_op(osd_op_t *cur_op, int retval)
     else
     {
         // FIXME add separate magic number for primary ops
-        auto cl_it = msgr.clients.find(cur_op->client_id);
-        if (cl_it != msgr.clients.end())
-        {
-            msgr.outbox_push(cur_op);
-        }
-        else
-        {
-            delete cur_op;
-        }
+        msgr.outbox_push(cur_op);
     }
 }
 
@@ -319,11 +311,10 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop)
     blockstore_op_t *bs_op = subop->bs_op;
     int64_t expected = bs_op->opcode == BS_OP_READ || bs_op->opcode == BS_OP_WRITE
         || bs_op->opcode == BS_OP_WRITE_STABLE ? bs_op->len : 0;
-    if (bs_op->retval != expected && bs_op->opcode != BS_OP_READ &&
-        (bs_op->opcode != BS_OP_WRITE && bs_op->opcode != BS_OP_WRITE_STABLE ||
-        bs_op->retval != -ENOSPC))
+    if (bs_op->retval != expected && bs_op->opcode != BS_OP_READ && bs_op->retval != -EAGAIN &&
+        (bs_op->opcode != BS_OP_WRITE && bs_op->opcode != BS_OP_WRITE_STABLE || bs_op->retval != -ENOSPC))
     {
-        // die on any error except ENOSPC during write
+        // die on any error except ENOSPC or EAGAIN during write
         if (bs_op->opcode == BS_OP_WRITE || bs_op->opcode == BS_OP_WRITE_STABLE)
         {
             printf(
@@ -436,7 +427,7 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
 #endif
         if (!(op_data->flags & OP_DATA_SKIP_VER_CHECK))
         {
-            if (op_data->fact_ver != version && (op_data->errors > 0 || op_data->done > 0) &&
+            if (op_data->fact_ver != version && op_data->done > 0 &&
                 (version != 0 && op_data->fact_ver != 0 || !(op_data->flags & OP_DATA_ENOENT_OK)))
             {
                 fprintf(
@@ -481,16 +472,12 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
             );
         }
         subop->rmw_buf = NULL;
-        // Error priority: ENOSPC > others > EIO > EDOM > EPIPE
-        if (op_data->errcode == 0 ||
-            retval == -ENOSPC && op_data->errcode != -ENOSPC ||
-            retval == -EIO && (op_data->errcode == -EDOM || op_data->errcode == -EPIPE) ||
-            retval == -EDOM && (op_data->errcode == -EPIPE) ||
-            retval != -EIO && retval != -EDOM && retval != -EPIPE)
+        // Error priority: ENOSPC > others > EIO > EDOM > EPIPE > EAGAIN
+        if (op_data->errcode == 0 || ERR_PRIO(retval) > ERR_PRIO(op_data->errcode))
         {
             op_data->errcode = retval;
         }
-        if (subop->client_id && retval != -EDOM && retval != -ERANGE &&
+        if (subop->client_id && retval != -EDOM && retval != -ERANGE && retval != -EAGAIN &&
             (retval != -ENOSPC || opcode != OSD_OP_SEC_WRITE && opcode != OSD_OP_SEC_WRITE_STABLE) &&
             (retval != -EIO || opcode != OSD_OP_SEC_READ))
         {
@@ -498,6 +485,32 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
             msgr.stop_client(subop->client_id);
             op_data->drops++;
         }
+        else if (opcode == OSD_OP_SEC_STABILIZE || opcode == OSD_OP_SEC_ROLLBACK)
+        {
+            // STABILIZE/ROLLBACK should never fail, but if they do, we repeer the PGs
+            assert(subop->iov.count == 1);
+            obj_ver_id *vers = (obj_ver_id*)subop->iov.buf[0].iov_base;
+            size_t n = subop->iov.buf[0].iov_len/sizeof(obj_ver_id);
+            robin_hood::unordered_flat_set<pool_pg_num_t> to_repeer;
+            for (size_t i = 0; i < n; i++)
+            {
+                to_repeer.insert((pool_pg_num_t){
+                    .pool_id = INODE_POOL(vers[n].oid.inode),
+                    .pg_num = map_to_pg(vers[n].oid),
+                });
+            }
+            ringloop->set_immediate([this, to_repeer = std::move(to_repeer)]()
+            {
+                for (auto & ppg: to_repeer)
+                {
+                    auto pg_it = pgs.find(ppg);
+                    if (pg_it != pgs.end() && (pg_it->second.state & PG_ACTIVE))
+                    {
+                        repeer_pg(pg_it->second);
+                    }
+                }
+            });
+        }
         // Increase op_data->errors after stop_client to prevent >= n_subops running twice
         op_data->errors++;
     }
@@ -711,6 +724,7 @@ void osd_t::submit_primary_stab_subops(osd_op_t *cur_op)
         {
             clock_gettime(CLOCK_REALTIME, &subops[i].tv_begin);
             subops[i].op_type = (uint64_t)cur_op;
+            subops[i].iov.push_back(op_data->unstable_writes + stab_osd.start, stab_osd.len * sizeof(obj_ver_id));
             subops[i].bs_op = new blockstore_op_t((blockstore_op_t){
                 .opcode = BS_OP_STABLE,
                 .callback = [subop = &subops[i], this](blockstore_op_t *bs_subop)
@@ -732,7 +746,7 @@ void osd_t::submit_primary_stab_subops(osd_op_t *cur_op)
                     .magic = SECONDARY_OSD_OP_MAGIC,
                     .opcode = OSD_OP_SEC_STABILIZE,
                 },
-                .len = (uint64_t)(stab_osd.len * sizeof(obj_ver_id)),
+                .len = (uint64_t)stab_osd.len * sizeof(obj_ver_id),
                 .flags = cur_op->client_id == SELF_CLIENT && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0,
             } };
             subops[i].iov.push_back(op_data->unstable_writes + stab_osd.start, stab_osd.len * sizeof(obj_ver_id));
@@ -786,6 +800,7 @@ void osd_t::submit_primary_rollback_subops(osd_op_t *cur_op, const uint64_t* osd
             {
                 clock_gettime(CLOCK_REALTIME, &subop->tv_begin);
                 subop->op_type = (uint64_t)cur_op;
+                subop->iov.push_back(op_data->unstable_writes + i, sizeof(obj_ver_id));
                 subop->bs_op = new blockstore_op_t((blockstore_op_t){
                     .opcode = BS_OP_ROLLBACK,
                     .callback = [subop, this](blockstore_op_t *bs_subop)

src/osd/osd_primary_write.cpp

@@ -7,24 +7,27 @@
 bool osd_t::check_write_queue(osd_op_t *cur_op, pg_t & pg)
 {
     osd_primary_op_data_t *op_data = cur_op->op_data;
+    op_data->st = 1;
     // First check if PG is not active anymore
     if (!(pg.state & PG_ACTIVE))
     {
         pg_cancel_write_queue(pg, cur_op, op_data->oid, -EPIPE);
         return false;
     }
-    // Check if actions are pending for this object
-    auto act_it = pg.flush_actions.lower_bound((obj_piece_id_t){
-        .oid = op_data->oid,
-        .osd_num = 0,
-    });
-    op_data->st = 1;
-    if (act_it != pg.flush_actions.end() &&
-        act_it->first.oid.inode == op_data->oid.inode &&
-        (act_it->first.oid.stripe & ~STRIPE_MASK) == op_data->oid.stripe)
+    if (pg.state & PG_HAS_UNCLEAN)
     {
-        pg.write_queue.emplace(op_data->oid, cur_op);
-        return false;
+        // Check if actions are pending for this object
+        auto act_it = pg.flush_actions.lower_bound((obj_piece_id_t){
+            .oid = op_data->oid,
+            .osd_num = 0,
+        });
+        if (act_it != pg.flush_actions.end() &&
+            act_it->first.oid.inode == op_data->oid.inode &&
+            (act_it->first.oid.stripe & ~STRIPE_MASK) == op_data->oid.stripe)
+        {
+            pg.write_queue.emplace(op_data->oid, cur_op);
+            return false;
+        }
     }
     // Check if there are other write requests to the same object
     auto vo_it = pg.write_queue.find(op_data->oid);
@@ -166,11 +169,14 @@ resume_3:
         if (pg.cur_set.data() != op_data->prev_set && (op_data->stripes[0].write_start != 0 ||
             op_data->stripes[0].write_end != bs_block_size))
         {
-            memcpy(
-                (uint8_t*)op_data->stripes[0].read_buf + op_data->stripes[0].req_start,
-                op_data->stripes[0].write_buf,
-                op_data->stripes[0].req_end - op_data->stripes[0].req_start
-            );
+            if (op_data->stripes[0].req_end > op_data->stripes[0].req_start)
+            {
+                memcpy(
+                    (uint8_t*)op_data->stripes[0].read_buf + op_data->stripes[0].req_start,
+                    op_data->stripes[0].write_buf,
+                    op_data->stripes[0].req_end - op_data->stripes[0].req_start
+                );
+            }
             op_data->stripes[0].write_buf = op_data->stripes[0].read_buf;
             op_data->stripes[0].write_start = 0;
             op_data->stripes[0].write_end = bs_block_size;
@@ -252,9 +258,21 @@ resume_4:
         return;
     }
 resume_5:
+    if (!(pg.state & PG_ACTIVE))
+    {
+        if (op_data->subops)
+        {
+            delete[] op_data->subops;
+            op_data->subops = NULL;
+        }
+        pg.ver_override.erase(op_data->oid);
+        deref_object_state(pg, &op_data->object_state, true);
+        pg_cancel_write_queue(pg, cur_op, op_data->oid, -EPIPE);
+        return;
+    }
     if (op_data->errors > 0)
     {
-        // Handle ENOSPC/EDOM/ERANGE/EIO. If some subops fail, but others succeed,
+        // Handle ENOSPC/EAGAIN/EDOM/ERANGE/EIO. If some subops fail, but others succeed,
         // next writes to the same object will also fail because they'll try
         // to overwrite the same version number which will result in EEXIST.
         // To fix it, we should mark the object as degraded for replicas,
@@ -263,6 +281,11 @@ resume_5:
         {
             if (pg.scheme != POOL_SCHEME_REPLICATED)
             {
+                if (op_data->subops)
+                {
+                    delete[] op_data->subops;
+                    op_data->subops = NULL;
+                }
                 submit_primary_rollback_subops(cur_op, pg.cur_set.data());
 resume_11:
                 if (op_data->n_subops > 0)
@@ -271,21 +294,17 @@ resume_11:
                     return;
                 }
 resume_12:
-                // Ignore ROLLBACK errors - submit_primary_subops will drop the connection if it fails
-                delete[] op_data->unstable_writes;
-                op_data->unstable_writes = NULL;
+                if (op_data->unstable_writes)
+                {
+                    delete[] op_data->unstable_writes;
+                    op_data->unstable_writes = NULL;
+                }
+                // Ignore ROLLBACK errors - submit_primary_subops will repeer the PG if it fails
             }
             else
             {
                 pg.ver_override.erase(op_data->oid);
                 mark_partial_write(pg, cur_op);
-                if (op_data->subops)
-                {
-                    delete[] op_data->subops;
-                    op_data->subops = NULL;
-                }
-                pg_cancel_write_queue(pg, cur_op, op_data->oid, op_data->errcode);
-                return;
             }
         }
         if (op_data->subops)
@@ -298,13 +317,6 @@ resume_12:
         cur_op->reply.hdr.retval = op_data->errcode;
         goto continue_others;
     }
-    if (!(pg.state & PG_ACTIVE))
-    {
-        pg.ver_override.erase(op_data->oid);
-        deref_object_state(pg, &op_data->object_state, true);
-        pg_cancel_write_queue(pg, cur_op, op_data->oid, -EPIPE);
-        return;
-    }
     if (pg.scheme != POOL_SCHEME_REPLICATED)
     {
         // Remove version override just after the write, but before stabilizing
@@ -549,6 +561,7 @@ resume_6:
                 return false;
             }
 resume_7:
+            // Ignore STABILIZE errors - submit_primary_subops will repeer the PG if it fails
             // FIXME: Free those in the destructor?
             delete op_data->unstable_write_osds;
             delete[] op_data->unstable_writes;

src/osd/osd_rmw.cpp

@@ -886,11 +886,14 @@ static void calc_rmw_parity_copy_mod(osd_rmw_stripe_t *stripes, int pg_size, int
                 (stripes[role].req_start != 0 || stripes[role].req_end != chunk_size))
             {
                 // Copy modified chunk into the read buffer to write it back
-                memcpy(
-                    (uint8_t*)stripes[role].read_buf + stripes[role].req_start,
-                    stripes[role].write_buf,
-                    stripes[role].req_end - stripes[role].req_start
-                );
+                if (stripes[role].req_end > stripes[role].req_start)
+                {
+                    memcpy(
+                        (uint8_t*)stripes[role].read_buf + stripes[role].req_start,
+                        stripes[role].write_buf,
+                        stripes[role].req_end - stripes[role].req_start
+                    );
+                }
                 stripes[role].write_buf = stripes[role].read_buf;
                 stripes[role].write_start = 0;
                 stripes[role].write_end = chunk_size;

src/test/mock/messenger.cpp

@@ -21,7 +21,13 @@ osd_messenger_t::~osd_messenger_t()
 
 void osd_messenger_t::outbox_push(osd_op_t *cur_op)
 {
-    auto cl = clients.at(cur_op->client_id);
+    auto cl_it = clients.find(cur_op->client_id);
+    if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
+    {
+        cur_op->cancel();
+        return;
+    }
+    auto cl = cl_it->second;
     if (cur_op->op_type == OSD_OP_OUT)
         cur_op->req.hdr.id = ++cl->send_op_id;
     cl->sent_ops[cur_op->req.hdr.id] = cur_op;

src/test/osd_test.cpp

@@ -55,8 +55,11 @@ static osd_op_t *make_write_op(inode_t inode, uint64_t offset, uint64_t len, uin
     op->req.rw.offset = offset;
     op->req.rw.len = len;
     op->req.rw.version = 0;
-    op->buf = malloc(len);
-    memset(op->buf, fill, len);
+    if (len > 0)
+    {
+        op->buf = malloc(len);
+        memset(op->buf, fill, len);
+    }
     return op;
 }
 

src/test/test_blockstore.cpp

@@ -137,12 +137,13 @@ struct bs_test_t
         }
     }
 
-    void exec_op(blockstore_op_t *op)
+    void exec_op(blockstore_op_t *op, bool print = true)
     {
         bool done = false;
         op->callback = [&](blockstore_op_t *op)
         {
-            printf("op opcode=%ju completed retval=%jd\n", op->opcode, op->retval);
+            if (print)
+                printf("op opcode=%ju completed retval=%jd\n", op->opcode, op->retval);
             done = true;
         };
         bs->enqueue_op(op);
@@ -976,6 +977,62 @@ static void test_list_limit()
     free(op.buf);
 }
 
+static void test_write_no_space_eagain()
+{
+    printf("\n-- test_write_no_space_eagain\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    // 1 MB journal (data_offset 32 MB - journal_offset)
+    test.config["journal_offset"] = std::to_string(31*1024*1024);
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
+    memset(buf, 0xAA, 4096);
+
+    // fill the journal with unstable writes
+    int i = 0, blocked_retval = 0;
+    auto do_write = [&]()
+    {
+        blockstore_op_t op;
+        op.opcode = BS_OP_WRITE;
+        op.oid = { .inode = 1, .stripe = (uint64_t)(i % 4) * 0x20000 };
+        op.version = 1 + i / 4;
+        op.offset = 0;
+        op.len = 4096;
+        op.buf = buf;
+        test.exec_op(&op, false);
+        if (op.retval != (int)op.len)
+            blocked_retval = op.retval;
+    };
+    for (i = 0; i < 100000 && !blocked_retval; i++)
+        do_write();
+    printf("write #%d refused with retval=%d\n", i, blocked_retval);
+    assert(blocked_retval == -EAGAIN);
+
+    // check that a stabilize operation and one more write can proceed
+    {
+        blockstore_op_t op;
+        op.opcode = BS_OP_STABLE;
+        op.len = 1;
+        op.buf = (uint8_t*)malloc_or_die(1 * sizeof(obj_ver_id));
+        ((obj_ver_id*)op.buf)[0] = {
+            .oid = { .inode = 1, .stripe = 0 },
+            .version = 2,
+        };
+        test.exec_op(&op);
+        assert(op.retval == 0);
+        free(op.buf);
+    }
+
+    blocked_retval = 0;
+    do_write();
+    assert(!blocked_retval);
+
+    free(buf);
+}
+
 // FIXME Add a simple intent_write / big_intent test
 
 int main(int narg, char *args[])
@@ -996,5 +1053,6 @@ int main(int narg, char *args[])
     test_fsync_almost_batch_big();
     test_fsync_batch_big();
     test_list_limit();
+    test_write_no_space_eagain();
     return 0;
 }

src/test/test_blockstore_v1.cpp

@@ -5,6 +5,7 @@
 #include "str_util.h"
 #include "ringloop_mock.h"
 #include "blockstore/v1/impl.h"
+#include "blockstore/v1/internal.h"
 
 struct bs_test_t
 {
@@ -45,11 +46,22 @@ struct bs_test_t
         return bs->used_blocks;
     }
 
+    uint64_t free_block_count()
+    {
+        return bs->get_free_block_count();
+    }
+
     bool is_data_loc_used(uint64_t loc)
     {
         return bs->data_alloc->get(loc / bs->dsk.data_block_size);
     }
 
+    v1::dirty_entry* find_dirty_entry(object_id oid, uint64_t version)
+    {
+        auto it = bs->dirty_db.find((obj_ver_id){ .oid = oid, .version = version });
+        return it == bs->dirty_db.end() ? NULL : &it->second;
+    }
+
     void force_compaction()
     {
         printf("triggering compaction\n");
@@ -315,6 +327,25 @@ static void test_validate_padded_journal()
     test.init();
     printf("blockstore initialized\n");
 
+    // initialize the beginning of data area with non-zero data to check
+    // explicit padding instead of the on-disk data verification
+    {
+        std::vector<uint8_t> nonzero_data(1048576, 0xF8);
+        ring_data_t data;
+        io_uring_sqe sqe;
+        iovec iov;
+        sqe.opcode = IORING_OP_WRITEV;
+        sqe.off = test.dsk().data_offset;
+        sqe.addr = (uint64_t)&iov;
+        sqe.len = 1;
+        sqe.rw_flags = 0;
+        iov.iov_base = nonzero_data.data();
+        iov.iov_len = nonzero_data.size();
+        sqe.user_data = (uint64_t)&data;
+        bool ok = test.data_disk->submit(&sqe);
+        assert(ok);
+    }
+
     printf("write v1 4+32k\n");
     blockstore_op_t op;
     op.opcode = BS_OP_WRITE_STABLE;
@@ -852,6 +883,319 @@ static void test_meta_tail_reload()
     printf("blockstore re-initialized with an overflowing entry\n");
 }
 
+static void test_write_no_space_eagain()
+{
+    printf("\n-- test_write_no_space_eagain\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    // 1 MB journal (data_offset 32 MB - journal_offset), so it fills in a few hundred writes
+    test.config["journal_offset"] = std::to_string(31*1024*1024);
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
+    memset(buf, 0xAA, 4096);
+
+    // fill the journal with unstable writes
+    int i = 0, blocked_retval = 0;
+    auto do_write = [&]()
+    {
+        blockstore_op_t op;
+        op.opcode = BS_OP_WRITE;
+        op.oid = { .inode = 1, .stripe = (uint64_t)(i % 4) * 0x20000 };
+        op.version = 1 + i / 4;
+        op.offset = 0;
+        op.len = 4096;
+        op.buf = buf;
+        test.exec_op(&op);
+        if (op.retval != (int)op.len)
+            blocked_retval = op.retval;
+    };
+    for (i = 0; i < 100000 && !blocked_retval; i++)
+        do_write();
+    printf("write #%d refused with retval=%d\n", i, blocked_retval);
+    assert(blocked_retval == -EAGAIN);
+
+    // check that a stabilize operation and one more write can proceed
+    // journal space_check is simplified so it requires slightly more space than actually needed
+    {
+        blockstore_op_t op;
+        op.opcode = BS_OP_STABLE;
+        op.len = 4;
+        op.buf = (uint8_t*)malloc_or_die(4 * sizeof(obj_ver_id));
+        ((obj_ver_id*)op.buf)[0] = {
+            .oid = { .inode = 1, .stripe = 0 },
+            .version = 2,
+        };
+        ((obj_ver_id*)op.buf)[1] = {
+            .oid = { .inode = 1, .stripe = 0x20000 },
+            .version = 2,
+        };
+        ((obj_ver_id*)op.buf)[2] = {
+            .oid = { .inode = 1, .stripe = 0x40000 },
+            .version = 1,
+        };
+        ((obj_ver_id*)op.buf)[3] = {
+            .oid = { .inode = 1, .stripe = 0x60000 },
+            .version = 1,
+        };
+        test.exec_op(&op);
+        assert(op.retval == 0);
+        free(op.buf);
+    }
+
+    blocked_retval = 0;
+    do_write();
+    assert(!blocked_retval);
+
+    free(buf);
+}
+
+// Check null dereference in read_bitmap over DELETE
+static void test_read_bitmap_of_deleted_object()
+{
+    printf("\n-- test_read_bitmap_of_deleted_object\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    // 128k/16k*4 = 32 bytes of checksums --> dyn_data allocated dynamically
+    test.config["data_csum_type"] = "crc32c";
+    test.config["csum_block_size"] = "16384";
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint8_t bmp[16];
+    memset(bmp, 0xCC, sizeof(bmp));
+
+    printf("write v1 0+128k\n");
+    blockstore_op_t op;
+    op.opcode = BS_OP_WRITE_STABLE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op.len);
+    op.bitmap = bmp;
+    memset(op.buf, 0xAA, op.len);
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+    {
+        // Check that big_write is here and with dyn_data
+        v1::dirty_entry *de = test.find_dirty_entry(op.oid, 1);
+        assert(de != NULL);
+        assert(de->dyn_data != NULL);
+    }
+
+    {
+        uint64_t version = 0;
+        assert(test.dsk().clean_entry_bitmap_size <= sizeof(bmp));
+        memset(bmp, 0, sizeof(bmp));
+        assert(test.bs->read_bitmap(op.oid, UINT64_MAX, bmp, &version) == 0);
+        assert(version == 1);
+        assert(memcheck(bmp, 0xCC, test.dsk().clean_entry_bitmap_size));
+    }
+
+    {
+        printf("delete v2\n");
+        blockstore_op_t del_op;
+        del_op.opcode = BS_OP_DELETE;
+        del_op.oid = { .inode = 1, .stripe = 0 };
+        del_op.version = 2;
+        test.exec_op(&del_op);
+        assert(del_op.retval == 0);
+    }
+
+    {
+        // Check that the delete is still in dirty_db
+        v1::dirty_entry *de = test.find_dirty_entry(op.oid, 2);
+        assert(de != NULL);
+        assert((de->state & BS_ST_TYPE_MASK) == BS_ST_DELETE);
+        assert(de->dyn_data == NULL);
+    }
+
+    auto check_deleted_bmp = [&]()
+    {
+        uint64_t version = UINT64_MAX;
+        memset(bmp, 0xCC, sizeof(bmp));
+        assert(test.bs->read_bitmap(op.oid, UINT64_MAX, bmp, &version) == -ENOENT);
+        assert(version == 0);
+        assert(memcheck(bmp, 0, test.dsk().clean_entry_bitmap_size));
+    };
+    printf("read_bitmap of the deleted object - should be -ENOENT\n");
+    check_deleted_bmp();
+
+    {
+        // Reading the data must agree with that
+        printf("read of the deleted object - should be -ENOENT\n");
+        blockstore_op_t read_op;
+        read_op.opcode = BS_OP_READ;
+        read_op.oid = { .inode = 1, .stripe = 0 };
+        read_op.version = UINT64_MAX;
+        read_op.offset = 0;
+        read_op.len = 128*1024;
+        read_op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, read_op.len);
+        test.exec_op(&read_op);
+        assert(read_op.retval == -ENOENT);
+        free(read_op.buf);
+    }
+
+    // And once the delete is flushed too, via the clean_db path
+    test.force_compaction();
+    printf("read_bitmap after the delete is flushed - should be -ENOENT\n");
+    check_deleted_bmp();
+
+    // Write over delete to also check that it doesn't read NULL dyn_data
+    printf("write v3 0+128k over the delete\n");
+    op.version = 3;
+    memset(op.buf, 0xAB, op.len);
+    memset(bmp, 0xDE, sizeof(bmp));
+    test.exec_op(&op);
+    assert(op.retval == op.len);
+
+    {
+        uint64_t version = 0;
+        memset(bmp, 0, sizeof(bmp));
+        assert(test.bs->read_bitmap(op.oid, UINT64_MAX, bmp, &version) == 0);
+        assert(memcheck(bmp, 0xDE, test.dsk().clean_entry_bitmap_size));
+        assert(version == 3);
+    }
+
+    free(op.buf);
+}
+
+// used_blocks and inode_space_stats count one data block per object and only when a big_write
+// or a delete is marked stable. counters should persist correctly over a restart.
+static void test_used_blocks_replay_stable_over_small_write()
+{
+    printf("\n-- test_used_blocks_replay_stable_over_small_write\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint64_t block_count = test.dsk().block_count;
+    assert(!test.used_blocks());
+
+    blockstore_op_t op;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    memset(op.buf, 0xAA, 128*1024);
+
+    // Unstable big write. BS_OP_WRITE (not WRITE_STABLE) so that it needs an explicit
+    // BS_OP_STABLE and stays in dirty_db until then.
+    printf("write v1 0+128k (big)\n");
+    op.opcode = BS_OP_WRITE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    // Small write - the entry that used to be misread as "the block is already counted"
+    printf("write v2 0+4k (small)\n");
+    op.version = 2;
+    op.len = 4096;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("write v3 0+128k (big)\n");
+    op.version = 3;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+
+    printf("stabilize v3\n");
+    op.opcode = BS_OP_STABLE;
+    op.len = 1;
+    *((obj_ver_id*)op.buf) = (obj_ver_id){ .oid = { .inode = 1, .stripe = 0 }, .version = 3 };
+    test.exec_op(&op);
+    assert(op.retval == 0);
+    // The live path walks the whole range, so it has always got this one right
+    printf("used_blocks after stabilize = %ju (expected 1)\n", test.used_blocks());
+    assert(test.used_blocks() == 1);
+    // Nothing must have been flushed - the journal alone has to reproduce the counters below
+    assert(test.flusher()->get_queue_size());
+
+    printf("restarting the blockstore (journal replay)\n");
+    test.destroy_bs();
+    test.init();
+    uint64_t after_replay = test.used_blocks();
+    printf("used_blocks after replay = %ju (expected 1)\n", after_replay);
+
+    // Deleting the object must bring the counter back to zero, not below it
+    printf("delete v4\n");
+    op.opcode = BS_OP_DELETE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 4;
+    op.offset = 0;
+    op.len = 0;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+    uint64_t after_delete = test.used_blocks();
+    uint64_t free_after_delete = test.free_block_count();
+    printf("used_blocks after delete = %ju (expected 0), free_block_count = %ju (expected %ju)\n",
+        after_delete, free_after_delete, block_count);
+
+    assert(after_replay == 1);
+    assert(after_delete == 0);
+    assert(free_after_delete == block_count);
+
+    free(op.buf);
+}
+
+// The same accounting rule for deletes. used_space counter must not underflow after a restart.
+static void test_used_blocks_replay_delete_over_unstable_big_write()
+{
+    printf("\n-- test_used_blocks_replay_delete_over_unstable_big_write\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint64_t block_count = test.dsk().block_count;
+    assert(!test.used_blocks());
+
+    blockstore_op_t op;
+    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
+    memset(op.buf, 0xAA, 128*1024);
+
+    printf("write v1 0+128k (big, left unstable)\n");
+    op.opcode = BS_OP_WRITE;
+    op.oid = { .inode = 1, .stripe = 0 };
+    op.version = 1;
+    op.offset = 0;
+    op.len = 128*1024;
+    test.exec_op(&op);
+    assert(op.retval == (int)op.len);
+    // An unstable big write owns a block but is not counted yet - that is what mark_stable() is for
+    assert(!test.used_blocks());
+
+    printf("delete v2\n");
+    op.opcode = BS_OP_DELETE;
+    op.version = 2;
+    op.offset = 0;
+    op.len = 0;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+    assert(!test.used_blocks());
+
+    printf("restarting the blockstore (journal replay)\n");
+    test.destroy_bs();
+    test.init();
+    uint64_t after_replay = test.used_blocks();
+    uint64_t free_after_replay = test.free_block_count();
+    printf("used_blocks after replay = %ju (expected 0), free_block_count = %ju (expected %ju)\n",
+        after_replay, free_after_replay, block_count);
+    assert(after_replay == 0);
+    assert(free_after_replay == block_count);
+
+    free(op.buf);
+}
+
 int main(int narg, char *args[])
 {
     test_preserve_corruption();
@@ -863,5 +1207,9 @@ int main(int narg, char *args[])
     test_read_retry_on_ring_full_1M_csum4k_clean();
     test_read_free_clean_loc_used();
     test_meta_tail_reload();
+    test_write_no_space_eagain();
+    test_read_bitmap_of_deleted_object();
+    test_used_blocks_replay_stable_over_small_write();
+    test_used_blocks_replay_delete_over_unstable_big_write();
     return 0;
 }