osd: Stabilize unstable writes of a removed pool before stopping its PGs

stop_pg() forgets the PG's unstable writes and leaves them to the next
primary, which stabilizes or rolls them back during peering. But nobody
peers a removed pool again, so its unstable writes stay on the OSDs
forever. The old store can't trim its journal past them: the journal
fills up and writes to all other pools on the same OSDs fail with
-EAGAIN, restarts don't help.

The monitor pauses PGs of a removed pool and waits for them to go
offline before it forgets them. Make the primary of such a PG stabilize
its unstable writes first, while the PG is still active and locked: it
refuses new writes, waits for the ones in flight, runs a sync, and then
stops the PG whatever the result. Pauses of existing pools (PG count
changes) stop at once as before.

.gitea/workflows/test.yml

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

CMakeLists.txt

@@ -2,7 +2,7 @@ cmake_minimum_required(VERSION 2.8...3.30)
 
 project(vitastor)
 
-set(VITASTOR_VERSION "3.2.2")
+set(VITASTOR_VERSION "3.2.3")
 
 include(CTest)
 

README-ru.md

@@ -75,6 +75,9 @@ Vitastor поддерживает QEMU-драйвер, протоколы UBLK,
   - [NFS](docs/usage/nfs.ru.md) кластерная файловая система и псевдо-ФС прокси
   - [Безопасность](docs/usage/security.ru.md)
   - [Администрирование](docs/usage/admin.ru.md)
+- Тесты
+  - [Хаос тест хранилища](docs/tests/blockstore-chaos.ru.md)
+  - [Хаос тест PG](docs/tests/pg-chaos.ru.md)
 - Производительность
   - [Понимание сути производительности](docs/performance/understanding.ru.md)
   - [Теоретический максимум](docs/performance/theoretical.ru.md)

README.md

@@ -75,6 +75,9 @@ Read more details in the documentation. You can start from here: [Quick Start](d
   - [NFS](docs/usage/nfs.en.md) clustered file system and pseudo-FS proxy
   - [Security](docs/usage/security.en.md)
   - [Administration](docs/usage/admin.en.md)
+- Tests
+  - [Blockstore chaos test](docs/tests/blockstore-chaos.en.md)
+  - [PG chaos test](docs/tests/pg-chaos.en.md)
 - Performance
   - [Understanding storage performance](docs/performance/understanding.en.md)
   - [Theoretical performance](docs/performance/theoretical.en.md)

csi/Makefile

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

csi/deploy/004-csi-nodeplugin.yaml

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

csi/deploy/007-csi-provisioner.yaml

@@ -87,7 +87,7 @@ spec:
             - name: socket-dir
               mountPath: /csi
         - name: csi-attacher
-          image: k8s.gcr.io/sig-storage/csi-attacher:v3.2.2
+          image: k8s.gcr.io/sig-storage/csi-attacher:v3.2.3
           args:
             - "--v=5"
             - "--csi-address=$(ADDRESS)"
@@ -121,7 +121,7 @@ spec:
             privileged: true
             capabilities:
               add: ["SYS_ADMIN"]
-          image: vitalif/vitastor-csi:v3.2.2
+          image: vitalif/vitastor-csi:v3.2.3
           args:
             - "--node=$(NODE_ID)"
             - "--endpoint=$(CSI_ENDPOINT)"

csi/src/config.go

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

debian/changelog

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

docker/Makefile

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

docker/etc/vitastor/docker.conf

@@ -4,7 +4,7 @@
 #
 
 # Desired Vitastor version
-VITASTOR_VERSION=v3.2.2
+VITASTOR_VERSION=v3.2.3
 
 # Additional arguments for all containers
 # For example, you may want to specify a custom logging driver here

docs/installation/docker.en.md

@@ -26,9 +26,9 @@ at Vitastor Kubernetes operator: https://github.com/Antilles7227/vitastor-operat
 The instruction is very simple.
 
 1. Download a Docker image of the desired version: \
-   `docker pull vitalif/vitastor:v3.2.2`
+   `docker pull vitalif/vitastor:v3.2.3`
 2. Install scripts to the host system: \
-   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.2 install.sh`
+   `docker run --rm -it -v /etc:/host-etc -v /usr/bin:/host-bin vitalif/vitastor:v3.2.3 install.sh`
 3. Reload udev rules: \
    `udevadm control --reload-rules`
 4. Enable the vitastor-host service: \

docs/installation/docker.ru.md

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

docs/tests/blockstore-chaos.en.md

@@ -0,0 +1,229 @@
+[Documentation](../../README.md#documentation) → Tests → The blockstore chaos-test model
+
+-----
+
+[Читать на русском](blockstore-chaos.ru.md)
+
+# The blockstore chaos-test model
+
+A seed-driven crash test of the blockstore (`src/test/test_blockstore_chaos.cpp`):
+the real blockstore - journal, metadata, flusher, compaction, checksums, recovery -
+running on a mock disk that lives entirely in RAM. Nothing above it exists: no OSD,
+no PG, no network, no etcd. The test talks to the blockstore directly and checks
+only what the storage itself promises. It is the counterpart of the
+[PG chaos test](pg-chaos.en.md), which assumes the storage is correct.
+
+Everything BELOW the blockstore is simulated: io_uring completions come back out of
+submission order after a random delay, a volatile write cache holds part of every
+write, and a power outage makes the writes that were in flight land fully, torn at a
+sector boundary, or not at all, while a random half of the cache disappears. The
+model of what was written is kept alongside, and after every outage the recovered
+state is checked against it.
+
+The same workload and the same checks run against both implementations - v1
+(meta_format 2) and v2 (heap, meta_format 3) - which are supposed to have identical
+semantics. The seed also picks the configuration, so a failing run is replayed by
+passing back the seed it printed; `-h` lists the options.
+
+The data is four 128 K objects (inode 1, four stripes) on one 256 MB mock disk that
+carries metadata at offset 0, the journal from 16 M and the data from 32 M; write
+granularity is 4 K.
+
+## Configurations (24)
+
+- v1 / v2 blockstore (meta_format 2 / 3) ×
+- immediate_commit none/all ×
+- `csum_block` none / 4096 / 16384 - 16384 is deliberately larger than the 4 K write
+  granularity, so a write covers only part of a checksummed block ×
+- instant/two-phase writes, like in replicated/EC pools (BS_OP_WRITE_STABLE / BS_OP_WRITE+BS_OP_STABLE)
+
+2 × 2 × 3 × 2 = 24; the configuration is `seed % 24`, `--config` overrides it. The
+two write styles can't be mixed inside one object: a stable version over an unstable
+one is rejected with EINVAL, and a pool is always one or the other.
+
+## The world model
+
+Per object the model keeps `submitted` (the highest sequence ever enqueued), `acked`,
+`synced` and `durable`, plus maps from the sequence number to the full 128 K content
+(an empty vector means "deleted"), to the version the store actually sees, and to
+the bitmap byte. Sequence numbers only ever grow and key the model, because version
+numbers restart from 1 after a deletion - the store explicitly allows writing low
+versions over a delete. `deleted_upto` stops an old sync snapshot from making a
+wiped-out version stabilizable, and `can_rollback` records that this instance really
+did stabilize the durable version (recovery may have merged it into one clean entry).
+
+Durability is derived purely from completions - there are no extra flags:
+
+- with `immediate_commit=all` a completed write is immediately synced; with a
+  volatile disk cache nothing is synced until a SYNC that covers it completes;
+- a synced deletion and a synced `BS_OP_WRITE_STABLE` write are stable by themselves
+  and so are durable at once; a two-phase write still has to be stabilized, and that
+  happens asynchronously, from a later step.
+
+Object contents are self-identifying: 32-byte records `{seq, version, object,
+offset}` whose size divides the 4 K granularity, so a record never straddles a write
+or a read, and any mismatch decodes into "which write put these bytes here".
+
+External bitmaps are self-identifying too: every write fills its 4-byte bitmap (one
+bit per 4 K granule) with one recognisable byte. The blockstore treats the bitmap as
+opaque, but it travels with the entry through compaction, recovery and rollbacks
+just like the data does.
+
+Invariant: only one operation at a time may touch an object. While its write, read,
+stabilize or rollback is in flight nothing else may go near it, so the base content
+of a write and the expected answer of a read cannot change underneath them.
+
+## The simulated reality
+
+- **Event loop**: one step is the completion of the earliest in-flight disk request,
+  and the virtual clock jumps straight to it, so the configured latencies don't make
+  the test slow. A single seeded PRNG drives the latencies, the workload and the
+  power-outage dice alike.
+- **Disk simulator**: per-request delays are drawn uniformly (10-200 us for reads
+  and writes, 100-2000 us for fsync) - that is what makes completions arrive out of
+  order; 4 K sector; writes up to 4 K are atomic and land whole or not at
+  all. In buffered mode (`immediate_commit=none`) writes go into a volatile cache
+  and only reach the media on fsync. On `power_loss()` every in-flight write
+  lands fully, torn at a sector boundary, or not at all, and every cached write
+  survives with probability 50% - the survivors are committed to the disk and the
+  cache is emptied, so nothing from it disappears at the next outage. Error injection
+  for read/write/fsync exists but is off by default: the blockstore treats most write
+  errors as fatal, by design.
+- **Ringloop stub**: an unknown fd aborts the test at once; `reset()` drops queued
+  and pending completions as if the process had died, so no stale callback can fire
+  into a destroyed blockstore. A power outage empties the disk's in-flight queue by
+  definition, and a clean restart waits for it to go idle explicitly.
+
+## Operations (one per step)
+
+Each step picks an action by weight (`--p-*`; 0 disables it):
+
+- **wait (500)** - submit nothing, just let the simulation run one event.
+- **write (60)** - write a random 4 K-aligned range of a random free object: a new
+  version (numbering restarts after a deletion), self-describing payload, its own
+  bitmap byte.
+- **delete (6)** - delete an existing object; deleting a missing one is a no-op the
+  store answers without creating an entry, so it isn't exercised. A deletion is
+  stable by itself.
+- **read (40)** - read a random 4 K-aligned range - deliberately NOT aligned to the
+  checksum block, so reads cut across checksum blocks and across the boundaries of
+  the extents that were actually written - or the whole object with `full_read`%
+  (25) chance; the answer is verified against the model when it completes.
+- **sync (30)** - flush; what it covers is the set of acknowledged versions as of
+  submit time, treated as a lower bound. Not submitted while a rollback is in
+  flight: a rollback rewinds the test model - `acked` and `synced` drop back to the
+  stable version, everything above it disappears from it - so the snapshot would
+  name versions that no longer exist. This is a limit of the test model, not of the
+  blockstore API.
+- **stabilize (50)** - commit a random non-empty prefix of everything that has
+  become stabilizable, as a single op carrying several (object, version) entries -
+  the shape an OSD uses; single-entry ops happen just as naturally.
+- **rollback (6)** - throw the unstable versions away instead of committing them.
+  Two-phase mode only (an instant write is stable at once), and only for objects
+  that have a stable non-deleted version below the newest one. On success everything
+  above the stable version is gone and the store's version counter for the object
+  goes back to it. Not submitted while a sync is in flight, for the same reason,
+  nor while another rollback is in flight.
+- **restart (1)** - a CLEAN restart: drain everything in flight, read all objects,
+  wait for the disk to go idle, destroy the blockstore, create it again, read
+  everything again.
+- **crash (2)** - pull the plug right now: power loss, whatever was in flight is
+  accounted as a legal-but-unconfirmed outcome, the blockstore is destroyed and
+  recreated, every object is checked against the recovery rule.
+
+At most `max_inflight` (8) operations are in flight at once; a step that finds the
+queue full waits for one completion instead of submitting anything - that is where
+hangs are checked.
+
+At the end of the run everything still in flight is drained, and every object is
+read whole and verified.
+
+## Op outcome accounting
+
+- Completed with the expected retval - the content is recorded at its sequence
+  number, `acked` moves to it, and the follow-up (synced / durable / needs-stabilize)
+  is derived from the configuration.
+- A write or delete with `-EAGAIN` - a refusal, not a loss (the journal is full and
+  the flusher can't keep up under pressure); the model is unchanged.
+- A failed sync - it simply didn't sync anything; also legal under pressure.
+- A read interrupted by an outage tells nothing and is dropped.
+- Anything else - including a failed stabilize or rollback - aborts the test with
+  the affected (object, version) entries printed.
+- A write that was in flight at the moment of the outage: its content is recorded as
+  a legal outcome all the same, but `acked` doesn't move, so it may still be lost.
+
+## Checks (every abort point)
+
+**Data consistency:**
+
+1. a read of a deleted or never-written object returns `-ENOENT`;
+2. a read returns exactly the acknowledged content: `retval == len` and byte-for-byte
+   equality over the whole range; the first differing record is printed with both
+   obj/seq/version/offset values, so stale data names its own author;
+3. the returned external bitmap must belong to the returned version ("returned the
+   bitmap of another version");
+4. after a crash or a restart a read answers either the whole object or `-ENOENT` -
+   any other retval means torn data was committed;
+5. whatever came back must be the content of a version at or above the guaranteed
+   floor: the last durable version after a power outage, the last acknowledged one
+   (the last synced one with a volatile cache) after a restart that lost nothing.
+   Anything else aborts the test with "came back ... holding unexpected content"
+   plus a diff against every candidate version - the one that almost matches usually
+   tells which write was lost or applied out of order;
+6. the object may legitimately be gone only when nothing was guaranteed at all; then
+   the model resets and numbering starts over from 1;
+7. with `immediate_commit=all` a clean restart must return exactly the same state -
+   same retval, same version, same bytes; the failure also prints the first
+   never-written journal sector, which points straight at the lost record;
+8. what came back becomes the new baseline but not the new durable point: the
+   durable content stays a legal outcome, `can_rollback` is cleared, and the next
+   outage may still roll the object back to the durable version;
+9. a rollback may only remove versions above the stable one, and the version counter
+   is rewound accordingly - later writes and reads would notice otherwise;
+10. the final whole-object reads must match the acknowledged content.
+
+**Liveness:**
+
+11. "operations did not complete in 1000000 steps - deadlock?" - the stuck
+    operations are printed together with the store's diagnostics
+    (completed/fsynced/compacted LSNs, compact queue) in the v2 implementation;
+12. "blockstore failed to start";
+13. "a read of object N did not complete in MAX_STEPS - deadlock?" and "the final
+    read of object N did not complete";
+14. "disk did not go idle before a clean restart" - a use-after-free guard: no
+    request may outlive the blockstore that submitted it.
+
+**Failure accounting:**
+
+15. only `-EAGAIN` is accepted as a refusal for a write/delete, a failed sync is
+    treated as "synced nothing", and any other error, including a failed stabilize
+    or rollback, aborts the test.
+
+**Test-stub guards:**
+
+16. an unexpected fd on the ringloop stub, a rejected SQE, `--max-inflight < 1`;
+17. overlapping concurrent writes: a write submitted while another in-flight write
+    still covers part of its range aborts the test - a device may apply such a pair
+    in either order, so relying on the newer one winning is a bug, and catching it
+    at submission time is deterministic instead of waiting for an unlucky completion
+    order.
+
+## Reporting and reproducibility
+
+By default the test runs 48 fixed seeds (every configuration twice) plus 152 random
+ones spread evenly over the configurations - 200 seeds × 10000 steps in total. Every
+seed runs in its own child process, so one abort doesn't take the whole sweep down,
+and the failing seeds are listed at the end. `--seed N` runs a single seed
+in-process, which is what you want under a debugger; `--seeds a-b`, `--random n`,
+`--steps n`, `--config n` and the `--p-*` weights shape the run (e.g.
+`--p-delete 0 --p-rollback 0` gives a workload without deletions and rollbacks).
+
+The OK line: writes (refused with EAGAIN) / deletes / syncs / stabilizes / rollbacks
+/ crashes / clean restarts / verified reads / disk ops.
+
+Determinism: one seed -> the same configuration and the same schedule; `-v` traces
+every operation and every simulated disk request, including its simulated latency,
+the power-loss decision taken for every in-flight write and every cached write
+dropped from the volatile cache.
+
+Built by `make build_tests` and run by ctest as `test_blockstore_chaos`.

docs/tests/blockstore-chaos.ru.md

@@ -0,0 +1,211 @@
+[Документация](../../README-ru.md#документация) → Тесты → Модель хаос-теста blockstore (`src/test/test_blockstore_chaos.cpp`)
+
+-----
+
+[Read in English](blockstore-chaos.en.md)
+
+# Модель хаос-теста blockstore
+
+Seed-driven тест на отказа питания (`src/test/test_blockstore_chaos.cpp`): настоящее хранилище —
+журнал, метаданные, flusher, compaction, контрольные суммы, recovery — работает на заглушке диска,
+полностью в RAM. Ничего над ним нет: ни OSD, ни PG, ни сети, ни etcd. Тест обращается к хранилищу
+напрямую и проверяет только то, что обещает само хранилище. Это пара к
+[хаос-тесту PG](pg-chaos.ru.md), который считает хранилище корректным.
+
+Всё, что НИЖЕ хранилища, симулируется: ответы io_uring приходят не в порядке отправки и со
+случайной задержкой, волатильный кэш записи удерживает часть записей, а потеря питания делает так,
+что бывшие в полёте записи сохраняются целиком, рвутся по границе сектора или не сохраняются вовсе,
+случайная половина кэша исчезает. Рядом ведётся модель того, что было записано, и после
+каждой имитации отказа восстановленное состояние сверяется с ней.
+
+Один и тот же набор нагрузок и проверок прогоняется на обеих реализациях — v1 (meta_format 2)
+и v2 (heap, meta_format 3), которые обязаны вести себя одинаково. Сид также задаёт
+конфигурацию, поэтому упавший прогон воспроизводится тем же сидом; опции — по `-h`.
+
+Данные: четыре объекта по 128 К (inode 1, четыре stripe) на одном mock-диске в 256 МБ, где
+метаданные лежат с нуля, журнал — с 16 М, данные — с 32 М; гранула записи 4 К.
+
+## Конфигурации (24)
+
+- blockstore v1 / v2 (meta_format 2 / 3) ×
+- immediate_commit none/all ×
+- csum_block нет / 4096 / 16384 — 16384 намеренно больше 4 К гранулы записи, то есть запись
+  покрывает лишь часть чексуммного блока ×
+- мгновенная/двухфазная запись, как в репликах/EC (BS_OP_WRITE_STABLE / BS_OP_WRITE+BS_OP_STABLE)
+
+2 × 2 × 3 × 2 = 24; конфигурация = `seed % 24`, переопределяется через `--config`. Смешать два
+стиля записей в одном объекте нельзя: stable-версия поверх unstable отклоняется с EINVAL, а пул —
+всегда что-то одно.
+
+## Модель мира
+
+Для каждого объекта хранятся `submitted` (максимальная отправленная последовательность), `acked`,
+`synced`, `durable`, а также отображения номер-последовательности → полное содержимое 128 К
+(пустой вектор = «удалён»), → номер версии, который видит хранилище, и → байт битмапа.
+Номер последовательности только растёт и именно он является ключом модели, потому что после
+удаления версии снова начинаются с 1 — хранилище явно разрешает писать низкие версии поверх
+удаления. `deleted_upto` не даёт старому снимку sync сделать stabilizable стёртую версию, а
+`can_rollback` помечает, что именно этот экземпляр действительно стабилизировал durable-версию
+(recovery мог слить её в одну чистую запись).
+
+Зафиксированность выводится только из завершения операций, дополнительных флагов нет:
+
+- при `immediate_commit=all` завершённая запись сразу считается synced; при волатильном кэше
+  диска — ничто не synced, пока не завершится накрывающий её SYNC;
+- synced-удаление и synced-запись `BS_OP_WRITE_STABLE` стабильны сами по себе и сразу становятся
+  durable; двухфазную запись надо ещё стабилизировать, и это происходит асинхронно, позже, из
+  другого шага.
+
+Содержимое объектов самоидентифицируется: 32-байтные записи `{seq, версия, объект, смещение}`,
+размер которых делит гранулу 4 К, поэтому такая запись никогда не пересекает границу операции
+записи или чтения, и любое расхождение декодируется в «какая версия записала эти байты».
+
+Внешние битмапы тоже самоидентифицируются: каждая запись заполняет свой 4-байтный битмап
+(один бит на 4 К гранулу) одним узнаваемым байтом. Для хранилища битмап непрозрачен, но он
+путешествует вместе с записью через compaction, recovery и откаты так же, как данные.
+
+Инвариант: за раз с объектом работает только одна операция. Пока в полёте его запись, чтение,
+stabilize или rollback, ничто другое к нему не подходит — поэтому базовое содержимое записи и
+ожидаемый ответ чтения не могут измениться под ними.
+
+## Симулируемая реальность
+
+- **Цикл событий**: один шаг — это завершение самого раннего запроса диска, и виртуальные часы
+  прыгают прямо к нему, поэтому заданные задержки не замедляют тест. Один seed-driven PRNG
+  задаёт и задержки, и нагрузку, и броски «костей» при отказе питания.
+- **Симулятор диска**: задержка на каждый запрос берётся равномерно (10–200 мкс на чтение и
+  запись, 100–2000 мкс на fsync) — именно из-за разброса completion'ы приходят не по порядку;
+  сектор 4 К; записи до 4 К атомарны и ложатся на диск целиком или не ложатся вовсе. В буферном
+  режиме (`immediate_commit=none`) записи попадают в волатильный кэш и доходят до носителя только
+  по fsync. При `power_loss()` каждая запись в полёте сохраняется целиком, рвётся по границе сектора или
+  не сохранятся вовсе, а записи в кэше выживают и сохраняются на диск с вероятностью 50%, после
+  чего кэш очищается, т.к. при следующем запуске он должен быть пустым и не исчезать при ещё одном
+  отказе. Инъекция ошибок read/write/fsync существует, но по умолчанию выключена: хранилище по замыслу
+  считает большинство ошибок записи фатальными.
+- **Заглушка ringloop**: неизвестный fd немедленно обрывает тест; `reset()` выбрасывает очереди и
+  ожидающие completion'ы, как если бы процесс умер, — ни один «просроченный» колбэк не может
+  прилететь в уничтоженное хранилище. Отказ питания по определению очищает очередь in-flight диска,
+  а чистая перезагрузка дожидается простоя диска явно.
+
+## Операции (одна на шаг)
+
+На каждом шаге действие выбирается по весу (`--p-*`; 0 выключает его):
+
+- **wait (500)** — ничего не слать, прокрутить одно событие симуляции.
+- **write (60)** — запись случайного выровненного по 4 К диапазона случайного свободного объекта:
+  новая версия (нумерация после удаления начинается заново), самоидентифицируемое содержимое, свой
+  битмап.
+- **delete (6)** — удаление существующего объекта; удаление несуществующего — no-op, на который
+  хранилище отвечает без создания записи, поэтому он не тестируется. Удаление стабильно само по себе.
+- **read (40)** — чтение случайного диапазона: выровненного по 4 К, но — назло — не выровненного по
+  размеру чексуммного блока, чтобы чтения резали и чексуммные блоки, и границы реально записанных
+  экстентов; либо целиком с вероятностью `full_read`% (25). Ответ проверяется по модели при
+  завершении.
+- **sync (30)** — сброс кэша; покрытие — множество подтверждённых версий на момент отправки, оно
+  трактуется как нижняя граница. Не отправляется, пока в полёте rollback: rollback откатывает
+  модель теста — `acked` и `synced` возвращаются к стабильной версии, и всё выше неё из модели
+  исчезает, — поэтому снимок покрытия начал бы ссылаться на версии, которых уже нет. Это
+  ограничение модели теста, а не API хранилища.
+- **stabilize (50)** — зафиксировать случайный непустой префикс всего ставшего stabilizable одной
+  операцией с несколькими парами (объект, версия) — ровно так, как делает OSD; одноэлементные
+  операции при этом возникают так же естественно.
+- **rollback (6)** — выбросить нестабильные версии вместо фиксации. Только двухфазный режим
+  (мгновенная запись стабильна сразу) и только для объектов со стабильной неудалённой версией ниже
+  последней. После успеха всё выше стабильной версии исчезает, и счётчик версий объекта в
+  хранилище откатывается к ней. Не отправляется, пока в полёте sync (по той же причине) и пока
+  в полёте другой rollback.
+- **restart (1)** — ЧИСТАЯ перезагрузка: дождаться завершения всего в полёте, прочитать все
+  объекты, дождаться простоя диска, уничтожить хранилище, создать заново, прочитать снова.
+- **crash (2)** — выдернуть шнур прямо сейчас: потеря питания, всё бывшее в полёте учитывается как
+  легальный, но неподтверждённый исход, хранилище уничтожается и создаётся заново, каждый объект
+  проверяется по правилу восстановления.
+
+Одновременно выполняется не более `max_inflight` (8) операций; шаг при полной очереди вместо
+отправки ждёт завершения одной из них — здесь проверяется зависание.
+
+В конце работы всё оставшееся в полёте дожидается, после чего каждый объект читается целиком и
+проверяется.
+
+## Учёт результатов операций (account)
+
+- Завершилась с ожидаемым кодом возврата — содержимое записано в модель под своей
+  последовательностью, `acked` сдвинут, а дальнейшее (synced / durable / требует stabilize)
+  выводится из конфигурации.
+- Запись или удаление с `-EAGAIN` — отказ, а не потеря (журнал заполнен, flusher не успевает под
+  нагрузкой); модель не меняется.
+- Упавший sync — значит, он ничего и не синхронизировал; это тоже легально под нагрузкой.
+- Чтение, прерванное отказом питания, ничего не сообщает и просто отбрасывается.
+- Всё остальное, включая упавшие stabilize/rollback, — падение теста с печатью затронутых пар
+  (объект, версия).
+- Запись, бывшая в полёте в момент отказа: её содержимое так же записывается в модель как легальный
+  исход, но `acked` не сдвигается, то есть она всё ещё может быть потеряна.
+
+## Проверки (все точки отмены)
+
+**Согласованность данных:**
+
+1. чтение удалённого или никогда не писанного объекта обязано вернуть `-ENOENT`;
+2. чтение обязано вернуть ровно подтверждённое содержимое: `retval == len` и побайтовое совпадение
+   по всему диапазону; первая расходящаяся 32-байтная запись печатается с obj/seq/version/offset и
+   ожидаемым значением — то есть устаревшие данные сами называют своего автора;
+3. внешний битмап в ответе обязан принадлежать именно возвращённой версии
+   («returned the bitmap of another version»);
+4. после отказа питания или перезапуска чтение отвечает либо целым объектом, либо `-ENOENT` —
+   любой иной retval означает, что на диск попало порванное содержимое;
+5. то, что объект «вернулся», обязано быть содержимым версии не ниже гарантированной границы:
+   последней durable — после отказа питания, последней подтверждённой (последней synced при
+   волатильном кэше) — после перезапуска, который ничего не потерял. Иначе — падение с
+   «came back ... holding unexpected content» и диффом по каждой версии-кандидату: та, что почти
+   совпадает, обычно и показывает, какая запись потерялась или применилась не в том порядке;
+6. объект вправе полностью исчезнуть только тогда, когда ничего не гарантировалось; тогда модель
+   сбрасывается, и нумерация начинается с 1;
+7. при `immediate_commit=all` чистая перезагрузка обязана вернуть ровно то же состояние — тот же
+   retval, та же версия, те же байты; в сообщении о падении печатается ещё и первый никогда не
+   писанный сектор журнала, который прямо указывает на потерянную запись;
+8. вернувшееся содержимое становится новой базой, но не новой durable-точкой: durable-содержимое
+   остаётся легальным исходом, `can_rollback` сбрасывается, и следующий отказ всё ещё может
+   откатить объект к durable-версии;
+9. rollback вправе удалить только версии выше стабильной, и счётчик версий откатывается
+   соответственно — иначе это заметят последующие записи и чтения;
+10. финальные чтения целых объектов обязаны совпасть с подтверждённым содержимым.
+
+**Живость:**
+
+11. «operations did not complete in 1000000 steps — deadlock?» — зависшие операции печатаются
+    вместе с диагностикой хранилища (completed/fsynced/compacted LSN, очередь compact) в v2;
+12. «blockstore failed to start»;
+13. «a read of object N did not complete in MAX_STEPS — deadlock?» и «the final read of object N
+    did not complete»;
+14. «disk did not go idle before a clean restart» — защита от use-after-free: ни один запрос не
+    может пережить хранилище, которое его послало.
+
+**Учёт отказов:**
+
+15. отказом записи/удаления принимается только `-EAGAIN`, упавший sync трактуется как «ничего не
+    синхронизировал», а любая другая ошибка — включая stabilize/rollback — обрывает тест.
+
+**Защиты тестовых заглушек:**
+
+16. неожиданный fd в заглушке ringloop, отклонённый SQE, `--max-inflight < 1`;
+17. перекрывающиеся одновременные записи: запись, отправленная, пока в полёте находится другая,
+    накрывающая часть её диапазона, — падение теста. Устройство вправе применить такую пару в
+    любом порядке, поэтому полагаться на то, что победит более новая, — ошибка; ловить это на
+    отправке детерминированно лучше, чем ждать неудачного порядка завершения.
+
+## Отчёт и воспроизводимость
+
+По умолчанию тест прогоняет 48 фиксированных сидов (каждая конфигурация дважды) плюс 152 случайных,
+равномерно разложенных по конфигурациям, — всего 200 сидов × 10000 шагов. Каждое сид выполняется в
+отдельном дочернем процессе, поэтому одно падение не роняет весь прогон, а упавшие сиды печатаются
+в конце. `--seed N` выполняет одно сид в том же процессе — именно это нужно под отладчиком;
+`--seeds a-b`, `--random n`, `--steps n`, `--config n` и веса `--p-*` настраивают прогон
+(например, `--p-delete 0 --p-rollback 0` даёт нагрузку без удалений и откатов).
+
+OK-строка: writes (refused with EAGAIN) / deletes / syncs / stabilizes / rollbacks / crashes /
+clean restarts / verified reads / disk ops.
+
+Детерминизм: один сид ⇒ та же конфигурация и то же расписание; `-v` даёт трассу каждой операции и
+каждого запроса к диску, включая смоделированную задержку, решение при отказе питания по каждой
+бывшей в полёте записи и каждую запись, выброшенную из волатильного кэша.
+
+Собирается через `make build_tests`, запускается ctest'ом как `test_blockstore_chaos`.

docs/tests/pg-chaos.en.md

@@ -18,7 +18,7 @@ the same `--seed`.
 What it mainly verifies is the PG machinery (the subop fan-out,
 peering/LIST/rollback/stabilize, repeer daemons, stop_pg, INCOMPLETE, recovery) and
 partly the primary OSD's storage; checking the storage is not the test's main goal -
-test_blockstore_chaos covers it.
+it is covered by the [blockstore chaos test](blockstore-chaos.en.md).
 
 ## Configurations (10)
 

docs/tests/pg-chaos.ru.md

@@ -15,7 +15,7 @@ EC 2+2 (members 2..4, spare 5, объект = один parity-stripe). Сид о
 
 Проверяется в первую очередь PG-машинерия (subop-веер, peering/LIST/rollback/stabilize, repeer-демоны,
 stop_pg, INCOMPLETE, recovery) и частично хранилище первичного OSD; однако, проверка хранилища —
-не главная цель теста, его покрывает test_blockstore_chaos.
+не главная цель теста, его покрывает [хаос-тест blockstore](blockstore-chaos.ru.md).
 
 ## Конфигурации (10)
 

mon/package.json

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

node-binding/package.json

@@ -1,6 +1,6 @@
 {
   "name": "vitastor",
-  "version": "3.2.2",
+  "version": "3.2.3",
   "description": "Low-level native bindings to Vitastor client library",
   "main": "index.js",
   "keywords": [

patches/cinder-vitastor.py

@@ -50,7 +50,7 @@ from cinder.volume import configuration
 from cinder.volume import driver
 from cinder.volume import volume_utils
 
-VITASTOR_VERSION = '3.2.2'
+VITASTOR_VERSION = '3.2.3'
 
 LOG = logging.getLogger(__name__)
 

rpm/vitastor-el10.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.2.2
+Version:        3.2.3
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-3.2.2.el10.tar.gz
+Source0:        vitastor-3.2.3.el10.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-c++

rpm/vitastor-el7.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.2.2
+Version:        3.2.3
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-3.2.2.el7.tar.gz
+Source0:        vitastor-3.2.3.el7.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  devtoolset-9-gcc-c++

rpm/vitastor-el8.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.2.2
+Version:        3.2.3
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-3.2.2.el8.tar.gz
+Source0:        vitastor-3.2.3.el8.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-toolset-9-gcc-c++

rpm/vitastor-el9.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        3.2.2
+Version:        3.2.3
 Release:        1%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.1
 URL:            https://vitastor.io/
-Source0:        vitastor-3.2.2.el9.tar.gz
+Source0:        vitastor-3.2.3.el9.tar.gz
 
 BuildRequires:  gperftools-devel
 BuildRequires:  gcc-c++

src/CMakeLists.txt

@@ -24,7 +24,7 @@ set(ENABLE_COVERAGE false CACHE BOOL "Enable code coverage")
 set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
-add_definitions(-DVITASTOR_VERSION="3.2.2")
+add_definitions(-DVITASTOR_VERSION="3.2.3")
 add_definitions(-D_GNU_SOURCE -D_LARGEFILE64_SOURCE -D_FILE_OFFSET_BITS=64 -Wall -Wno-c99-designator -Wno-vla-cxx-extension -Wno-sign-compare -Wno-comment -Wno-parentheses -Wno-pointer-arith -fdiagnostics-color=always -fno-omit-frame-pointer -fvisibility=hidden -I ${CMAKE_SOURCE_DIR}/src)
 add_link_options(-fno-omit-frame-pointer)
 if (${WITH_ASAN})

src/client/cluster_client.cpp

@@ -12,23 +12,27 @@
 #define TRY_SEND_CONNECTING 1
 #define TRY_SEND_OK 2
 
-cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json config, std::unique_ptr<etcd_state_client_t> st_cli_ptr)
+cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json passed_config, std::unique_ptr<etcd_state_client_t> st_cli_ptr)
 {
     wb = new writeback_cache_t();
+    wb->parent = this;
 
-    cli_config = config.object_items();
-    file_config = osd_messenger_t::read_config(config);
+    cli_config = passed_config.object_items();
+    file_config = osd_messenger_t::read_config(passed_config);
     config = osd_messenger_t::merge_configs(cli_config, file_config, etcd_global_config, {});
 
+    this->exit_on_error = config.find("client_exit_on_error") == config.end()
+        ? true : config["client_exit_on_error"].bool_value();
     this->ringloop = ringloop;
     this->tfd = tfd;
 
-    msgr.osd_num = 0;
-    msgr.tfd = tfd;
-    msgr.ringloop = ringloop;
-    msgr.repeer_pgs = [this](osd_num_t peer_osd)
+    msgr = new osd_messenger_t();
+    msgr->osd_num = 0;
+    msgr->tfd = tfd;
+    msgr->ringloop = ringloop;
+    msgr->repeer_pgs = [this](osd_num_t peer_osd)
     {
-        if (msgr.osd_peers.find(peer_osd) != msgr.osd_peers.end())
+        if (msgr->osd_peers.find(peer_osd) != msgr->osd_peers.end())
         {
             // peer_osd just connected
             continue_ops();
@@ -45,14 +49,15 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
             }
         }
     };
-    msgr.exec_op = [this](osd_op_t *op)
+    msgr->exec_op = [this](osd_op_t *op)
     {
         // Garbage in
         fprintf(stderr, "Can't handle incoming operation from client %ju\n", op->client_id);
-        msgr.stop_client(op->client_id);
+        msgr->stop_client(op->client_id);
         delete op;
     };
-    msgr.parse_config(config, true);
+    msgr->on_fatal_error_hook = [this](int errcode, const std::string & error) { fatal_error(errcode, error); };
+    msgr->parse_config(config, true);
 
     st_cli = std::move(st_cli_ptr);
     st_cli->on_load_config_hook = [this](json11::Json::object & cfg) { on_load_config_hook(cfg); };
@@ -64,6 +69,7 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
     st_cli->on_load_pgs_hook = [this](bool success) { on_load_pgs_hook(success); };
     st_cli->on_reload_hook = [this]() { this->st_cli->load_global_config(); };
     st_cli->on_inode_change_hook = [this](uint64_t inode, bool removed) { on_change_inode_hook(inode, removed); };
+    st_cli->on_fatal_error_hook = msgr->on_fatal_error_hook;
 
     st_cli->parse_config(config);
     st_cli->infinite_start = false;
@@ -72,11 +78,21 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
         st_cli->infinite_start = config["client_infinite_start"].bool_value();
     }
     st_cli->load_global_config();
+
+    if (ringloop)
+    {
+        ringloop->loop();
+    }
+}
+
+void cluster_client_t::set_on_fatal_error_hook(std::function<void(int, const std::string&)> cb)
+{
+    on_fatal_error_hook = cb;
 }
 
 cluster_client_t::~cluster_client_t()
 {
-    vault_destroy();
+    aborted = ECANCELED;
     if (retry_timeout_id >= 0)
     {
         tfd->clear_timer(retry_timeout_id);
@@ -89,13 +105,44 @@ cluster_client_t::~cluster_client_t()
         list_retry_timeout_id = -1;
         list_retry_time = {};
     }
-    msgr.repeer_pgs = [](osd_num_t){};
+    msgr->repeer_pgs = [](osd_num_t){};
     if (ringloop)
     {
         ringloop->unregister_consumer(&consumer);
     }
+    delete msgr;
+    msgr = NULL;
+    while (op_queue_head)
+    {
+        op_queue_head->retval = -ECANCELED;
+        erase_op(op_queue_head);
+    }
     delete wb;
     wb = NULL;
+    vault_destroy();
+}
+
+void cluster_client_t::fatal_error(int errcode, const std::string & error)
+{
+    fprintf(stderr, "Error: %s%s\n", error.c_str(), exit_on_error ? ", exiting" : "");
+    if (exit_on_error)
+    {
+        // Simple old behaviour by default - stop the whole process
+        exit(1);
+    }
+    if (!pgs_loaded)
+    {
+        // Client becomes "ready" (but aborted) on error
+        for (auto fn: on_ready_hooks)
+            fn();
+        on_ready_hooks.clear();
+    }
+    if (!aborted && on_fatal_error_hook)
+    {
+        (std::function<void(int, const std::string&)>(on_fatal_error_hook))(errcode, error);
+    }
+    aborted = true;
+    abort_error = error;
 }
 
 cluster_op_t::~cluster_op_t()
@@ -150,13 +197,15 @@ std::vector<osd_num_t> cluster_op_t::get_left_on_dead()
 
 void cluster_client_t::continue_raw_ops(osd_num_t peer_osd)
 {
+    if (aborted)
+        return;
     auto it = raw_ops.find(peer_osd);
     while (it != raw_ops.end() && it->first == peer_osd)
     {
         auto op = it->second;
         op->op_type = OSD_OP_OUT;
-        op->client_id = msgr.osd_peers.at(peer_osd)->client_id;
-        msgr.outbox_push(op);
+        op->client_id = msgr->osd_peers.at(peer_osd)->client_id;
+        msgr->outbox_push(op);
         raw_ops.erase(it++);
     }
 }
@@ -165,14 +214,14 @@ void cluster_client_t::init_msgr()
 {
     if (msgr_initialized)
         return;
-    msgr.init();
+    msgr->init();
     msgr_initialized = true;
     if (ringloop)
     {
         consumer.loop = [this]()
         {
-            msgr.read_requests();
-            msgr.send_replies();
+            msgr->read_requests();
+            msgr->send_replies();
             this->ringloop->submit();
         };
         ringloop->register_consumer(&consumer);
@@ -482,7 +531,7 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_co
     }
     // vault
     vault_parse_config();
-    msgr.parse_config(config, false);
+    msgr->parse_config(config, false);
     st_cli->parse_config(config);
     st_cli->load_pgs();
 }
@@ -636,9 +685,9 @@ bool cluster_client_t::get_immediate_commit(uint64_t inode)
 void cluster_client_t::on_change_osd_state_hook(osd_num_t peer_osd)
 {
     osd_tree_metrics.erase(peer_osd);
-    if (msgr.wanted_peers.find(peer_osd) != msgr.wanted_peers.end())
+    if (msgr->wanted_peers.find(peer_osd) != msgr->wanted_peers.end())
     {
-        msgr.connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
+        msgr->connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
         continue_lists();
     }
 }
@@ -651,12 +700,21 @@ void cluster_client_t::on_change_node_placement_hook()
 
 bool cluster_client_t::is_ready()
 {
-    return pgs_loaded;
+    return pgs_loaded || aborted;
+}
+
+int cluster_client_t::is_failed(std::string & error)
+{
+    if (aborted)
+    {
+        error = abort_error;
+    }
+    return aborted;
 }
 
 void cluster_client_t::on_ready(std::function<void(void)> fn)
 {
-    if (pgs_loaded)
+    if (is_ready())
     {
         fn();
     }
@@ -855,6 +913,13 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
 
 void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
 {
+    if (aborted)
+    {
+        op->retval = -ECANCELED;
+        auto cb = std::move(op->callback);
+        cb(op);
+        return;
+    }
     if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1)
     {
         // Atomic writes to multiple stripes are unsupported
@@ -871,7 +936,14 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
         sync_op->flags = 0;
         sync_op->callback = [this, op](cluster_op_t* sync_op)
         {
-            execute_cas(op, true);
+            if (sync_op->retval != 0)
+            {
+                op->retval = sync_op->retval;
+                auto cb = std::move(op->callback);
+                cb(op);
+            }
+            else
+                execute_cas(op, true);
             delete sync_op;
         };
         execute_internal(sync_op);
@@ -887,18 +959,25 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
     op->needs_reslice = false;
     int res = try_send(op, 0, [this, op](osd_op_t *part)
     {
+        if (aborted)
+        {
+            op->retval = -ECANCELED;
+            auto cb = std::move(op->callback);
+            cb(op);
+            return;
+        }
         int expected = part->req.hdr.opcode == OSD_OP_DELETE ? 0 : part->req.rw.len;
         op->retval = part->reply.hdr.retval;
         if (op->retval != expected && op->retval >= 0)
             op->retval = -EIO;
         op->retval = op->retval == -EPIPE ? -EINTR : op->retval;
-        auto peer_it = msgr.osd_peers.find(op->parts[0].osd_num);
+        auto peer_it = msgr->osd_peers.find(op->parts[0].osd_num);
         if (op->retval != expected || (op->flags & OP_IMMEDIATE_COMMIT))
         {
             auto cb = std::move(op->callback);
             cb(op);
         }
-        else if (peer_it == msgr.osd_peers.end())
+        else if (peer_it == msgr->osd_peers.end())
         {
             // Care must be taken to make sure that the client doesn't reconnect to the OSD
             // before executing the previously completed operation callback (!)
@@ -930,7 +1009,7 @@ void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
                     cb(op);
                 },
             };
-            msgr.outbox_push(part);
+            msgr->outbox_push(part);
         }
     });
     if (res == TRY_SEND_CONNECTING || res == TRY_SEND_OFFLINE)
@@ -1061,23 +1140,25 @@ bool cluster_client_t::check_rw(cluster_op_t *op)
 
 void cluster_client_t::execute_raw(osd_num_t osd_num, osd_op_t *op)
 {
-    auto peer_it = msgr.osd_peers.find(osd_num);
-    if (peer_it != msgr.osd_peers.end())
+    auto peer_it = msgr->osd_peers.find(osd_num);
+    if (peer_it != msgr->osd_peers.end())
     {
         op->op_type = OSD_OP_OUT;
         op->client_id = peer_it->second->client_id;
-        msgr.outbox_push(op);
+        msgr->outbox_push(op);
     }
     else
     {
-        if (msgr.wanted_peers.find(osd_num) == msgr.wanted_peers.end())
-            msgr.connect_peer(osd_num, st_cli->peer_states[osd_num]);
+        if (msgr->wanted_peers.find(osd_num) == msgr->wanted_peers.end())
+            msgr->connect_peer(osd_num, st_cli->peer_states[osd_num]);
         raw_ops.emplace(osd_num, op);
     }
 }
 
 int cluster_client_t::continue_rw(cluster_op_t *op)
 {
+    if (aborted)
+        return 0;
     if (op->state == 0)
         goto resume_0;
     else if (op->state == 1)
@@ -1450,8 +1531,8 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
                 primary_osd = nearest_osd;
         }
         part->osd_num = primary_osd;
-        auto peer_it = msgr.osd_peers.find(primary_osd);
-        if (peer_it != msgr.osd_peers.end())
+        auto peer_it = msgr->osd_peers.find(primary_osd);
+        if (peer_it != msgr->osd_peers.end())
         {
             osd_client_t *cl = peer_it->second;
             part->flags |= PART_SENT|PART_VALID;
@@ -1495,12 +1576,12 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
                 },
             };
             part->op.iov = part->iov;
-            msgr.outbox_push(&part->op);
+            msgr->outbox_push(&part->op);
             return TRY_SEND_OK;
         }
-        else if (msgr.wanted_peers.find(primary_osd) == msgr.wanted_peers.end())
+        else if (msgr->wanted_peers.find(primary_osd) == msgr->wanted_peers.end())
         {
-            msgr.connect_peer(primary_osd, st_cli->peer_states[primary_osd]);
+            msgr->connect_peer(primary_osd, st_cli->peer_states[primary_osd]);
             return TRY_SEND_CONNECTING;
         }
     }
@@ -1509,6 +1590,8 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
 
 int cluster_client_t::continue_sync(cluster_op_t *op)
 {
+    if (aborted)
+        return 0;
     if (op->state == 1)
         goto resume_1;
     if (!dirty_osds.size())
@@ -1522,8 +1605,8 @@ int cluster_client_t::continue_sync(cluster_op_t *op)
     for (auto do_it = dirty_osds.begin(); do_it != dirty_osds.end(); )
     {
         osd_num_t sync_osd = *do_it;
-        auto peer_it = msgr.osd_peers.find(sync_osd);
-        if (peer_it == msgr.osd_peers.end())
+        auto peer_it = msgr->osd_peers.find(sync_osd);
+        if (peer_it == msgr->osd_peers.end())
             dirty_osds.erase(do_it++);
         else
             do_it++;
@@ -1578,7 +1661,7 @@ resume_1:
 
 void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part)
 {
-    osd_client_t *cl = msgr.osd_peers.at(part->osd_num);
+    osd_client_t *cl = msgr->osd_peers.at(part->osd_num);
     part->flags |= PART_SENT;
     op->inflight_count++;
     part->op = (osd_op_t){
@@ -1595,7 +1678,7 @@ void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part)
             handle_op_part(part);
         },
     };
-    msgr.outbox_push(&part->op);
+    msgr->outbox_push(&part->op);
 }
 
 static inline void mem_or(void *res, const void *r2, unsigned int len)
@@ -1650,7 +1733,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
         reset_retry_timer(op->retry_after);
         if (stop_client_id)
         {
-            msgr.stop_client(stop_client_id);
+            msgr->stop_client(stop_client_id);
         }
         op->inflight_count--;
         if (op->inflight_count == 0 && !op->retry_after)

src/client/cluster_client.h

@@ -110,6 +110,7 @@ class __attribute__((visibility("default"))) cluster_client_t
 {
     timerfd_manager_t *tfd = NULL;
     ring_loop_t *ringloop = NULL;
+    std::function<void(int, const std::string&)> on_fatal_error_hook;
 
     // config:
 
@@ -123,6 +124,7 @@ class __attribute__((visibility("default"))) cluster_client_t
     uint64_t client_max_buffered_ops = 0;
     uint64_t client_max_writeback_iodepth = 0;
     std::string conf_hostname;
+    bool exit_on_error = false;
 
     std::string vault_url;
     std::string vault_client_cert;
@@ -178,11 +180,13 @@ class __attribute__((visibility("default"))) cluster_client_t
     std::multimap<osd_num_t, osd_op_t*> raw_ops;
     int continuing_ops = 0;
     bool msgr_initialized = false;
+    int aborted = 0;
+    std::string abort_error;
 
 public:
     std::unique_ptr<etcd_state_client_t> st_cli;
+    osd_messenger_t* msgr = NULL;
 
-    osd_messenger_t msgr;
     void init_msgr();
 
     json11::Json::object cli_config, file_config, etcd_global_config;
@@ -194,8 +198,10 @@ public:
     void execute(cluster_op_t *op);
     void execute_raw(osd_num_t osd_num, osd_op_t *op);
     bool is_ready();
+    int is_failed(std::string & error);
     void on_ready(std::function<void(void)> fn);
     bool flush();
+    void set_on_fatal_error_hook(std::function<void(int, const std::string&)>);
 
     bool get_immediate_commit(uint64_t inode);
 
@@ -204,6 +210,7 @@ public:
 
 protected:
     void continue_ops(int time_passed = 0);
+    void fatal_error(int errcode, const std::string & error);
 
     std::shared_ptr<inode_cache_t> inode_cache_get(inode_t ino);
     void vault_parse_config();

src/client/cluster_client_impl.h

@@ -46,6 +46,7 @@ public:
     int writeback_queue_size = 0;
     int writebacks_active = 0;
     uint64_t last_flush_id = 0;
+    cluster_client_t *parent = NULL;
 
     std::map<object_id, cluster_buffer_t> dirty_buffers;
     std::vector<cluster_op_t*> writeback_overflow;

src/client/cluster_client_list.cpp

@@ -301,14 +301,14 @@ int cluster_client_t::start_pg_listing(inode_list_pg_t *pg)
     bool conn = true;
     for (osd_num_t peer_osd: all_peers)
     {
-        if (msgr.osd_peers.find(peer_osd) == msgr.osd_peers.end())
+        if (msgr->osd_peers.find(peer_osd) == msgr->osd_peers.end())
         {
             // Initiate connection
             if (st_cli->peer_states[peer_osd].is_null())
             {
                 return LIST_PG_WAIT_ACTIVE;
             }
-            msgr.connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
+            msgr->connect_peer(peer_osd, st_cli->peer_states[peer_osd]);
             conn = false;
         }
     }
@@ -346,7 +346,7 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
     osd_op_t *op = new osd_op_t();
     op->op_type = OSD_OP_OUT;
     // Already checked that it exists above, but anyway
-    op->client_id = msgr.osd_peers.at(cur_list->osd_num)->client_id;
+    op->client_id = msgr->osd_peers.at(cur_list->osd_num)->client_id;
     op->req = (osd_any_op_t){
         .sec_list = {
             .header = {
@@ -425,7 +425,7 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
         finish_list_pg(cur_list->pg, true);
         continue_listing(cur_list->pg->lst);
     };
-    msgr.outbox_push(op);
+    msgr->outbox_push(op);
 }
 
 void cluster_client_t::finish_list_pg(inode_list_pg_t *pg, bool retry_epipe)

src/client/cluster_client_real.cpp

@@ -92,14 +92,19 @@ void cluster_client_t::vault_load_keys()
     {
         return;
     }
+    if (vault_url.empty())
+    {
+        fatal_error(EINVAL, "Vault URL is missing");
+        return;
+    }
     if (!vault_http_ctx)
     {
         std::string error;
         vault_http_ctx = http_context_init(tfd, vault_client_cert, vault_client_key, vault_ca, true, error);
         if (!vault_http_ctx)
         {
-            fprintf(stderr, "Failed to initialize HTTP context for Vault: %s\n", error.c_str());
-            exit(1);
+            fatal_error(EINVAL, "Failed to initialize HTTP context for Vault: "+error);
+            return;
         }
         vault_http_cli = http_init(vault_http_ctx);
     }

src/client/cluster_client_wb.cpp

@@ -336,7 +336,7 @@ void writeback_cache_t::flush_buffers(cluster_client_t *cli, dirty_buf_it_t from
     {
         // Buffer flushes are always retried, regardless of the error,
         // so they should never result in an error here
-        assert(op->retval == op->len);
+        assert(op->retval == op->len || parent->aborted);
         for (auto fl_it = flushed_buffers.lower_bound(op->flush_id);
             fl_it != flushed_buffers.end() && fl_it->first == op->flush_id; )
         {

src/client/etcd_state_client.cpp

@@ -10,6 +10,19 @@
 #include "str_util.h"
 #include "json_util.h"
 
+void etcd_state_client_t::fatal_error(int errcode, const std::string & error)
+{
+    if (on_fatal_error_hook)
+    {
+        (std::function<void(int, const std::string&)>(on_fatal_error_hook))(errcode, error);
+    }
+    else
+    {
+        fprintf(stderr, "Error: %s\n", error.c_str());
+        exit(1);
+    }
+}
+
 etcd_state_client_t::~etcd_state_client_t()
 {
     for (auto watch: watches)
@@ -333,7 +346,7 @@ void etcd_state_client_t::load_pgs(std::function<void(const std::string &)> cb)
             return;
         }
         reset_pg_exists();
-        etcd_watch_revision_config = etcd_watch_revision_osd = etcd_watch_revision_pg = data["header"]["revision"].uint64_value()+1;
+        etcd_watch_revision_config = etcd_watch_revision_osd = etcd_watch_revision_pg = data["header"]["revision"].uint64_value();
         if (this->log_level > 3)
         {
             fprintf(stderr, "Loaded revision %ju of PG configuration\n", etcd_watch_revision_pg-1);

src/client/etcd_state_client.h

@@ -144,6 +144,7 @@ protected:
     std::set<osd_num_t> seen_peers;
     bool new_pg_config = false;
 
+    void fatal_error(int errcode, const std::string & error);
     void add_etcd_url(std::string);
     void reset_pg_exists();
     void clean_nonexistent_pgs();
@@ -192,6 +193,7 @@ public:
     std::function<void()> on_reload_hook;
     std::function<void(inode_t, bool)> on_inode_change_hook;
     std::function<void(http_co_t *)> on_start_watcher_hook;
+    std::function<void(int, const std::string&)> on_fatal_error_hook;
 
     json11::Json::object serialize_inode_cfg(inode_config_t *cfg);
     inode_config_t deserialize_inode_cfg(uint64_t inode_num, json11::Json value, uint64_t mod_revision);

src/client/etcd_state_client_http.cpp

@@ -31,6 +31,11 @@ etcd_state_client_http_t::~etcd_state_client_http_t()
         tfd->clear_timer(load_pgs_timer_id);
         load_pgs_timer_id = -1;
     }
+    for (int timer_id: timers)
+    {
+        tfd->clear_timer(timer_id);
+    }
+    timers.clear();
     if (http_ctx)
     {
         http_context_destroy(http_ctx);
@@ -49,7 +54,11 @@ void etcd_state_client_http_t::etcd_add_watch(json11::Json watch)
 
 std::string etcd_state_client_http_t::get_username()
 {
-    return http_context_get_ssl_cn(get_http_ctx());
+    if (!get_http_ctx())
+    {
+        return "";
+    }
+    return http_context_get_ssl_cn(http_ctx);
 }
 
 http_context_t *etcd_state_client_http_t::get_http_ctx()
@@ -60,8 +69,7 @@ http_context_t *etcd_state_client_http_t::get_http_ctx()
         http_ctx = http_context_init(tfd, etcd_client_cert, etcd_client_key, etcd_ca, true, error);
         if (!http_ctx)
         {
-            fprintf(stderr, "Failed to initialize HTTP context: %s\n", error.c_str());
-            exit(1);
+            fatal_error(EINVAL, "Failed to initialize HTTP context: "+error);
         }
     }
     return http_ctx;
@@ -70,7 +78,11 @@ http_context_t *etcd_state_client_http_t::get_http_ctx()
 void etcd_state_client_http_t::etcd_call_oneshot(const std::string & etcd_url, const std::string & api, json11::Json payload,
     int timeout, std::function<void(std::string, json11::Json)> callback)
 {
-    auto http_cli = http_init(get_http_ctx());
+    if (!get_http_ctx())
+    {
+        return;
+    }
+    auto http_cli = http_init(http_ctx);
     http_json_post(http_cli, etcd_url+api, payload, "", { .timeout = timeout }, [http_cli, callback](http_message_t *response)
     {
         std::string err;
@@ -93,7 +105,29 @@ void etcd_state_client_http_t::etcd_call(const std::string & api, json11::Json p
 void etcd_state_client_http_t::etcd_call_selected(const std::string & api, json11::Json payload, int timeout,
     int retries, int interval, std::function<void(std::string, json11::Json)> callback)
 {
+    retries--;
     const auto & url = selected_etcd_url;
+    if (url.addr.empty())
+    {
+        // Resolving failed
+        if (retries > 0)
+        {
+            if (log_level > 0)
+            {
+                fprintf(stderr, "Warning: etcd request failed, retrying %d more times\n", retries);
+            }
+            timers.insert(tfd->set_timer(interval < 500 ? 500 : interval, false, [this, api, payload, timeout, retries, interval, callback](int timer_id)
+            {
+                timers.erase(timer_id);
+                etcd_call(api, payload, timeout, retries, interval, callback);
+            }));
+        }
+        else
+        {
+            callback("None of etcd addresses could be resolved", json11::Json());
+        }
+        return;
+    }
     std::string req = payload.dump();
     req = "POST "+url.path+api+" HTTP/1.1\r\n"
         "Host: "+url.hostname+"\r\n"
@@ -102,7 +136,6 @@ void etcd_state_client_http_t::etcd_call_selected(const std::string & api, json1
         "Connection: keep-alive\r\n"
         "Keep-Alive: timeout="+std::to_string(etcd_keepalive_timeout)+"\r\n"
         "\r\n"+req;
-    retries--;
     auto cb = [this, api, payload, timeout, retries, interval, callback,
         cur_addr = url.addr](http_message_t *response)
     {
@@ -124,11 +157,11 @@ void etcd_state_client_http_t::etcd_call_selected(const std::string & api, json1
                 }
                 if (interval > 0)
                 {
-                    // FIXME: Prevent destruction of etcd_state_client if timers or requests are active
-                    tfd->set_timer(interval, false, [this, api, payload, timeout, retries, interval, callback](int)
+                    timers.insert(tfd->set_timer(interval, false, [this, api, payload, timeout, retries, interval, callback](int timer_id)
                     {
+                        timers.erase(timer_id);
                         etcd_call(api, payload, timeout, retries, interval, callback);
-                    });
+                    }));
                 }
                 else
                     etcd_call(api, payload, timeout, retries, interval, callback);
@@ -140,7 +173,11 @@ void etcd_state_client_http_t::etcd_call_selected(const std::string & api, json1
             callback(err, data);
     };
     if (!keepalive_client)
-        keepalive_client = http_init(get_http_ctx());
+    {
+        if (!get_http_ctx())
+            return;
+        keepalive_client = http_init(http_ctx);
+    }
     http_request(keepalive_client, url.addr, req, { .timeout = timeout, .keepalive = true, .ssl = url.ssl }, cb);
 }
 
@@ -153,14 +190,24 @@ void etcd_state_client_http_t::parse_config(const json11::Json & config)
         stop_ws_keepalive();
         start_ws_keepalive();
     }
+    if (!etcd_addresses.size() && !etcd_local.size())
+    {
+        fatal_error(EINVAL, "etcd_address is missing in Vitastor configuration");
+        return;
+    }
+    get_http_ctx();
 }
 
 void etcd_state_client_http_t::pick_next_etcd(std::function<void()> cb)
 {
     if (!etcd_addresses.size() && !etcd_local.size())
     {
-        fprintf(stderr, "etcd_address is missing in Vitastor configuration\n");
-        exit(1);
+        fatal_error(EINVAL, "etcd_address is missing in Vitastor configuration");
+        return;
+    }
+    if (!get_http_ctx())
+    {
+        return;
     }
     if (selected_etcd_url.addr != "")
     {
@@ -196,7 +243,7 @@ void etcd_state_client_http_t::pick_next_etcd(std::function<void()> cb)
     for (auto & url: etcd_name_urls)
     {
         resolve_count++;
-        http_resolve(get_http_ctx(), url.ssl, url.addr, [this, url](const std::string & error, const std::vector<std::string>& addresses)
+        http_resolve(http_ctx, url.ssl, url.addr, [this, url](const std::string & error, const std::vector<std::string>& addresses)
         {
             if (error != "")
                 fprintf(stderr, "Error resolving %s: %s\n", url.addr.c_str(), error.c_str());
@@ -228,8 +275,13 @@ void etcd_state_client_http_t::pick_next_etcd_on_resolve()
 {
     if (!etcd_urls_to_try.size())
     {
-        fprintf(stderr, "None of etcd_address could be resolved\n");
-        exit(1);
+        selected_etcd_url = {};
+        auto cbs = std::move(on_resolve_queue);
+        for (auto cb: cbs)
+        {
+            cb();
+        }
+        return;
     }
     if (!rand_initialized)
     {
@@ -266,6 +318,16 @@ void etcd_state_client_http_t::start_etcd_watcher_selected()
 {
     const auto & url = selected_etcd_url;
     etcd_watches_initialised = 0;
+    if (url.addr.empty())
+    {
+        // Resolving failed, retry in <etcd_slow_timeout>
+        timers.insert(tfd->set_timer(etcd_slow_timeout, false, [this](int timer_id)
+        {
+            timers.erase(timer_id);
+            start_etcd_watcher();
+        }));
+        return;
+    }
     ws_alive = 1;
     if (this->log_level > 1)
     {
@@ -274,7 +336,11 @@ void etcd_state_client_http_t::start_etcd_watcher_selected()
             etcd_watch_revision_config, etcd_watch_revision_osd, etcd_watch_revision_pg);
     }
     if (!etcd_watch_ws)
-        etcd_watch_ws = http_init(get_http_ctx());
+    {
+        if (!get_http_ctx())
+            return;
+        etcd_watch_ws = http_init(http_ctx);
+    }
     else
         http_close(etcd_watch_ws);
     open_websocket(etcd_watch_ws, url.addr, url.hostname, url.path+"/watch", { .timeout = etcd_slow_timeout, .ssl = url.ssl },
@@ -324,20 +390,17 @@ void etcd_state_client_http_t::start_etcd_watcher_selected()
                                 etcd_watch_revision_config = etcd_watch_revision_osd = etcd_watch_revision_pg = 0;
                                 on_reload_hook();
                             }
-                            return;
                         }
                         else
                         {
-                            fprintf(stderr, "Revisions before %ju were compacted by etcd, exiting\n",
-                                data["result"]["compact_revision"].uint64_value());
-                            exit(1);
+                            fatal_error(EIO, "Revisions before "+data["result"]["compact_revision"].as_string()+" were compacted by etcd, exiting");
                         }
                     }
                     else
                     {
-                        fprintf(stderr, "Watch canceled by etcd, reason: %s, exiting\n", data["result"]["cancel_reason"].string_value().c_str());
-                        exit(1);
+                        fatal_error(EIO, "Watch canceled by etcd, reason: "+data["result"]["cancel_reason"].string_value()+", exiting");
                     }
+                    return;
                 }
                 // Save revision only if it's present in the message - because sometimes etcd sends something without a header, like:
                 // {"error": {"grpc_code": 14, "http_code": 503, "http_status": "Service Unavailable", "message": "error reading from server: EOF"}}
@@ -400,10 +463,11 @@ void etcd_state_client_http_t::start_etcd_watcher_selected()
             if (etcd_watches_initialised == 0)
             {
                 // Connection not established, retry in <etcd_quick_timeout>
-                tfd->set_timer(etcd_quick_timeout, false, [this](int)
+                timers.insert(tfd->set_timer(etcd_quick_timeout, false, [this](int timer_id)
                 {
+                    timers.erase(timer_id);
                     start_etcd_watcher();
-                });
+                }));
             }
             else if (etcd_watches_initialised > 0)
             {
@@ -492,17 +556,18 @@ void etcd_state_client_http_t::load_global_config()
     {
         if (err != "")
         {
-            fprintf(stderr, "Error reading configuration from etcd: %s\n", err.c_str());
             if (infinite_start)
             {
-                tfd->set_timer(etcd_slow_timeout, false, [this](int timer_id)
+                fprintf(stderr, "Failed to read configuration from etcd: %s\n", err.c_str());
+                timers.insert(tfd->set_timer(etcd_slow_timeout, false, [this](int timer_id)
                 {
+                    timers.erase(timer_id);
                     load_global_config();
-                });
+                }));
             }
             else
             {
-                exit(1);
+                fatal_error(EIO, "Failed to read configuration from etcd: "+err);
             }
         }
     });
@@ -533,10 +598,11 @@ void etcd_state_client_http_t::load_pgs()
         {
             // Retry indefinitely
             fprintf(stderr, "Error loading PGs from etcd: %s\n", err.c_str());
-            tfd->set_timer(etcd_slow_timeout, false, [this](int timer_id)
+            timers.insert(tfd->set_timer(etcd_slow_timeout, false, [this](int timer_id)
             {
+                timers.erase(timer_id);
                 load_pgs();
-            });
+            }));
         }
         else
         {

src/client/etcd_state_client_http.h

@@ -22,6 +22,7 @@ protected:
     http_context_t *http_ctx = NULL;
     size_t local_to_try = 0;
     std::vector<http_url_t> etcd_urls_to_try;
+    std::set<int> timers;
     http_url_t selected_etcd_url;
     size_t resolve_count = 0;
     std::vector<std::function<void()>> on_resolve_queue;

src/client/messenger.cpp

@@ -17,6 +17,19 @@
 #include "msgr_rdma.h"
 #endif
 
+void osd_messenger_t::fatal_error(int errcode, const std::string & error)
+{
+    if (on_fatal_error_hook)
+    {
+        (std::function<void(int, const std::string&)>(on_fatal_error_hook))(errcode, error);
+    }
+    else
+    {
+        fprintf(stderr, "Error: %s\n", error.c_str());
+        exit(1);
+    }
+}
+
 void osd_messenger_t::init()
 {
     init_tls();
@@ -201,7 +214,7 @@ osd_messenger_t::~osd_messenger_t()
     destroy_tls();
 }
 
-static int parse_proto_checksums(const json11::Json & val, int default_value)
+int osd_messenger_t::parse_proto_checksums(const json11::Json & val, int default_value)
 {
     if (val.is_string())
     {
@@ -219,9 +232,9 @@ static int parse_proto_checksums(const json11::Json & val, int default_value)
     }
     else if (val.is_null())
         return default_value;
-    fprintf(stderr, "proto_checksums should be \"full\", \"payload\", \"gcm\", \"none\""
-        ", \"\" or null (default), but it is: %s\n", val.dump().c_str());
-    exit(1);
+    fatal_error(EINVAL, "proto_checksums should be \"full\", \"payload\", \"gcm\", \"none\""
+        ", \"\" or null (default), but it is: "+val.dump());
+    return default_value;
 }
 
 void osd_messenger_t::parse_config(const json11::Json & config, bool init)
@@ -813,7 +826,8 @@ void osd_messenger_t::accept_connections(int listen_fd)
     }
     if (peer_fd == -1 && errno != EAGAIN)
     {
-        throw std::runtime_error(std::string("accept: ") + strerror(errno));
+        fprintf(stderr, "Error: accept: %s (code %d), exiting\n", strerror(errno), errno);
+        exit(1);
     }
 }
 

src/client/messenger.h

@@ -259,6 +259,7 @@ protected:
     bool gcm_enabled = false;
     msgr_handshake_ctx_i *hs_ctx = NULL;
 
+    void fatal_error(int errcode, const std::string & error);
     void init_tls();
     void destroy_tls();
     bool init_tls_client(osd_client_t *cl);
@@ -319,6 +320,7 @@ public:
     std::function<void(osd_num_t)> break_pg_locks;
     std::function<bool(osd_client_t*, json11::Json)> check_config_hook;
     std::function<void(osd_client_t*)> handshake_hook;
+    std::function<void(int, const std::string&)> on_fatal_error_hook;
     void read_requests();
     void send_replies();
     void accept_connections(int listen_fd);
@@ -346,7 +348,8 @@ public:
     void measure_exec(osd_op_t *cur_op);
 
 protected:
-    void try_connect_peer(osd_num_t osd_num);
+    int parse_proto_checksums(const json11::Json & val, int default_value);
+    void try_connect_peer(uint64_t osd_num);
     void try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_host, int peer_port);
     void handle_peer_epoll(int peer_fd, int epoll_events);
     void handle_connect_epoll(int peer_fd);

src/client/msgr_encrypt.cpp

@@ -597,10 +597,9 @@ void osd_messenger_t::init_tls()
         if (client_cert.empty() || client_key.empty() || osd_ca.empty() || osd_num && client_ca.empty())
         {
             if (osd_num)
-                fprintf(stderr, "Vitastor transport encryption requires osd_cert, osd_pkey, osd_ca, client_ca options for OSDs\n");
+                fatal_error(EINVAL, "Vitastor transport encryption requires osd_cert, osd_pkey, osd_ca, client_ca options for OSDs");
             else
-                fprintf(stderr, "Vitastor transport encryption requires cert, pkey and osd_ca options\n");
-            exit(1);
+                fatal_error(EINVAL, "Vitastor transport encryption requires cert, pkey and osd_ca options");
         }
         else
         {
@@ -609,8 +608,8 @@ void osd_messenger_t::init_tls()
             hs_ctx = msgr_handshake_ctx_i::create_ctx();
             if (!hs_ctx->init(client_cert, client_key, osd_ca, client_ca, admin_ca))
             {
-                fprintf(stderr, "Error: %s\n", hs_ctx->get_error().c_str());
-                exit(1);
+                gcm_enabled = false;
+                fatal_error(EINVAL, hs_ctx->get_error());
             }
 #endif
         }

src/client/msgr_rdma.cpp

@@ -575,7 +575,7 @@ json11::Json osd_messenger_t::connect_rdma(uint64_t client_id, std::string rdma_
     return json11::Json();
 }
 
-static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
+static int try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
 {
     ibv_send_wr *bad_wr = NULL;
     ibv_send_wr wr = {
@@ -586,12 +586,12 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
         .send_flags = IBV_SEND_SIGNALED,
     };
     int err = ibv_post_send(cl->rdma_conn->qp, &wr, &bad_wr);
-    if (err || bad_wr)
-    {
-        fprintf(stderr, "RDMA send failed: %s (code %d)\n", strerror(err), err);
-        exit(1);
-    }
+    if (err)
+        return err;
+    if (bad_wr)
+        return EIO;
     cl->rdma_conn->cur_send++;
+    return 0;
 }
 
 bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
@@ -605,8 +605,11 @@ bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
         rc->send_out.mr = ibv_reg_mr(rc->ctx->pd, rc->send_out.buf, rc->send_out_size, 0);
         if (!rc->send_out.mr)
         {
-            fprintf(stderr, "Failed to register RDMA memory region: %s\n", strerror(errno));
-            exit(1);
+            rc->send_out_size = 0;
+            free(rc->send_out.buf);
+            rc->send_out.buf = NULL;
+            fatal_error(EIO, std::string("Failed to register RDMA memory region: ")+strerror(errno)+" (code "+std::to_string(errno)+")");
+            return false;
         }
     }
     // Copy data into the buffer and send it
@@ -641,7 +644,12 @@ bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
                 .length = (uint32_t)copied,
                 .lkey = rc->send_out.mr->lkey,
             };
-            try_send_rdma_wr(cl, &sge, 1);
+            int err = try_send_rdma_wr(cl, &sge, 1);
+            if (err)
+            {
+                fatal_error(EIO, std::string("RDMA send failed: ")+strerror(err)+" (code "+std::to_string(err)+")");
+                return false;
+            }
             rc->send_sizes.push_back(copied);
             cl->unref_send_free(send_free_from, cl->next_send_free + cl->send_free.size()); // unref at once - everything is already copied
         }
@@ -649,7 +657,7 @@ bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
     return true;
 }
 
-static void try_recv_rdma_wr(osd_client_t *cl, void *buf)
+static int try_recv_rdma_wr(osd_client_t *cl, void *buf)
 {
     ibv_sge sge = {
         .addr = (uintptr_t)buf,
@@ -663,12 +671,12 @@ static void try_recv_rdma_wr(osd_client_t *cl, void *buf)
         .num_sge = 1,
     };
     int err = ibv_post_recv(cl->rdma_conn->qp, &wr, &bad_wr);
-    if (err || bad_wr)
-    {
-        fprintf(stderr, "RDMA receive failed: %s\n", strerror(err));
-        exit(1);
-    }
+    if (err)
+        return err;
+    if (bad_wr)
+        return EIO;
     cl->rdma_conn->cur_recv++;
+    return 0;
 }
 
 bool osd_messenger_t::init_recv_rdma(osd_client_t *cl)
@@ -679,14 +687,21 @@ bool osd_messenger_t::init_recv_rdma(osd_client_t *cl)
     rc->recv_buf.mr = ibv_reg_mr(rc->ctx->pd, rc->recv_buf.buf, rc->max_msg * rc->max_recv, IBV_ACCESS_LOCAL_WRITE);
     if (!rc->recv_buf.mr)
     {
-        fprintf(stderr, "Failed to register RDMA memory region: %s\n", strerror(errno));
-        exit(1);
+        free(rc->recv_buf.buf);
+        rc->recv_buf.buf = NULL;
+        fatal_error(EIO, std::string("Failed to register RDMA memory region: ")+strerror(errno)+" (code "+std::to_string(errno)+")");
+        return false;
     }
     for (uint32_t i = 0; i < rc->max_recv; i++)
     {
         uint8_t *b = rc->recv_buf.buf + i*rc->max_msg;
         rc->recv_buffers.push_back(b);
-        try_recv_rdma_wr(cl, b);
+        int err = try_recv_rdma_wr(cl, b);
+        if (err)
+        {
+            fatal_error(EIO, std::string("RDMA receive failed: ")+strerror(err)+" (code "+std::to_string(err)+")");
+            return false;
+        }
     }
     return true;
 }
@@ -705,8 +720,8 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context)
     }
     if (ibv_req_notify_cq(rdma_context->cq, 0) != 0)
     {
-        fprintf(stderr, "Failed to request RDMA completion notification, exiting\n");
-        exit(1);
+        fatal_error(EIO, "Failed to request RDMA completion notification");
+        return;
     }
     ibv_wc wc[RDMA_EVENTS_AT_ONCE];
     int event_count;

src/client/msgr_rdmacm.cpp

@@ -92,8 +92,8 @@ void osd_messenger_t::handle_rdmacm_events()
         {
             if (errno == EAGAIN || errno == EINTR)
                 break;
-            fprintf(stderr, "Failed to get RDMA-CM event: %s (code %d)\n", strerror(errno), errno);
-            exit(1);
+            fatal_error(EIO, std::string("Failed to get RDMA-CM event: ")+strerror(errno)+" (code "+std::to_string(errno)+")");
+            return;
         }
         // ...so we save a copy of all events EXCEPT connection requests, otherwise they sometimes fail with EVENT_DISCONNECT
         if (ev->event == RDMA_CM_EVENT_CONNECT_REQUEST)
@@ -107,8 +107,8 @@ void osd_messenger_t::handle_rdmacm_events()
         r = rdma_ack_cm_event(ev);
         if (r != 0)
         {
-            fprintf(stderr, "Failed to ack (free) RDMA-CM event: %s (code %d)\n", strerror(errno), errno);
-            exit(1);
+            fatal_error(EIO, std::string("Failed to ack (free) RDMA-CM event: ")+strerror(errno)+" (code "+std::to_string(errno)+")");
+            return;
         }
     }
     for (auto & evl: events_copy)

src/client/qemu_driver.c

@@ -627,7 +627,8 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
     VitastorRPC task;
     VitastorClient *client = bs->opaque;
     void *image = NULL;
-    int64_t ret = 0;
+    char *errstr = NULL;
+    int ret = 0;
     qemu_mutex_init(&client->mutex);
     client->config_path = g_strdup(qdict_get_try_str(options, "config-path"));
     // FIXME: Rename to etcd_address
@@ -658,6 +659,7 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
     strarray_push_kv(&opt, "rdma_gid_index", qdict_get_try_str(options, "rdma-gid-index"));
     strarray_push_kv(&opt, "rdma_mtu", qdict_get_try_str(options, "rdma-mtu"));
     strarray_push_kv(&opt, "client_writeback_allowed", (flags & BDRV_O_NOCACHE) ? "0" : "1");
+    strarray_push_kv(&opt, "client_exit_on_error", "0");
     strarray_push_kv(&opt, "__version_check_buffer", version_buffer);
     client->proxy = vitastor_c_create_uring_json(opt.items, opt.len);
     client->auto_loop = 0;
@@ -670,7 +672,7 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
             fprintf(stderr, "vitastor: failed to create io_uring eventfd: %s\n", strerror(errno));
             error_setg(errp, "failed to create io_uring eventfd");
             vitastor_close(bs);
-            return -1;
+            return -EOPNOTSUPP;
         }
         universal_aio_set_fd_handler(client->ctx, client->uring_eventfd, vitastor_uring_handler, NULL, client);
         int vitastor_version[3] = { 0 };
@@ -714,6 +716,13 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
 #endif
         BDRV_POLL_WHILE(bs, !task.complete);
     }
+    if ((ret = vitastor_c_is_failed(client->proxy, &errstr)))
+    {
+        error_setg(errp, errstr);
+        free(errstr);
+        vitastor_close(bs);
+        return -ret;
+    }
     client->image = image;
     if (client->image)
     {
@@ -724,7 +733,7 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
         {
             error_setg(errp, "image does not exist");
             vitastor_close(bs);
-            return -1;
+            return -ENOENT;
         }
         if (!client->size)
         {
@@ -747,7 +756,7 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
     {
         error_setg(errp, "image size not specified");
         vitastor_close(bs);
-        return -1;
+        return -EINVAL;
     }
     bs->total_sectors = client->size / BDRV_SECTOR_SIZE;
 #if QEMU_VERSION_MAJOR > 5 || QEMU_VERSION_MAJOR == 5 && QEMU_VERSION_MINOR >= 1
@@ -768,7 +777,7 @@ static int vitastor_file_open(BlockDriverState *bs, QDict *options, int flags, E
     qdict_del(options, "pool");
     qdict_del(options, "size");
     qdict_del(options, "skip-parents");
-    return ret;
+    return 0;
 }
 
 static void vitastor_close(BlockDriverState *bs)
@@ -815,6 +824,10 @@ static void vitastor_close(BlockDriverState *bs)
         g_free(client->etcd_host);
     if (client->etcd_prefix)
         g_free(client->etcd_prefix);
+    if (client->file_mirror_path)
+        g_free(client->file_mirror_path);
+    if (client->rdma_device)
+        g_free(client->rdma_device);
     if (client->image)
         g_free(client->image);
     free(client->last_bitmap);

src/client/vitastor.pc.in

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

src/client/vitastor_c.cpp

@@ -30,6 +30,18 @@ struct vitastor_c
 
     QEMUSetFDHandler *aio_set_fd_handler = NULL;
     void *aio_ctx = NULL;
+
+    ~vitastor_c()
+    {
+        if (cli)
+            delete cli;
+        if (epmgr)
+            delete epmgr;
+        else if (tfd)
+            delete tfd;
+        if (ringloop)
+            delete ringloop;
+    }
 };
 
 extern "C" {
@@ -217,13 +229,6 @@ void* vitastor_c_get_internal_client(vitastor_c *client)
 
 void vitastor_c_destroy(vitastor_c *client)
 {
-    delete client->cli;
-    if (client->epmgr)
-        delete client->epmgr;
-    else if (client->tfd)
-        delete client->tfd;
-    if (client->ringloop)
-        delete client->ringloop;
     delete client;
 }
 
@@ -232,6 +237,20 @@ int vitastor_c_is_ready(vitastor_c *client)
     return client->cli->is_ready();
 }
 
+int vitastor_c_is_failed(vitastor_c *client, char **errstr)
+{
+    std::string err;
+    int errcode = client->cli->is_failed(err);
+    if (errcode)
+    {
+        *errstr = (char*)malloc(err.size()+1);
+        if (*errstr)
+            memcpy(*errstr, err.c_str(), err.size()+1);
+        return errcode;
+    }
+    return 0;
+}
+
 void vitastor_c_on_ready(vitastor_c *client, VitastorIOHandler cb, void *opaque)
 {
     client->cli->on_ready([=]()

src/client/vitastor_c.h

@@ -53,6 +53,7 @@ vitastor_c *vitastor_c_create_epoll_json(const char **options, int options_len);
 void* vitastor_c_get_internal_client(vitastor_c *client);
 void vitastor_c_destroy(vitastor_c *client);
 int vitastor_c_is_ready(vitastor_c *client);
+int vitastor_c_is_failed(vitastor_c *client, char **errstr);
 void vitastor_c_on_ready(vitastor_c *client, VitastorIOHandler cb, void *opaque);
 int vitastor_c_uring_register_eventfd(vitastor_c *client);
 void vitastor_c_uring_wait_ready(vitastor_c *client);

src/cmd/cli.h

@@ -56,7 +56,7 @@ public:
     json11::Json parse_tags(std::string tags);
 
     json11::Json::object format_image(const inode_config_t & cfg);
-    void change_parent(inode_t cur, inode_t new_parent, cli_result_t *result);
+    void change_parent(inode_t cur, inode_t old_parent, inode_t new_parent, cli_result_t *result);
     inode_config_t* get_inode_cfg(const std::string & name);
 
     bool check_image_perm(const inode_config_t & cfg, bool write);

src/cmd/cli_common.cpp

@@ -54,6 +54,10 @@ json11::Json::object cli_tool_t::format_image(const inode_config_t & cfg)
         {
             img["parent_name"] = parent_it->second.name;
         }
+        else
+        {
+            img["parent_name"] = "Pool:"+std::to_string(INODE_POOL(cfg.parent_id))+",ID:"+std::to_string(INODE_NO_POOL(cfg.parent_id));
+        }
         img["parent_inode_id"] = cfg.parent_id;
         img["parent_inode_num"] = INODE_NO_POOL(cfg.parent_id);
         img["parent_pool_id"] = (uint64_t)INODE_POOL(cfg.parent_id);
@@ -61,7 +65,7 @@ json11::Json::object cli_tool_t::format_image(const inode_config_t & cfg)
     return img;
 }
 
-void cli_tool_t::change_parent(inode_t cur, inode_t new_parent, cli_result_t *result)
+void cli_tool_t::change_parent(inode_t cur, inode_t old_parent, inode_t new_parent, cli_result_t *result)
 {
     auto cur_cfg_it = cli->st_cli->inode_config.find(cur);
     if (cur_cfg_it == cli->st_cli->inode_config.end())
@@ -79,15 +83,29 @@ void cli_tool_t::change_parent(inode_t cur, inode_t new_parent, cli_result_t *re
     new_cfg.parent_id = new_parent;
     json11::Json::object cur_cfg_json = cli->st_cli->serialize_inode_cfg(&new_cfg);
     waiting++;
+    json11::Json::array cmp = json11::Json::array {
+        json11::Json::object {
+            { "target", "MOD" },
+            { "key", cur_cfg_key },
+            { "result", "LESS" },
+            { "mod_revision", new_cfg.mod_revision+1 },
+        },
+    };
+    if (old_parent)
+    {
+        std::string parent_cfg_key = base64_encode(cli->st_cli->etcd_prefix+
+            "/config/inode/"+std::to_string(INODE_POOL(old_parent))+
+            "/"+std::to_string(INODE_NO_POOL(old_parent)));
+        auto & parent_cfg = cli->st_cli->inode_config.at(old_parent);
+        cmp.push_back(json11::Json::object {
+            { "target", "MOD" },
+            { "key", parent_cfg_key },
+            { "result", "LESS" },
+            { "mod_revision", parent_cfg.mod_revision+1 },
+        });
+    }
     cli->st_cli->etcd_txn_slow(json11::Json::object {
-        { "compare", json11::Json::array {
-            json11::Json::object {
-                { "target", "MOD" },
-                { "key", cur_cfg_key },
-                { "result", "LESS" },
-                { "mod_revision", new_cfg.mod_revision+1 },
-            },
-        } },
+        { "compare", cmp },
         { "success", json11::Json::array {
             json11::Json::object {
                 { "request_put", json11::Json::object {

src/cmd/cli_create.cpp

@@ -40,6 +40,7 @@ struct image_creator_t
     inode_t new_parent_id = 0;
     inode_t new_id = 0, old_id = 0;
     uint64_t max_id_mod_rev = 0, idx_mod_rev = 0;
+    uint64_t new_parent_mod_rev = 0;
     inode_config_t cur_cfg;
     inode_config_t new_cfg;
 
@@ -154,6 +155,7 @@ struct image_creator_t
             if (ic.second.name == new_parent)
             {
                 new_parent_id = ic.second.num;
+                new_parent_mod_rev = ic.second.mod_revision;
                 if (!new_pool_id)
                 {
                     new_pool_id = INODE_POOL(ic.second.num);
@@ -162,6 +164,13 @@ struct image_creator_t
                 {
                     size = ic.second.size;
                 }
+                if (ic.second.deleted && !force)
+                {
+                    result = (cli_result_t){ .err = ENOENT, .text = "Layer "+ic.second.name+
+                        " is deleted and doesn't accept new children" };
+                    state = 100;
+                    return;
+                }
             }
         }
         if (new_parent != "" && !new_parent_id)
@@ -226,12 +235,24 @@ resume_3:
                 state = 100;
                 return;
             }
-            if (!parent->etcd_result["succeeded"].bool_value() &&
-                parent->etcd_result["responses"][0]["response_range"]["kvs"].array_items().size() > 0)
+            if (!parent->etcd_result["succeeded"].bool_value())
             {
-                result = (cli_result_t){ .err = EEXIST, .text = "Image "+image_name+" already exists" };
-                state = 100;
-                return;
+                if (parent->etcd_result["responses"][0]["response_range"]["kvs"].array_items().size() > 0)
+                {
+                    result = (cli_result_t){ .err = EEXIST, .text = "Image "+image_name+" already exists" };
+                    state = 100;
+                    return;
+                }
+                if (new_parent_id)
+                {
+                    auto kv = parent->cli->st_cli->parse_etcd_kv(parent->etcd_result["responses"][1]["response_range"]["kvs"][0]);
+                    if (kv.value.is_null() || kv.mod_revision > new_parent_mod_rev)
+                    {
+                        result = (cli_result_t){ .err = EAGAIN, .text = "Parent image "+kv.value["name"].string_value()+" modified during creation" };
+                        state = 100;
+                        return;
+                    }
+                }
             }
         } while (!parent->etcd_result["succeeded"].bool_value());
         // Save into inode_config for library users to be able to take it from there immediately
@@ -478,6 +499,13 @@ resume_3:
                 size = cur_cfg.size;
             }
         }
+        if (cur_cfg.deleted && !force)
+        {
+            result = (cli_result_t){ .err = ENOENT, .text = "Layer "+cur_cfg.name+
+                " is deleted and doesn't accept new children" };
+            state = 100;
+            return;
+        }
     }
 
     void attempt_create()
@@ -613,7 +641,24 @@ resume_3:
                     }).dump()) },
                 } },
             });
-        };
+        }
+        else if (new_parent_id)
+        {
+            auto new_parent_key = base64_encode(
+                parent->cli->st_cli->etcd_prefix+"/config/inode/"+
+                std::to_string(INODE_POOL(new_parent_id))+"/"+std::to_string(INODE_NO_POOL(new_parent_id))
+            );
+            checks.push_back(json11::Json::object {
+                { "target", "MOD" },
+                { "mod_revision", new_parent_mod_rev },
+                { "key", new_parent_key },
+            });
+            failure.push_back(json11::Json::object {
+                { "request_range", json11::Json::object {
+                    { "key", new_parent_key },
+                } },
+            });
+        }
         parent->etcd_txn(json11::Json::object {
             { "compare", checks },
             { "success", success },

src/cmd/cli_flatten.cpp

@@ -21,6 +21,7 @@ struct snap_flattener_t
 
     std::string top_parent_name;
     inode_t target_id = 0;
+    inode_config_t old_cfg, old_parent_cfg;
     int state = 0;
     std::function<bool(cli_result_t &)> merger_cb;
     cli_result_t result;
@@ -41,6 +42,7 @@ struct snap_flattener_t
             state = 100;
             return;
         }
+        old_cfg = *target_cfg;
         target_id = target_cfg->num;
         std::vector<inode_t> chain_list;
         inode_config_t *cur = target_cfg;
@@ -63,6 +65,10 @@ struct snap_flattener_t
                 state = 100;
                 return;
             }
+            if (cur->num == target_cfg->num)
+            {
+                old_parent_cfg = it->second;
+            }
             cur = &it->second;
             chain_list.push_back(cur->num);
         }
@@ -121,7 +127,24 @@ resume_1:
             return;
         }
         // Change parent
-        parent->change_parent(target_id, 0, &result);
+        {
+            inode_config_t *target_cfg = parent->get_inode_cfg(target_name);
+            if (!target_cfg || target_cfg->mod_revision != old_cfg.mod_revision)
+            {
+                result = (cli_result_t){ .err = EAGAIN, .text = "Layer "+target_name+" modified during deletion" };
+                state = 100;
+                return;
+            }
+            auto parent_cfg_it = parent->cli->st_cli->inode_config.find(target_cfg->parent_id);
+            if (parent_cfg_it == parent->cli->st_cli->inode_config.end() ||
+                parent_cfg_it->second.mod_revision != old_parent_cfg.mod_revision)
+            {
+                result = (cli_result_t){ .err = EAGAIN, .text = "Layer "+old_parent_cfg.name+" modified during deletion" };
+                state = 100;
+                return;
+            }
+        }
+        parent->change_parent(target_id, old_cfg.parent_id, 0, &result);
         // Wait for it to complete
         state = 2;
 resume_2:

src/cmd/cli_rm.cpp

@@ -59,19 +59,23 @@ struct snap_remover_t
 
     std::map<inode_t,int> sources;
     std::map<inode_t,uint64_t> inode_used;
-    std::vector<inode_t> merge_children;
+    std::map<inode_t,uint64_t> merge_children; // inode => mod_revision
     std::vector<inode_t> chain_list;
     std::map<inode_t,int> inverse_candidates;
     inode_t inverse_parent = 0, inverse_child = 0;
     inode_t new_parent = 0;
     int state = 0;
     int current_child = 0;
+    std::map<inode_t,uint64_t>::iterator child_it;
     std::function<bool(cli_result_t &)> cb;
 
     std::vector<std::string> rebased_images, deleted_images;
     std::vector<uint64_t> deleted_ids;
     std::string inverse_child_name, inverse_parent_name;
     inode_config_t inverse_child_new_cfg, inverse_parent_new_cfg;
+    inode_config_t merge_child_cfg;
+    uint64_t merge_parent_rev = 0;
+    uint64_t mark_del_rev = 0;
     cli_result_t result;
 
     bool is_done()
@@ -101,6 +105,8 @@ struct snap_remover_t
             goto resume_9;
         else if (state == 10)
             goto resume_10;
+        else if (state == 11)
+            goto resume_11;
         else if (state == 100)
             goto resume_100;
         assert(!state);
@@ -114,10 +120,23 @@ struct snap_remover_t
         {
             to_name = from_name;
         }
-        // Get children to merge
-        get_merge_children();
+        // Find all images to delete
+        get_rm_ranks();
         if (state == 100)
             return;
+        // Mark images as deleted to prevent rebasing newer images on them
+        mark_deleted();
+        state = 8;
+resume_8:
+        if (parent->waiting > 0)
+            return;
+        // Wait until we get all changes up to mark_del_rev from the watcher
+        state = 11;
+resume_11:
+        if (parent->cli->st_cli->etcd_watch_revision_config < mark_del_rev)
+            return;
+        // Get children to merge - from this point, images shouldn't be rebased to deleted layers
+        get_merge_children();
         // Try to select an inode for the "inverse" optimized scenario
         // Read statistics from etcd to do it
         read_stats();
@@ -129,16 +148,37 @@ resume_1:
             return;
         choose_inverse_candidate();
         // Merge children one by one, except our "inverse" child
-        for (current_child = 0; current_child < merge_children.size(); current_child++)
+        for (child_it = merge_children.begin(); child_it != merge_children.end(); child_it++)
         {
-            if (merge_children[current_child] == inverse_child)
+            if (child_it->first == inverse_child)
                 continue;
-            rebased_images.push_back(parent->cli->st_cli->inode_config.at(merge_children[current_child]).name);
-            start_merge_child(merge_children[current_child], merge_children[current_child]);
+            if (!recheck_cfg(child_it->first, child_it->second, &merge_child_cfg, true))
+                return;
+            if (!merge_child_cfg.num || !merge_child_cfg.parent_id)
+                continue;
+            merge_parent_rev = parent->cli->st_cli->inode_config.at(merge_child_cfg.parent_id).mod_revision;
+            rebased_images.push_back(merge_child_cfg.name);
+            start_merge_child(child_it->first, child_it->first);
 resume_2:
             while (!wait_result(2))
                 return;
-            parent->change_parent(merge_children[current_child], new_parent, &result);
+            if (!recheck_cfg(child_it->first, child_it->second, NULL, false))
+                return;
+            if (!recheck_cfg(merge_child_cfg.parent_id, merge_parent_rev, NULL, false))
+                return;
+            if (new_parent)
+            {
+                auto np_it = parent->cli->st_cli->inode_config.find(new_parent);
+                if (np_it == parent->cli->st_cli->inode_config.end() || np_it->second.deleted)
+                {
+                    // New parent is marked as deleted => failure
+                    result = (cli_result_t){ .err = ENOENT, .text = "Layer "+parent->cli->st_cli->inode_config
+                        .at(new_parent).name+" is deleted and doesn't accept new children" };
+                    state = 100;
+                    return;
+                }
+            }
+            parent->change_parent(child_it->first, merge_child_cfg.parent_id, new_parent, &result);
             state = 3;
 resume_3:
             if (parent->waiting > 0)
@@ -193,11 +233,6 @@ resume_7:
                     deleted_images.push_back(parent_it->second.name);
                 deleted_ids.push_back(chain_list[current_child]);
             }
-            // Mark child as deleted
-            start_mark_deleted(chain_list[current_child]);
-resume_8:
-            while (!wait_result(8))
-                return;
             start_delete_source(chain_list[current_child]);
 resume_9:
             while (!wait_result(9))
@@ -221,6 +256,33 @@ resume_100:
         return;
     }
 
+    bool recheck_cfg(inode_t ino, uint64_t mod_rev, inode_config_t *copy, bool allow_enoent)
+    {
+        auto cfg_it = parent->cli->st_cli->inode_config.find(ino);
+        if (cfg_it == parent->cli->st_cli->inode_config.end())
+        {
+            if (!allow_enoent)
+            {
+                result = (cli_result_t){ .err = ENOENT, .text = "Inode "+std::to_string(INODE_NO_POOL(ino))+
+                    " from pool "+std::to_string(INODE_POOL(ino))+" disappeared during deletion" };
+                state = 100;
+                return false;
+            }
+            if (copy)
+                *copy = {};
+            return true;
+        }
+        if (cfg_it->second.mod_revision != mod_rev)
+        {
+            result = (cli_result_t){ .err = EAGAIN, .text = "Layer "+cfg_it->second.name+" modified during deletion" };
+            state = 100;
+            return false;
+        }
+        if (copy)
+            *copy = cfg_it->second;
+        return true;
+    }
+
     bool wait_result(int base_state)
     {
         if (state == 100 || !cb)
@@ -253,9 +315,8 @@ resume_100:
         return obj;
     }
 
-    void get_merge_children()
+    void get_rm_ranks()
     {
-        // Get all children of from..to
         inode_config_t *from_cfg = parent->get_inode_cfg(from_name);
         if (!from_cfg)
         {
@@ -307,6 +368,11 @@ resume_100:
         {
             sources[item] = i--;
         }
+    }
+
+    void get_merge_children()
+    {
+        // Get all children of from..to
         for (auto & ic: parent->cli->st_cli->inode_config)
         {
             if (!ic.second.parent_id)
@@ -322,7 +388,7 @@ resume_100:
                     state = 100;
                     return;
                 }
-                merge_children.push_back(ic.second.num);
+                merge_children[ic.second.num] = ic.second.mod_revision;
                 if (ic.second.readonly || writers_stopped)
                 {
                     inverse_candidates[ic.second.num] = it->second;
@@ -683,24 +749,55 @@ resume_100:
         });
     }
 
-    void start_mark_deleted(inode_t inode)
+    void mark_deleted()
     {
-        auto ino_it = parent->cli->st_cli->inode_config.find(inode);
-        if (ino_it == parent->cli->st_cli->inode_config.end())
+        json11::Json::array cmp, ok;
+        for (current_child = 0; current_child < chain_list.size(); current_child++)
         {
-            char buf[1024];
-            snprintf(buf, 1024, "Inode 0x%jx disappeared", inode);
-            result = (cli_result_t){ .err = EIO, .text = std::string(buf) };
-            state = 100;
-            return;
+            inode_t ino = chain_list[current_child];
+            auto & inode_cfg = parent->cli->st_cli->inode_config.at(ino);
+            std::string inode_cfg_key = base64_encode(
+                parent->cli->st_cli->etcd_prefix+
+                "/config/inode/"+std::to_string(INODE_POOL(ino))+
+                "/"+std::to_string(INODE_NO_POOL(ino))
+            );
+            inode_cfg.deleted = true;
+            cmp.push_back(json11::Json::object {
+                { "target", "MOD" },
+                { "key", inode_cfg_key },
+                { "result", "LESS" },
+                { "mod_revision", inode_cfg.mod_revision+1 },
+            });
+            ok.push_back(json11::Json::object {
+                { "request_put", json11::Json::object {
+                    { "key", inode_cfg_key },
+                    { "value", base64_encode(json11::Json(parent->cli->st_cli->serialize_inode_cfg(&inode_cfg)).dump()) },
+                } },
+            });
         }
-        if (ino_it->second.deleted)
+        parent->waiting++;
+        parent->cli->st_cli->etcd_txn_slow(json11::Json::object {
+            { "compare", cmp },
+            { "success", ok },
+        }, [this](std::string err, json11::Json res)
         {
-            return;
-        }
-        cb = parent->start_modify(json11::Json::object {
-            { "image", ino_it->second.name },
-            { "deleted", true },
+            parent->waiting--;
+            if (err != "")
+            {
+                result = (cli_result_t){ .err = EIO, .text = "Error marking layers as deleted: "+err };
+                state = 100;
+                return;
+            }
+            if (!res["succeeded"].bool_value())
+            {
+                result = (cli_result_t){ .err = EAGAIN, .text = "Some layers between "+from_name+" and "+to_name+" were modified during deletion" };
+                state = 100;
+                return;
+            }
+            mark_del_rev = res["header"]["revision"].uint64_value();
+            if (parent->progress)
+                printf("Layers between %s and %s marked as deleted\n", from_name.c_str(), to_name.c_str());
+            parent->ringloop->wakeup();
         });
     }
 

src/osd/osd.h

@@ -368,6 +368,8 @@ class osd_t
 
     // primary ops
     void autosync();
+    bool start_pause_sync(pg_t & pg);
+    void submit_pause_sync(pg_t & pg);
     bool prepare_primary_rw(osd_op_t *cur_op);
     void continue_primary_read(osd_op_t *cur_op);
     void continue_primary_scrub(osd_op_t *cur_op);

src/osd/osd_cluster.cpp

@@ -903,11 +903,20 @@ void osd_t::apply_pg_config()
             }
             if (currently_taken && !take)
             {
-                // Stop this PG
-                stop_pg(pg_it->second);
+                // Stop this PG. The monitor pauses PGs of a removed pool before forgetting them
+                bool removed = pg_cfg.pause && !pool_cfg.exists;
+                if (!removed && !pg_it->second.pause_sync || !start_pause_sync(pg_it->second))
+                {
+                    stop_pg(pg_it->second);
+                }
             }
             else if (take)
             {
+                if (currently_taken)
+                {
+                    // Unpaused before the stop
+                    pg_it->second.pause_sync = 0;
+                }
                 // Take this PG
                 std::set<osd_num_t> all_peers;
                 for (osd_num_t pg_osd: pg_cfg.target_set)

src/osd/osd_flush.cpp

@@ -251,7 +251,7 @@ bool osd_t::pick_next_recovery(osd_recovery_op_t &op)
             for (auto pg_it = pgs.lower_bound(recovery_last_pg); pg_it != pgs.end(); pg_it++)
             {
                 auto & src = recovery_last_degraded ? pg_it->second.degraded_objects : pg_it->second.misplaced_objects;
-                if ((pg_it->second.state & mask) == check && src.size() > 0)
+                if ((pg_it->second.state & mask) == check && !pg_it->second.pause_sync && src.size() > 0)
                 {
                     auto pool_it = st_cli->pool_config.find(pg_it->first.pool_id);
                     if (pool_it != st_cli->pool_config.end() && pool_it->second.backfillfull)

src/osd/osd_peering.cpp

@@ -682,6 +682,7 @@ void osd_t::release_peering_state(pg_t & pg)
 
 bool osd_t::stop_pg(pg_t & pg)
 {
+    pg.pause_sync = 0;
     drop_dirty_pg_connections({ .pool_id = pg.pool_id, .pg_num = pg.pg_num });
     release_peering_state(pg);
     if (pg.state & (PG_STOPPING | PG_OFFLINE))
@@ -794,6 +795,10 @@ void osd_t::continue_pg(pg_t & pg)
     {
         finish_stop_pg(pg);
     }
+    else if (pg.pause_sync == PG_PAUSE_SYNC_WAIT && pg.inflight == 0)
+    {
+        submit_pause_sync(pg);
+    }
     else if ((pg.state & PG_REPEERING) && pg.can_repeer())
     {
         start_pg_peering(pg);

src/osd/osd_peering_pg.h

@@ -14,6 +14,10 @@
 
 #define PG_EPOCH_BITS 48
 
+// pg_t::pause_sync
+#define PG_PAUSE_SYNC_WAIT 1
+#define PG_PAUSE_SYNC_RUNNING 2
+
 struct pg_obj_loc_t
 {
     uint64_t role;
@@ -129,6 +133,8 @@ struct pg_t
     pg_flush_batch_t *flush_batch = NULL;
 
     int inflight = 0; // including write_queue
+    // Pool removed: stabilize unstable writes before stopping, nobody peers a removed pool
+    int pause_sync = 0;
     std::multimap<object_id, osd_op_t*> write_queue;
 
     pg_osd_set_state_t* add_object_to_state(const object_id oid, const uint64_t state, const pg_osd_set_t & osd_set);

src/osd/osd_primary.cpp

@@ -57,9 +57,10 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
             return false;
         }
     }
-    else if (!(pg_it->second.state & PG_ACTIVE))
+    else if (!(pg_it->second.state & PG_ACTIVE) ||
+        pg_it->second.pause_sync && cur_op->req.hdr.opcode != OSD_OP_READ)
     {
-        // This OSD is not primary for this PG or the PG is inactive
+        // This OSD is not primary for this PG or the PG is inactive (or about to stop)
         // FIXME: Allow reads from PGs degraded under pg_minsize, but don't allow writes
         finish_op(cur_op, -EPIPE);
         return false;

src/osd/osd_primary_subops.cpp

@@ -41,6 +41,69 @@ void osd_t::autosync()
     }
 }
 
+// stop_pg() forgets unstable writes and leaves them to the peering of the next primary,
+// but nobody peers a removed pool. So a PG of a removed pool stabilizes them first,
+// while it's still active and holds its locks. Returns false if it may be stopped right away
+bool osd_t::start_pause_sync(pg_t & pg)
+{
+    if (pg.pause_sync)
+    {
+        return true;
+    }
+    if (immediate_commit == IMMEDIATE_ALL || !(pg.state & PG_ACTIVE) ||
+        !pg.inflight && dirty_pgs.find({ .pool_id = pg.pool_id, .pg_num = pg.pg_num }) == dirty_pgs.end())
+    {
+        return false;
+    }
+    printf("[PG %u/%u] Pool removed, stabilizing unstable writes before stopping\n", pg.pool_id, pg.pg_num);
+    // New writes are refused, ones in flight are waited for in continue_pg()
+    pg.pause_sync = PG_PAUSE_SYNC_WAIT;
+    continue_pg(pg);
+    return true;
+}
+
+void osd_t::submit_pause_sync(pg_t & pg)
+{
+    pool_pg_num_t pg_id = { .pool_id = pg.pool_id, .pg_num = pg.pg_num };
+    if (!(pg.state & PG_ACTIVE) || dirty_pgs.find(pg_id) == dirty_pgs.end())
+    {
+        pg.pause_sync = 0;
+        stop_pg(pg);
+        return;
+    }
+    pg.pause_sync = PG_PAUSE_SYNC_RUNNING;
+    osd_op_t *op = new osd_op_t();
+    pending_owned_ops.insert(op);
+    op->op_type = OSD_OP_IN;
+    op->client_id = SELF_CLIENT;
+    op->req = (osd_any_op_t){
+        .sync = {
+            .header = {
+                .magic = SECONDARY_OSD_OP_MAGIC,
+                .id = 1,
+                .opcode = OSD_OP_SYNC,
+            },
+        },
+    };
+    op->callback = [this, pg_id](osd_op_t *op)
+    {
+        if (op->reply.hdr.retval < 0)
+        {
+            printf("[PG %u/%u] Failed to stabilize unstable writes before stopping: %jd (%s)\n",
+                pg_id.pool_id, pg_id.pg_num, -op->reply.hdr.retval, strerror(-op->reply.hdr.retval));
+        }
+        pending_owned_ops.erase(op);
+        delete op;
+        auto pg_it = pgs.find(pg_id);
+        if (!in_destruction && pg_it != pgs.end() && pg_it->second.pause_sync == PG_PAUSE_SYNC_RUNNING)
+        {
+            pg_it->second.pause_sync = 0;
+            stop_pg(pg_it->second);
+        }
+    };
+    exec_op(op);
+}
+
 void osd_t::finish_op(osd_op_t *cur_op, int retval)
 {
     inflight_ops--;

src/osd/osd_scrub.cpp

@@ -129,7 +129,8 @@ int osd_t::pick_next_scrub(object_id & next_oid)
     }
     while (pg_it != pgs.end())
     {
-        if ((pg_it->second.state & PG_ACTIVE) && pg_it->second.next_scrub && pg_it->second.next_scrub <= tv_now.tv_sec)
+        if ((pg_it->second.state & PG_ACTIVE) && !pg_it->second.pause_sync &&
+            pg_it->second.next_scrub && pg_it->second.next_scrub <= tv_now.tv_sec)
         {
             // Continue scrubbing from the next object
             if (scrub_last_pg == pg_it->first)

src/test/mock/messenger.cpp

@@ -8,6 +8,12 @@
 
 #include "messenger.h"
 
+void osd_messenger_t::fatal_error(int retval, const std::string & error)
+{
+    fprintf(stderr, "%s\n", error.c_str());
+    exit(1);
+}
+
 void osd_messenger_t::init()
 {
 }

src/test/osd_test.cpp

@@ -3719,6 +3719,197 @@ void test_replicated_add_object_to_set_degraded()
     printf("test_replicated_add_object_to_set_degraded passed\n");
 }
 
+static void set_pause_ec_pool(osd_test_fixture_t & f, bool exists)
+{
+    json11::Json::object pools;
+    if (exists)
+    {
+        pools["1"] = json11::Json::object {
+            { "name", "pool_1" },
+            { "scheme", "ec" },
+            { "pg_size", 3 },
+            { "pg_minsize", 2 },
+            { "parity_chunks", 1 },
+            { "pg_count", 1 },
+            { "failure_domain", "osd" },
+            { "immediate_commit", "none" },
+        };
+    }
+    f.st_cli->set("/vitastor/config/pools", pools);
+}
+
+// EC 2+1 pool with immediate_commit=none, one PG on OSDs 1-3 with OSD 1 primary
+static void set_pause_ec_pg_config(osd_test_fixture_t & f, bool pause)
+{
+    json11::Json::object pg_cfg = {
+        { "osd_set", json11::Json::array{ 1, 2, 3 } },
+        { "primary", 1 },
+    };
+    if (pause)
+        pg_cfg["pause"] = true;
+    f.st_cli->set("/vitastor/pg/config", json11::Json::object {
+        { "items", json11::Json::object{ { "1", json11::Json::object { { "1", pg_cfg } } } } },
+    });
+}
+
+static void start_pause_ec_pg(osd_test_fixture_t & f)
+{
+    set_pause_ec_pool(f, true);
+    set_pause_ec_pg_config(f, false);
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "none" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+    });
+    f.connect_peer(2);
+    f.connect_peer(3);
+    f.complete_peering_empty();
+    assert(f.pg(1, 1).state == PG_ACTIVE);
+}
+
+static void pause_ec_write_complete(osd_test_fixture_t & f);
+
+// Full-stripe write which leaves unstable versions on OSDs 1-3. Completes it unless <complete> is false
+static osd_op_t *pause_ec_write(osd_test_fixture_t & f, uint64_t client_id, bool complete = true)
+{
+    auto *wr = make_write_op(INODE_WITH_POOL(1, 1), 0, 256*1024, 0xab);
+    wr->client_id = client_id;
+    f.exec(wr);
+    f.bs_zero_read_ok(0);
+    if (complete)
+        pause_ec_write_complete(f);
+    return wr;
+}
+
+static void pause_ec_write_complete(osd_test_fixture_t & f)
+{
+    f.bs_write_ok(BS_OP_WRITE, 1);
+    for (osd_num_t p = 2; p <= 3; p++)
+    {
+        auto *pw = f.peer_take(p, OSD_OP_SEC_WRITE);
+        pw->reply.sec_rw.version = 1;
+        f.peer_complete(pw, pw->req.sec_rw.len);
+    }
+}
+
+// SYNC and then STABILIZE on the local store and both peers
+static void pause_ec_complete_sync(osd_test_fixture_t & f)
+{
+    for (osd_num_t p = 2; p <= 3; p++)
+        f.peer_complete(f.peer_take(p, OSD_OP_SEC_SYNC), 0);
+    auto *bs_sync = f.bs->take(BS_OP_SYNC);
+    bs_sync->retval = 0;
+    bs_sync->callback(bs_sync);
+    auto *bs_stab = f.bs->take(BS_OP_STABLE);
+    bs_stab->retval = 0;
+    bs_stab->callback(bs_stab);
+    for (osd_num_t p = 2; p <= 3; p++)
+        f.peer_complete(f.peer_take(p, OSD_OP_SEC_STABILIZE), 0);
+}
+
+// stop_pg() forgets unstable writes for the next primary's peering, but nobody peers a removed pool.
+// A PG of a removed pool, paused by the monitor before it's forgotten, stabilizes them first,
+// while it's still active, and refuses new writes meanwhile
+void test_pause_stabilizes_before_stop()
+{
+    printf("test_pause_stabilizes_before_stop\n");
+
+    osd_test_fixture_t f;
+    start_pause_ec_pg(f);
+    uint64_t client_id = f.connect_client();
+    pause_ec_write(f, client_id);
+
+    set_pause_ec_pool(f, false);
+    set_pause_ec_pg_config(f, true);
+    auto & pg = f.pg(1, 1);
+    assert(pg.state == PG_ACTIVE);
+    assert(pg.pause_sync == PG_PAUSE_SYNC_RUNNING);
+
+    auto *wr = make_write_op(INODE_WITH_POOL(1, 1), 0, 4096, 0xcd);
+    wr->client_id = client_id;
+    wr->req.hdr.id = 2;
+    f.exec(wr);
+    assert(wr->reply.hdr.retval == -EPIPE);
+
+    pause_ec_complete_sync(f);
+    f.pump();
+    assert(!f.has_pg(1, 1) || f.pg(1, 1).state == PG_OFFLINE);
+
+    printf("test_pause_stabilizes_before_stop passed\n");
+}
+
+void test_unpause_cancels_pause_sync()
+{
+    printf("test_unpause_cancels_pause_sync\n");
+
+    osd_test_fixture_t f;
+    start_pause_ec_pg(f);
+    uint64_t client_id = f.connect_client();
+    pause_ec_write(f, client_id);
+
+    set_pause_ec_pool(f, false);
+    set_pause_ec_pg_config(f, true);
+    assert(f.pg(1, 1).pause_sync == PG_PAUSE_SYNC_RUNNING);
+    // Created again with the same ID
+    set_pause_ec_pool(f, true);
+    set_pause_ec_pg_config(f, false);
+    assert(f.pg(1, 1).pause_sync == 0);
+
+    pause_ec_complete_sync(f);
+    f.pump();
+    assert(f.pg(1, 1).state == PG_ACTIVE);
+
+    printf("test_unpause_cancels_pause_sync passed\n");
+}
+
+// A PG count change pauses PGs of an existing pool: their new primaries peer them, so they stop at once
+void test_pause_of_existing_pool_stops_at_once()
+{
+    printf("test_pause_of_existing_pool_stops_at_once\n");
+
+    osd_test_fixture_t f;
+    start_pause_ec_pg(f);
+    uint64_t client_id = f.connect_client();
+    pause_ec_write(f, client_id);
+
+    set_pause_ec_pg_config(f, true);
+    assert(!f.peer_has_sent_op(2, OSD_OP_SEC_SYNC));
+    f.pump();
+    assert(!f.has_pg(1, 1) || f.pg(1, 1).state == PG_OFFLINE);
+
+    printf("test_pause_of_existing_pool_stops_at_once passed\n");
+}
+
+// A write in flight when the PG is paused is stabilized too: the sync waits for it
+void test_pause_sync_waits_for_inflight_write()
+{
+    printf("test_pause_sync_waits_for_inflight_write\n");
+
+    osd_test_fixture_t f;
+    start_pause_ec_pg(f);
+    uint64_t client_id = f.connect_client();
+    auto *wr = pause_ec_write(f, client_id, false);
+
+    set_pause_ec_pool(f, false);
+    set_pause_ec_pg_config(f, true);
+    auto & pg = f.pg(1, 1);
+    assert(pg.state == PG_ACTIVE);
+    assert(pg.pause_sync == PG_PAUSE_SYNC_WAIT);
+    assert(!f.peer_has_sent_op(2, OSD_OP_SEC_SYNC));
+
+    pause_ec_write_complete(f);
+    assert(wr->reply.hdr.retval == wr->req.rw.len);
+    assert(pg.pause_sync == PG_PAUSE_SYNC_RUNNING);
+
+    pause_ec_complete_sync(f);
+    f.pump();
+    assert(!f.has_pg(1, 1) || f.pg(1, 1).state == PG_OFFLINE);
+
+    printf("test_pause_sync_waits_for_inflight_write passed\n");
+}
+
 int main(int narg, char *args[])
 {
     test_load_global_config();
@@ -3763,5 +3954,9 @@ int main(int narg, char *args[])
     test_delete_waits_for_stabilize_snapshot();
     test_delete_waits_only_for_covered_objects();
     test_replicated_add_object_to_set_degraded();
+    test_pause_stabilizes_before_stop();
+    test_unpause_cancels_pause_sync();
+    test_pause_sync_waits_for_inflight_write();
+    test_pause_of_existing_pool_stops_at_once();
     return 0;
 }

src/test/test_cluster_client.cpp

@@ -168,16 +168,16 @@ void pretend_disconnected(osd_messenger_t *msgr, osd_num_t osd_num)
 
 void pretend_connected(cluster_client_t *cli, osd_num_t osd_num)
 {
-    pretend_connected(&cli->msgr, osd_num);
+    pretend_connected(cli->msgr, osd_num);
 }
 
 void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)
 {
-    pretend_disconnected(&cli->msgr, osd_num);
+    pretend_disconnected(cli->msgr, osd_num);
 }
 
 #define check_disconnected(cli, osd_num) {\
-    if (cli->msgr.osd_peers.find(osd_num) != cli->msgr.osd_peers.end())\
+    if (cli->msgr->osd_peers.find(osd_num) != cli->msgr->osd_peers.end())\
     {\
         printf("OSD %ju not disconnected as it ought to be\n", (uint64_t)osd_num);\
         assert(0);\
@@ -185,7 +185,7 @@ void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)
 }
 
 #define check_op_count(cli, osd_num, ops) {\
-    osd_client_t *cl = cli->msgr.osd_peers.at(osd_num);\
+    osd_client_t *cl = cli->msgr->osd_peers.at(osd_num);\
     int real_ops = cl->sent_ops.size();\
     if (real_ops != ops)\
     {\
@@ -196,7 +196,7 @@ void pretend_disconnected(cluster_client_t *cli, osd_num_t osd_num)
 
 osd_op_t *find_op(cluster_client_t *cli, osd_num_t osd_num, uint64_t opcode, uint64_t offset, uint64_t len)
 {
-    osd_client_t *cl = cli->msgr.osd_peers.at(osd_num);
+    osd_client_t *cl = cli->msgr->osd_peers.at(osd_num);
     auto op_it = cl->sent_ops.begin();
     while (op_it != cl->sent_ops.end())
     {
@@ -225,7 +225,7 @@ void pretend_op_completed(cluster_client_t *cli, osd_op_t *op, int64_t retval)
         ? "sync" : (op->req.hdr.opcode == OSD_OP_WRITE ? "write" : "read"), op->req.rw.offset, op->req.rw.len);
     uint64_t op_id = op->req.hdr.id;
     uint64_t client_id = op->client_id;
-    cli->msgr.clients[client_id]->sent_ops.erase(op_id);
+    cli->msgr->clients[client_id]->sent_ops.erase(op_id);
     op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
     op->reply.hdr.id = op->req.hdr.id;
     op->reply.hdr.opcode = op->req.hdr.opcode;
@@ -309,7 +309,7 @@ void test1()
         uint64_t replay_start = UINT64_MAX;
         uint64_t replay_end = 0;
         std::vector<osd_op_t*> replay_ops;
-        auto osd_cl = cli->msgr.osd_peers.at(1);
+        auto osd_cl = cli->msgr->osd_peers.at(1);
         for (auto & op_p: osd_cl->sent_ops)
         {
             auto op = op_p.second;
@@ -742,7 +742,7 @@ void test_deoptimize_snapshot_read()
         };
         cli->execute(op);
         check_op_count(cli, 1, 1);
-        auto cl = cli->msgr.osd_peers.at(1);
+        auto cl = cli->msgr->osd_peers.at(1);
         osd_op_t *read_op = cl->sent_ops.begin()->second;
         assert(read_op->req.hdr.opcode == OSD_OP_READ);
         assert(read_op->req.rw.inode == child_inode);
@@ -784,7 +784,7 @@ void test_deoptimize_snapshot_read()
         };
         cli->execute(op);
         check_op_count(cli, 1, 1);
-        auto cl = cli->msgr.osd_peers.at(1);
+        auto cl = cli->msgr->osd_peers.at(1);
         osd_op_t *read_op = cl->sent_ops.begin()->second;
         assert(read_op->req.hdr.opcode == OSD_OP_READ);
         assert(read_op->req.rw.inode == child_inode);
@@ -818,7 +818,7 @@ void test_deoptimize_snapshot_read()
         };
         cli->execute(op);
         check_op_count(cli, 1, 1);
-        auto cl = cli->msgr.osd_peers.at(1);
+        auto cl = cli->msgr->osd_peers.at(1);
         osd_op_t *child_read = cl->sent_ops.begin()->second;
         assert(child_read->req.hdr.opcode == OSD_OP_READ);
         assert(child_read->req.rw.inode == child_inode);

tests/run_tests.sh

@@ -71,6 +71,8 @@ OLD=1 ./test_move_reappear.sh
 ./test_rm_degraded.sh
 ./test_describe.sh
 
+OLD=1 ./test_rm_pool_unstable.sh
+
 ./test_snapshot_chain.sh
 SCHEME=ec ./test_snapshot_chain.sh
 ENCRYPTED=1 ./test_snapshot_chain.sh
@@ -83,6 +85,8 @@ SCHEME=ec ./test_snapshot_down.sh
 OLD=1 ./test_snapshot_down.sh
 OLD=1 SCHEME=ec ./test_snapshot_down.sh
 
+./test_snapshot_rm_parallel.sh
+
 ./test_kv_stress.sh
 IMMEDIATE_COMMIT=1 ./test_kv_stress.sh
 

tests/test_rm_pool_unstable.sh

@@ -0,0 +1,95 @@
+#!/bin/bash -ex
+
+# The primary must stabilize unstable writes of a pool being removed before its PGs stop:
+# nobody peers a removed pool, so nobody else would stabilize or roll them back, and
+# the old store can't trim its journal past them, so all other pools get stuck with -EAGAIN
+
+OLD=1
+SCHEME=ec
+OSD_COUNT=3
+PG_SIZE=3
+PG_MINSIZE=2
+PG_DATA_SIZE=2
+OFFSET_ARGS="$OFFSET_ARGS --journal_size 16M"
+# Without the autosync timer, writes stay unstable until the pool is removed.
+# The other autosync limit is journal_size/block_size/2 = 64 writes on a primary
+GLOBAL_CONFIG="$GLOBAL_CONFIG"',"autosync_interval":0,"client_enable_writeback":false'
+
+. `dirname $0`/run_3osds.sh
+
+# A primary stabilizes all its unstable writes on any SYNC, so the removed pools
+# and the remaining one need different primaries
+$ETCDCTL put /vitastor/config/osd/1 '{"tags":["other"]}'
+$ETCDCTL put /vitastor/config/osd/2 '{"tags":["removed"]}'
+$VITASTOR_CLI modify-pool testpool --primary_affinity_tags removed
+$VITASTOR_CLI create-pool testpool2 -s 3 -n 1 --failure_domain osd --primary_affinity_tags other
+
+pool_primary()
+{
+    $ETCDCTL get /vitastor/pg/config --print-value-only | jq -r '.items["'$1'"]["1"].primary'
+}
+
+check_primaries()
+{
+    wait_pool_up 60 $1 3 1
+    wait_pool_up 60 2 3 1
+    if [[ $(pool_primary $1) != 2 || $(pool_primary 2) != 1 ]]; then
+        format_error "Primaries are not where the test needs them"
+    fi
+}
+
+# Number of unstable writes of inode 1 of the given pool on all OSDs
+count_unstable()
+{
+    local pool=$1
+    local inode=$(printf 0x%x $(((pool << 48) | 1)))
+    for i in $(seq 1 $OSD_COUNT); do
+        local offsets=$(build/src/disk_tool/vitastor-disk simple-offsets --format json ./testdata/bin/test_osd$i.bin $OFFSET_ARGS)
+        build/src/disk_tool/vitastor-disk dump-journal --io cached --json ./testdata/bin/test_osd$i.bin 4096 \
+            $(echo "$offsets" | jq -r '"\(.journal_offset) \(.meta_offset-.journal_offset)"') | \
+            jq --arg inode $inode '[ .[] | select(.inode == $inode) ] |
+                ([ .[] | select(.type == "stable") | {stripe, ver: (.ver|tonumber)} ]) as $stable |
+                ([ .[] | select(.type == "rollback") | {stripe, ver: (.ver|tonumber)} ]) as $rollback |
+                [ .[] | select(.type == "big_write" or .type == "small_write") | .stripe as $s | (.ver|tonumber) as $v |
+                    select(([ $stable[] | select(.stripe == $s and .ver >= $v) ] | length) == 0 and
+                        ([ $rollback[] | select(.stripe == $s and .ver < $v) ] | length) == 0) ] | length'
+    done | jq -s add
+}
+
+write_unstable()
+{
+    local pool=$1
+    # 8 full stripes per pass, both passes together stay below the autosync limit
+    if ! timeout -k 10 60 $VITASTOR_FIO -bs=256k -direct=1 -iodepth=1 -rw=write -pool=$pool -inode=1 -size=2M -end_fsync=0; then
+        format_error "Writes to pool $pool got stuck"
+    fi
+    # A sync from another pool syncs the whole store, which puts these writes into the journal
+    $VITASTOR_FIO -bs=4k -direct=1 -iodepth=1 -rw=write -pool=2 -inode=1 -size=4k -end_fsync=1
+    if [[ $(count_unstable $pool) -eq 0 ]]; then
+        format_error "No unstable writes of pool $pool on OSDs, the test checks nothing"
+    fi
+}
+
+# Writes 4x the journal size, so the journal has to be trimmed past the removed pool's writes
+write_other_pool()
+{
+    if ! timeout -k 10 60 $VITASTOR_FIO -bs=4k -direct=1 -iodepth=16 -fsync=32 -rw=randwrite \
+        -pool=2 -inode=1 -size=128M -number_ios=16384 -randrepeat=0; then
+        format_error "$1: writes to another pool got stuck"
+    fi
+}
+
+check_rolled_back()
+{
+    if [[ $(count_unstable $1) -ne 0 ]]; then
+        format_error "Unstable writes of removed pool $1 are still on OSDs"
+    fi
+}
+
+check_primaries 1
+write_unstable 1
+$VITASTOR_CLI rm-pool testpool
+write_other_pool "Pool removed"
+check_rolled_back 1
+
+format_green OK

tests/test_snapshot_rm_parallel.sh

@@ -0,0 +1,43 @@
+#!/bin/bash -ex
+
+. `dirname $0`/run_3osds.sh
+check_qemu
+
+# Test parallel rm of two snapshots
+
+$VITASTOR_CLI create -s 32M repro-golden
+$VITASTOR_FIO -bs=1M -direct=1 -iodepth=4 -end_fsync=1 -rw=write \
+    -image=repro-golden -mirror_file=./testdata/bin/mirror.bin
+qemu-img convert -p \
+    -f raw "vitastor:config_path=$VITASTOR_CFG:image=repro-golden" \
+    -O raw ./testdata/bin/res.bin
+cmp ./testdata/bin/mirror.bin ./testdata/bin/res.bin
+
+$VITASTOR_CLI snap-create repro-golden@init
+
+for i in {1..10}; do
+    $VITASTOR_CLI ls --tree
+    $VITASTOR_CLI snap-create repro-golden@a$i
+    $VITASTOR_CLI snap-create repro-golden@b$i
+    $VITASTOR_CLI create --parent repro-golden@a$i -s 32M repro-cl-a$i
+    $VITASTOR_CLI create --parent repro-golden@b$i -s 32M repro-cl-b$i
+    ($VITASTOR_CLI rm repro-golden@a$i &>./testdata/rm_a$i.log || true) & # two concurrent rm's
+    CLI1=$!
+    ($VITASTOR_CLI rm repro-golden@b$i &>./testdata/rm_b$i.log || true) &
+    CLI2=$!
+    wait $CLI1 $CLI2
+    qemu-img convert -p \
+        -f raw "vitastor:config_path=$VITASTOR_CFG:image=repro-golden" \
+        -O raw ./testdata/bin/res.bin
+    if ! cmp ./testdata/bin/mirror.bin ./testdata/bin/res.bin; then
+        sleep 100000
+    fi
+done
+
+$VITASTOR_CLI ls --tree
+
+$VITASTOR_CLI rm --matching repro-golden@a* repro-golden@b*
+
+$VITASTOR_CLI ls --tree
+
+format_green OK