Do not check IOV_MAX

...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_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/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)

src/blockstore/v1/impl.h

@@ -238,7 +238,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

@@ -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/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/stable.cpp

@@ -463,6 +463,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 +471,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 +525,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,10 +537,20 @@ 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() &&

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

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/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_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);
     }
 }
 

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;

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_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,250 @@ static void test_meta_tail_reload()
     printf("blockstore re-initialized with an overflowing entry\n");
 }
 
+// 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 +1138,8 @@ 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_read_bitmap_of_deleted_object();
+    test_used_blocks_replay_stable_over_small_write();
+    test_used_blocks_replay_delete_over_unstable_big_write();
     return 0;
 }