.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

src/blockstore/blockstore.h

@@ -214,6 +214,13 @@ public:
     // Simplified synchronous operation: get object bitmap & current version
     virtual int read_bitmap(object_id oid, uint64_t target_version, void *bitmap, uint64_t *result_version = NULL) = 0;
 
+    // Simplified synchronous operation: get objects with unstable writes, each with the version
+    // to roll them back to. Only the old store returns them: it can't trim its journal past them
+    virtual std::vector<obj_ver_id> get_unstable_objects() { return {}; }
+
+    // True for the new store (meta_format heap)
+    virtual bool is_new() { return false; }
+
     // Get per-inode space usage statistics
     virtual const std::map<uint64_t, uint64_t> & get_inode_space_stats() = 0;
 

src/blockstore/blockstore_impl.h

@@ -219,6 +219,8 @@ public:
     // Simplified synchronous operation: get object bitmap & current version
     int read_bitmap(object_id oid, uint64_t target_version, void *bitmap, uint64_t *result_version = NULL);
 
+    bool is_new() { return true; }
+
     // Set per-pool no_inode_stats
     void set_no_inode_stats(const std::vector<uint64_t> & pool_ids);
 

src/blockstore/v1/impl.h

@@ -334,6 +334,9 @@ public:
     // Simplified synchronous operation: get object bitmap & current version
     int read_bitmap(object_id oid, uint64_t target_version, void *bitmap, uint64_t *result_version = NULL);
 
+    // Get objects with unstable writes, each with the version preceding the first unstable one
+    std::vector<obj_ver_id> get_unstable_objects();
+
     // Unstable writes are added here (map of object_id -> version)
     std::unordered_map<object_id, uint64_t> unstable_writes;
 

src/blockstore/v1/rollback.cpp

@@ -294,4 +294,27 @@ void blockstore_impl_t::free_dirty_dyn_data(copy_buffer_t & e)
     }
 }
 
+std::vector<obj_ver_id> blockstore_impl_t::get_unstable_objects()
+{
+    std::vector<obj_ver_id> result;
+    for (auto dirty_it = dirty_db.begin(); dirty_it != dirty_db.end(); )
+    {
+        // Instant writes and deletes become stable on sync by themselves
+        if (IS_STABLE(dirty_it->second.state) || IS_INSTANT(dirty_it->second.state))
+        {
+            dirty_it++;
+            continue;
+        }
+        result.push_back((obj_ver_id){
+            .oid = dirty_it->first.oid,
+            .version = dirty_it->first.version-1,
+        });
+        dirty_it = dirty_db.lower_bound((obj_ver_id){
+            .oid = dirty_it->first.oid,
+            .version = UINT64_MAX,
+        });
+    }
+    return result;
+}
+
 } // namespace v1

src/osd/osd.h

@@ -202,6 +202,11 @@ class osd_t
     // to release what a dropped callback will never free
     robin_hood::unordered_set<osd_op_t*> pending_owned_ops;
     std::map<object_id, osd_op_t*> scrub_ops;
+    // Rollback of unstable writes of removed pools: pools to roll back and pools seen in the config
+    std::set<pool_id_t> rm_pools, rm_pools_existing;
+    std::vector<obj_ver_id> rm_pools_versions;
+    size_t rm_pools_version_pos = 0;
+    uint64_t rm_pools_rolled_back_count = 0;
     bool recovery_last_degraded = true;
     pool_pg_num_t recovery_last_pg;
     object_id recovery_last_oid;
@@ -331,6 +336,10 @@ class osd_t
     void submit_pg_flush_ops(pg_t & pg);
     void handle_flush_op(bool rollback, pool_id_t pool_id, pg_num_t pg_num, pg_flush_batch_t *fb, osd_num_t peer_osd, int retval);
     bool submit_flush_op(pool_id_t pool_id, pg_num_t pg_num, pg_flush_batch_t *fb, bool rollback, osd_num_t peer_osd, int count, obj_ver_id *data);
+    void check_removed_pools(bool on_load);
+    void rollback_removed_pools();
+    void submit_removed_pool_rollback();
+    void finish_removed_pools_rollback();
     bool pick_next_recovery(osd_recovery_op_t &op);
     void submit_recovery_op(osd_recovery_op_t *op);
     void finish_recovery_op(osd_recovery_op_t *op);

src/osd/osd_cluster.cpp

@@ -492,19 +492,27 @@ void osd_t::on_change_etcd_state_hook(std::map<std::string, etcd_kv_t> & changes
         parse_config(false);
     }
     bool pools = changes.find(st_cli->etcd_prefix+"/config/pools") != changes.end();
+    bool pgs = changes.find(st_cli->etcd_prefix+"/pg/config") != changes.end();
     if (pools)
     {
         apply_no_inode_stats();
     }
     if (run_primary)
     {
-        bool pgs = changes.find(st_cli->etcd_prefix+"/pg/config") != changes.end();
         if (pools || pgs)
         {
             apply_pg_count();
         }
         apply_pg_config();
     }
+    if (pools)
+    {
+        check_removed_pools(false);
+    }
+    else if (pgs && rm_pools.size())
+    {
+        rollback_removed_pools();
+    }
 }
 
 void osd_t::on_load_config_hook(json11::Json::object & global_config)
@@ -728,6 +736,7 @@ void osd_t::on_load_pgs_hook(bool success)
             apply_pg_count();
             apply_pg_config();
         }
+        check_removed_pools(true);
     }
 }
 

src/osd/osd_flush.cpp

@@ -4,6 +4,8 @@
 #include "osd.h"
 
 #define FLUSH_BATCH 512
+// Small enough to fit into an almost full journal
+#define RM_POOL_ROLLBACK_BATCH 32
 
 static_assert(FLUSH_BATCH <= MAX_SIMPLE_PAYLOAD_SIZE / sizeof(obj_ver_id));
 
@@ -473,3 +475,166 @@ bool osd_t::continue_recovery()
     }
     return true;
 }
+
+// Peering never visits a removed pool again, so nothing would ever stabilize or roll back
+// its unstable writes, and the old store can't trim its journal past them.
+// Only runs on load and when the pool config changes, and does nothing until a pool disappears
+void osd_t::check_removed_pools(bool on_load)
+{
+    if (!bs || bs->is_new())
+    {
+        return;
+    }
+    if (on_load)
+    {
+        // Pools may have been removed while the OSD was down
+        for (auto & ov: bs->get_unstable_objects())
+        {
+            auto pool_it = st_cli->pool_config.find(INODE_POOL(ov.oid.inode));
+            if (pool_it == st_cli->pool_config.end() || !pool_it->second.exists)
+            {
+                rm_pools.insert(INODE_POOL(ov.oid.inode));
+            }
+        }
+    }
+    for (auto & pp: st_cli->pool_config)
+    {
+        if (pp.second.exists)
+        {
+            rm_pools_existing.insert(pp.first);
+        }
+        else if (rm_pools_existing.erase(pp.first))
+        {
+            rm_pools.insert(pp.first);
+        }
+    }
+    if (rm_pools.size())
+    {
+        rollback_removed_pools();
+    }
+}
+
+void osd_t::rollback_removed_pools()
+{
+    if (rm_pools_versions.size())
+    {
+        // finish_removed_pools_rollback() comes back here
+        return;
+    }
+    std::set<pool_id_t> ready;
+    for (auto rm_it = rm_pools.begin(); rm_it != rm_pools.end(); )
+    {
+        pool_id_t pool_id = *rm_it;
+        auto pool_it = st_cli->pool_config.find(pool_id);
+        if (pool_it != st_cli->pool_config.end() && pool_it->second.exists)
+        {
+            // Created again with the same ID
+            rm_pools.erase(rm_it++);
+            continue;
+        }
+        // Wait until its PGs are removed and stopped, so that no write lands after the pass
+        bool busy = false;
+        if (pool_it != st_cli->pool_config.end())
+        {
+            for (auto & pg_item: pool_it->second.pg_config)
+            {
+                if (pg_item.second.config_exists)
+                {
+                    busy = true;
+                    break;
+                }
+            }
+        }
+        for (auto pg_it = pgs.lower_bound({ .pool_id = pool_id, .pg_num = 0 });
+            !busy && pg_it != pgs.end() && pg_it->first.pool_id == pool_id; pg_it++)
+        {
+            busy = pg_it->second.state != PG_OFFLINE;
+        }
+        if (busy)
+        {
+            rm_it++;
+            continue;
+        }
+        ready.insert(pool_id);
+        rm_pools.erase(rm_it++);
+    }
+    if (!ready.size())
+    {
+        return;
+    }
+    // List the store only now: writes still in flight on removal may have been stabilized since
+    for (auto & ov: bs->get_unstable_objects())
+    {
+        if (ready.find(INODE_POOL(ov.oid.inode)) != ready.end())
+        {
+            rm_pools_versions.push_back(ov);
+        }
+    }
+    rm_pools_version_pos = 0;
+    rm_pools_rolled_back_count = 0;
+    submit_removed_pool_rollback();
+}
+
+void osd_t::submit_removed_pool_rollback()
+{
+    std::vector<obj_ver_id> batch;
+    while (rm_pools_version_pos < rm_pools_versions.size() && batch.size() < RM_POOL_ROLLBACK_BATCH)
+    {
+        auto & ov = rm_pools_versions[rm_pools_version_pos++];
+        // The pool ID may already be taken by a pool created after the listing
+        auto pool_it = st_cli->pool_config.find(INODE_POOL(ov.oid.inode));
+        if (pool_it == st_cli->pool_config.end() || !pool_it->second.exists)
+        {
+            batch.push_back(ov);
+        }
+    }
+    if (!batch.size())
+    {
+        finish_removed_pools_rollback();
+        return;
+    }
+    osd_op_t *op = new osd_op_t();
+    pending_owned_ops.insert(op);
+    op->buf = malloc_or_die(sizeof(obj_ver_id) * batch.size());
+    memcpy(op->buf, batch.data(), sizeof(obj_ver_id) * batch.size());
+    op->bs_op = new blockstore_op_t((blockstore_op_t){
+        .opcode = BS_OP_ROLLBACK,
+        .callback = [this, op](blockstore_op_t *bs_op)
+        {
+            int retval = bs_op->retval;
+            uint32_t count = bs_op->len;
+            delete op->bs_op;
+            op->bs_op = NULL;
+            pending_owned_ops.erase(op);
+            delete op;
+            if (retval < 0)
+            {
+                printf("[OSD %ju] Failed to roll back unstable writes of removed pools: retval=%d\n", osd_num, retval);
+                rm_pools_version_pos = rm_pools_versions.size();
+            }
+            else
+            {
+                rm_pools_rolled_back_count += count;
+            }
+            submit_removed_pool_rollback();
+        },
+        {
+            .len = (uint32_t)batch.size(),
+        },
+        .buf = (uint8_t*)op->buf,
+    });
+    bs->enqueue_op(op->bs_op);
+}
+
+void osd_t::finish_removed_pools_rollback()
+{
+    if (rm_pools_rolled_back_count > 0)
+    {
+        printf("[OSD %ju] Rolled back %ju unstable object versions of removed pools\n", osd_num, rm_pools_rolled_back_count);
+    }
+    rm_pools_versions.clear();
+    if (rm_pools.size())
+    {
+        rollback_removed_pools();
+    }
+}

src/osd/osd_peering.cpp

@@ -706,6 +706,10 @@ void osd_t::finish_stop_pg(pg_t & pg)
     pg.state = PG_OFFLINE;
     reset_pg(pg);
     report_pg_state(pg);
+    if (rm_pools.find(pg.pool_id) != rm_pools.end() && !in_destruction)
+    {
+        rollback_removed_pools();
+    }
 }
 
 void osd_t::report_pg_state(pg_t & pg)

src/test/osd_test.cpp

@@ -3719,6 +3719,149 @@ void test_replicated_add_object_to_set_degraded()
     printf("test_replicated_add_object_to_set_degraded passed\n");
 }
 
+// Unstable writes of a pool removed while the OSD was down are rolled back on start:
+// peering never visits the pool again, and the old store can't trim its journal past them
+void test_rollback_removed_pool_on_start()
+{
+    printf("test_rollback_removed_pool_on_start\n");
+
+    osd_test_fixture_t f;
+    f.configure_replicated_pool(2, 2, 1, 1, { { 1, 2 } });
+    object_id a = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0 };
+    object_id b = { .inode = INODE_WITH_POOL(1, 1), .stripe = 0x20000 };
+    object_id c = { .inode = INODE_WITH_POOL(2, 1), .stripe = 0 };
+    object_id d = { .inode = INODE_WITH_POOL(3, 1), .stripe = 0 };
+    f.bs_unstable_objects = {
+        { .oid = a, .version = 2 },
+        { .oid = b, .version = 0 },
+        { .oid = c, .version = 5 },
+        { .oid = d, .version = 1 },
+    };
+    f.construct(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "none" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+    });
+    f.osd->start();
+
+    // Everything outside of existing pools
+    auto *rb = f.bs->take(BS_OP_ROLLBACK);
+    assert(rb->len == 3);
+    auto *ov = (obj_ver_id*)rb->buf;
+    assert(ov[0].oid == a && ov[0].version == 2);
+    assert(ov[1].oid == b && ov[1].version == 0);
+    assert(ov[2].oid == d && ov[2].version == 1);
+    rb->retval = 0;
+    std::function<void(blockstore_op_t*)>(rb->callback)(rb);
+    assert(!f.bs->take(BS_OP_ROLLBACK, false));
+
+    printf("test_rollback_removed_pool_on_start passed\n");
+}
+
+// Unstable writes of a pool removed online are rolled back once its PGs are removed
+// and stopped, so that no write can land after the pass
+void test_rollback_removed_pool_online()
+{
+    printf("test_rollback_removed_pool_online\n");
+
+    osd_test_fixture_t f;
+    f.configure_replicated_pool(1, 2, 1, 1, { { 1, 2 } });
+    f.start(json11::Json::object {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "none" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+    });
+    f.connect_peer(2);
+    f.complete_peering_empty();
+    assert(f.pg(1, 1).state == PG_ACTIVE);
+
+    // Hold a write in flight
+    inode_t inode = INODE_WITH_POOL(1, 1);
+    uint64_t client_id = f.connect_client();
+    auto *wr = make_write_op(inode, 0, 4096, 0xab);
+    wr->client_id = client_id;
+    f.exec(wr);
+    f.bs_zero_read_ok(0);
+    auto *local = f.bs->take(BS_OP_WRITE_STABLE);
+    auto *peer_wr = f.peer_take(2, OSD_OP_SEC_WRITE_STABLE);
+    object_id a = { .inode = inode, .stripe = 0 };
+    f.bs->unstable_objects = { { .oid = a, .version = 0 } };
+    int calls = f.bs->unstable_object_calls;
+
+    // The pool is removed, but its PGs are still there
+    f.st_cli->set("/vitastor/config/pools", json11::Json::object {});
+    assert(f.bs->unstable_object_calls == calls);
+
+    // PGs are removed, but the PG is still stopping
+    f.st_cli->set("/vitastor/pg/config", json11::Json::object {
+        { "items", json11::Json::object {} },
+    });
+    assert(f.pg(1, 1).state & PG_STOPPING);
+    assert(f.bs->unstable_object_calls == calls);
+
+    local->retval = local->len;
+    local->callback(local);
+    peer_wr->reply.sec_rw.version = 1;
+    f.peer_complete(peer_wr, peer_wr->req.sec_rw.len);
+    assert(!f.has_pg(1, 1) || f.pg(1, 1).state == PG_OFFLINE);
+    assert(f.bs->unstable_object_calls == calls+1);
+
+    auto *rb = f.bs->take(BS_OP_ROLLBACK);
+    assert(rb->len == 1);
+    assert(((obj_ver_id*)rb->buf)[0].oid == a && ((obj_ver_id*)rb->buf)[0].version == 0);
+    rb->retval = 0;
+    std::function<void(blockstore_op_t*)>(rb->callback)(rb);
+    assert(!f.bs->take(BS_OP_ROLLBACK, false));
+
+    printf("test_rollback_removed_pool_online passed\n");
+}
+
+// Without a removed pool, pool and PG changes don't look into the store,
+// and the new store isn't looked into at all
+void test_rollback_removed_pool_idle()
+{
+    printf("test_rollback_removed_pool_idle\n");
+
+    json11::Json::object cfg = {
+        { "osd_num", 1 },
+        { "etcd_address", "127.0.0.1:2379" },
+        { "immediate_commit", "none" },
+        { "block_size", 131072 },
+        { "bitmap_granularity", 4096 },
+    };
+    object_id a = { .inode = INODE_WITH_POOL(2, 1), .stripe = 0 };
+    {
+        osd_test_fixture_t f;
+        f.configure_replicated_pool(1, 2, 1, 1, { { 1, 2 } });
+        f.start(cfg);
+        assert(f.bs->unstable_object_calls == 1);
+        f.bs->unstable_objects = { { .oid = a, .version = 0 } };
+        f.configure_replicated_pool(1, 2, 1, 1, { { 2, 1 } });
+        f.configure_replicated_pool(1, 2, 1, 1, { { 1, 2 } });
+        assert(f.bs->unstable_object_calls == 1);
+        assert(!f.bs->take(BS_OP_ROLLBACK, false));
+    }
+    {
+        osd_test_fixture_t f;
+        f.bs_new_store = true;
+        f.bs_unstable_objects = { { .oid = a, .version = 0 } };
+        f.configure_replicated_pool(1, 2, 1, 1, { { 1, 2 } });
+        f.start(cfg);
+        f.st_cli->set("/vitastor/config/pools", json11::Json::object {});
+        f.st_cli->set("/vitastor/pg/config", json11::Json::object {
+            { "items", json11::Json::object {} },
+        });
+        assert(f.bs->unstable_object_calls == 0);
+        assert(!f.bs->take(BS_OP_ROLLBACK, false));
+    }
+
+    printf("test_rollback_removed_pool_idle passed\n");
+}
+
 int main(int narg, char *args[])
 {
     test_load_global_config();
@@ -3763,5 +3906,8 @@ 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_rollback_removed_pool_on_start();
+    test_rollback_removed_pool_online();
+    test_rollback_removed_pool_idle();
     return 0;
 }

src/test/osd_test_fixture.h

@@ -22,12 +22,26 @@ class capturing_bs_t: public blockstore_mock_t
 public:
     using blockstore_mock_t::blockstore_mock_t;
     std::vector<blockstore_op_t*> queued;
+    std::vector<obj_ver_id> unstable_objects;
+    int unstable_object_calls = 0;
+    bool new_store = false;
 
     void enqueue_op(blockstore_op_t *op) override
     {
         queued.push_back(op);
     }
 
+    std::vector<obj_ver_id> get_unstable_objects() override
+    {
+        unstable_object_calls++;
+        return unstable_objects;
+    }
+
+    bool is_new() override
+    {
+        return new_store;
+    }
+
     // Pop the first queued op
     blockstore_op_t *take()
     {
@@ -79,6 +93,9 @@ struct osd_test_fixture_t
     etcd_state_client_mock_t *st_cli = nullptr; // owned by osd_t after start()
     osd_t *osd = nullptr;
     capturing_bs_t *bs = nullptr;
+    // What the store reports to get_unstable_objects() from the start
+    std::vector<obj_ver_id> bs_unstable_objects;
+    bool bs_new_store = false;
 
     osd_test_fixture_t()
     {
@@ -178,7 +195,10 @@ struct osd_test_fixture_t
         osd = new osd_t(osd_config, ringloop, tfd,
             std::unique_ptr<etcd_state_client_t>(st_cli),
             [this](blockstore_config_t & cfg) -> blockstore_i* {
-                return (bs = new capturing_bs_t(cfg));
+                bs = new capturing_bs_t(cfg);
+                bs->unstable_objects = bs_unstable_objects;
+                bs->new_store = bs_new_store;
+                return bs;
             });
     }
 

src/test/test_blockstore_v1.cpp

@@ -2888,6 +2888,71 @@ static void test_zero_write_over_rollback_to_zero()
     free(buf);
 }
 
+// get_unstable_objects() returns each object with unstable writes once, with the version
+// to roll it back to. Instant writes become stable on sync by themselves and are skipped
+static void test_get_unstable_objects()
+{
+    printf("\n-- test_get_unstable_objects\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    test.config["immediate_commit"] = "none";
+    test.init();
+
+    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
+    memset(buf, 0xAA, 4096);
+    object_id a = { .inode = 1, .stripe = 0 };
+    object_id b = { .inode = 1, .stripe = 0x20000 };
+    object_id c = { .inode = 2, .stripe = 0 };
+    object_id d = { .inode = 3, .stripe = 0 };
+    auto write = [&](object_id oid, uint64_t version, int opcode)
+    {
+        blockstore_op_t op;
+        op.opcode = opcode;
+        op.oid = oid;
+        op.version = version;
+        op.offset = 0;
+        op.len = 4096;
+        op.buf = buf;
+        test.exec_op(&op);
+        assert(op.retval == (int)op.len);
+    };
+    auto stab = [&](int opcode, std::vector<obj_ver_id> list)
+    {
+        blockstore_op_t op;
+        op.opcode = opcode;
+        op.len = list.size();
+        op.buf = (uint8_t*)list.data();
+        test.exec_op(&op);
+        assert(op.retval == 0);
+    };
+    write(a, 1, BS_OP_WRITE);
+    write(a, 2, BS_OP_WRITE);
+    write(a, 3, BS_OP_WRITE);
+    write(b, 1, BS_OP_WRITE);
+    write(c, 1, BS_OP_WRITE);
+    write(d, 1, BS_OP_WRITE_STABLE);
+    blockstore_op_t op;
+    op.opcode = BS_OP_SYNC;
+    test.exec_op(&op);
+    assert(op.retval == 0);
+    stab(BS_OP_STABLE, { { .oid = a, .version = 1 }, { .oid = c, .version = 1 } });
+    write(d, 2, BS_OP_WRITE_STABLE);
+
+    auto unstable = test.bs->get_unstable_objects();
+    assert(unstable.size() == 2);
+    assert(unstable[0].oid == a && unstable[0].version == 1);
+    assert(unstable[1].oid == b && unstable[1].version == 0);
+
+    stab(BS_OP_ROLLBACK, unstable);
+    assert(!test.bs->get_unstable_objects().size());
+    assert(test.find_dirty_entry(a, 1));
+    assert(!test.find_dirty_entry(a, 2));
+    assert(!test.find_dirty_entry(b, 1));
+
+    free(buf);
+}
+
 int main(int narg, char *args[])
 {
     test_preserve_corruption();
@@ -2921,5 +2986,6 @@ int main(int narg, char *args[])
     test_rollback_over_stable_write();
     test_sync_waits_for_delete_taking_over_unsynced();
     test_zero_write_over_rollback_to_zero();
+    test_get_unstable_objects();
     return 0;
 }

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

tests/test_rm_pool_unstable.sh

@@ -0,0 +1,118 @@
+#!/bin/bash -ex
+
+# Unstable writes of a removed pool must not stay in the old store forever:
+# nobody is left to stabilize or roll them back, and the journal can't be
+# trimmed 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
+}
+
+# 1. Pool removed while its OSDs are running
+
+check_primaries 1
+write_unstable 1
+$VITASTOR_CLI rm-pool testpool
+write_other_pool "Pool removed online"
+check_rolled_back 1
+
+# 2. Pool removed while its OSDs are down
+
+$VITASTOR_CLI create-pool testpool3 --ec 2+1 -n 1 --failure_domain osd --primary_affinity_tags removed
+check_primaries 3
+write_unstable 3
+# All at once, so that the primary doesn't roll them back during peering
+for i in $(seq 1 $OSD_COUNT); do
+    pid=OSD${i}_PID
+    kill -9 ${!pid}
+done
+for i in $(seq 1 $OSD_COUNT); do
+    $ETCDCTL del /vitastor/osd/state/$i
+done
+$VITASTOR_CLI rm-pool testpool3
+for i in $(seq 1 $OSD_COUNT); do
+    start_osd $i
+done
+wait_pool_up 60 2 3 1
+write_other_pool "Pool removed offline"
+check_rolled_back 3
+
+format_green OK