osd: Roll back unstable writes of removed pools in the old store

Nothing stabilizes or rolls back unstable writes of a removed pool, its
PGs are never peered again. The old store can't trim its journal past
them, so the journal fills up and writes to all other pools on the same
OSDs fail with -EAGAIN forever, restarts don't help.

Add a synchronous get_unstable_objects() which walks dirty_db of the old
store, and roll back unstable writes of removed pools with it. The OSD
only picks up a pool when it disappears from the pool configuration, or
on start when the store has unstable writes of a pool that doesn't
exist, and lists the store once the pool's PGs are removed and stopped.
The new store keeps such writes in its metadata without harm, so the
OSD does nothing for it.

.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