.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/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/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;
 }

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,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