Add an option to not move primary OSDs of unclean PGs

Co-Authored-By: Yury Luneff <koolkhel@mail.ru>

docs/usage/cli.en.md

@@ -231,8 +231,8 @@ Options:
 * `--writers-stopped` allows optimised removal in case of a single 'slim' read-write
   child and 'fat' removed parent: the child is merged into parent and parent is renamed
   to child in that case. In other cases parent layers are always merged into children.
-* `--exact` - remove multiple images with given names.
-* `--matching` - remove multiple images with names matching given glob patterns.
+* `--exact` - remove multiple images with names matching given glob patterns.
+* `--matching` - remove multiple images with given names
 * `--down-ok` - continue deletion/merging even if some data will be left on unavailable OSDs.
 
 ## flatten
@@ -287,7 +287,7 @@ Describe unclean object locations in the cluster. Options:
 
 ## fix
 
-`vitastor-cli fix [--objects <objects>] --bad-osds <osds> [--part <part>] [--check no]`
+`vitastor-cli fix [--objects <objects>] [--bad-osds <osds>] [--part <part>] [--check no]`
 
 Fix inconsistent objects in the cluster by deleting some copies.
 
@@ -298,7 +298,7 @@ Fix inconsistent objects in the cluster by deleting some copies.
     Plain text format: 0x<inode>:0x<stripe> <any delimiter> 0x<inode>:0x<stripe> ...
     JSON format: [{"inode":"0x...","stripe":"0x..."},...]
 --bad-osds <osds>
-    Required. Remove inconsistent copies/parts of objects from these OSDs, effectively
+    Remove inconsistent copies/parts of objects from these OSDs, effectively
     marking them bad and allowing Vitastor to recover objects from other copies.
 --part <number>
     Only remove EC part <number> (from 0 to pg_size-1), required for extreme
@@ -316,17 +316,14 @@ Allocate a new OSD number and reserve it by creating empty `/osd/stats/<n>` key.
 
 ## rm-osd
 
-`vitastor-cli rm-osd [-f|--force] [--allow-data-loss] [--allow-up] [--dry-run] <osd_id> [osd_id...]`
+`vitastor-cli rm-osd [--force] [--allow-data-loss] [--dry-run] <osd_id> [osd_id...]`
 
 Remove metadata and configuration for specified OSD(s) from etcd.
 
-Refuses to remove OSDs that are up or whose removal makes PGs degraded, offline or incomplete.
+Refuses to remove OSDs with data without `--force` and `--allow-data-loss`.
 
-* `-f|--force` - remove OSDs even if they are up or PGs become degraded or offline.
-* `--allow-data-loss` - remove OSDs even if PGs or objects become incomplete (DATA LOSS). Implies `--force`.
-* `--allow-up` - remove OSDs even if they are up.
-* `--dry-run` - only check if deletion is possible without data loss and
-  redundancy degradation.
+With `--dry-run` only checks if deletion is possible without data loss and
+redundancy degradation.
 
 ## osd-tree
 

docs/usage/cli.ru.md

@@ -239,8 +239,8 @@ vitastor-cli dd [iimg=<image> | if=<file>] [oimg=<image> | of=<file>] [bs=1M] \
   у удаляемой цепочки есть только один дочерний образ, содержащий небольшой объём данных.
   В этом случае дочерний образ вливается в родительский и удаляется, а родительский
   переименовывается в дочерний.
-* `--exact` - удалить все образы с точно заданными именами.
-* `--matching` - удалить все образы с именами, подходящими под переданные glob-шаблоны.
+* `--exact` - удалить все образы с именами, подходящими под переданные glob-шаблоны.
+* `--matching` - удалить все образы с точно заданными именами.
 * `--down-ok` - продолжать удаление/слияние, даже если часть данных останется неудалённой на недоступных OSD.
 
 ## flatten
@@ -302,7 +302,7 @@ vitastor-cli dd [iimg=<image> | if=<file>] [oimg=<image> | of=<file>] [bs=1M] \
 
 ## fix
 
-`vitastor-cli fix [--objects <объекты>] --bad-osds <osds> [--part <номер>] [--check no]`
+`vitastor-cli fix [--objects <объекты>] [--bad-osds <osds>] [--part <номер>] [--check no]`
 
 Исправить неконсистентные (неоднозначные) объекты путём удаления части копий.
 
@@ -313,9 +313,9 @@ vitastor-cli dd [iimg=<image> | if=<file>] [oimg=<image> | of=<file>] [bs=1M] \
     Простой формат: 0x<инод>:0x<смещение> <любой разделитель> 0x<инод>:0x<смещение> ...
     Формат JSON: [{"inode":"0x<инод>","stripe":"0x<смещение>"},...]
 --bad-osds <osds>
-    Обязательная опция. Удалить неконсистентные копии/части объектов с данных
-    OSD, таким образом признавая потерю этих копий и позволяя Vitastor-у
-    восстановить объекты из других копий.
+    Удалить неконсистентные копии/части объектов с данных OSD, таким образом
+    признавая потерю этих копий и позволяя Vitastor-у восстановить объекты из
+    других копий.
 --part <номер>
     Удалить только части EC с заданным номером (от 0 до pg_size-1). Нужно только
     в редких граничных случаях, когда один и тот же OSD содержит несколько частей
@@ -334,19 +334,14 @@ vitastor-cli dd [iimg=<image> | if=<file>] [oimg=<image> | of=<file>] [bs=1M] \
 
 ## rm-osd
 
-`vitastor-cli rm-osd [-f|--force] [--allow-data-loss] [--allow-up] [--dry-run] <osd_id> [osd_id...]`
+`vitastor-cli rm-osd [--force] [--allow-data-loss] [--dry-run] <osd_id> [osd_id...]`
 
 Удалить метаданные и конфигурацию для заданных OSD из etcd.
 
-Отказывается удалять работающие OSD, а также OSD, после удаления которых PG станут
-деградированными, недоступными (offline) или неполными (incomplete).
+Отказывается удалять OSD с данными без опций `--force` и `--allow-data-loss`.
 
-* `-f|--force` - удалить OSD, даже если они работают или PG станут деградированными или недоступными.
-* `--allow-data-loss` - удалить OSD, даже если PG или объекты станут неполными (ПОТЕРЯ ДАННЫХ).
-  Включает действие `--force`.
-* `--allow-up` - удалить OSD, даже если они работают.
-* `--dry-run` - только проверить, возможно ли удаление без потери данных и деградации
-  избыточности.
+С опцией `--dry-run` только проверяет, возможно ли удаление без потери данных и деградации
+избыточности.
 
 ## osd-tree
 

mon/etcd_schema.js

@@ -65,6 +65,7 @@ const etcd_tree = {
             placement_levels: { datacenter: 1, rack: 2, host: 3, osd: 4, ... },
             use_old_pg_combinator: false,
             osd_backfillfull_ratio: 0.99,
+            mon_hold_unclean_pg_primary: false,
             mon_metrics_path: '/prometheus',
             mon_healthcheck_path: '/healthcheck',
             // client and osd

mon/mon.js

@@ -74,6 +74,7 @@ class Mon
         this.state = JSON.parse(JSON.stringify(etcd_tree));
         this.prev_stats = { osd_stats: {}, osd_diff: {} };
         this.recheck_pgs_active = false;
+        this.primaries_held = false;
         this.updating_total_stats = false;
         this.watcher_active = false;
         this.old_pg_config = false;
@@ -199,6 +200,7 @@ class Mon
         {
             this.config.osd_out_time = 600; // 10 minutes by default
         }
+        this.config.mon_hold_unclean_pg_primary = !!this.config.mon_hold_unclean_pg_primary;
     }
 
     on_message(msg)
@@ -252,7 +254,7 @@ class Mon
         {
             this.schedule_update_stats();
         }
-        if (changed)
+        if (changed || this.primaries_held && pg_states_changed)
         {
             this.schedule_recheck();
         }
@@ -564,7 +566,8 @@ class Mon
         else
         {
             // Nothing changed, but we still want to recheck the distribution of primaries
-            let new_pg_config = recheck_primary(this.state, this.config, up_osds, osd_tree);
+            let { new_pg_config, held } = recheck_primary(this.state, this.config, up_osds, osd_tree, held);
+            this.primaries_held = held;
             if (new_pg_config)
             {
                 const ok = await this.save_pg_config(new_pg_config);
@@ -626,7 +629,7 @@ class Mon
                 etcd_request.success.push({ requestDeleteRange: {
                     key: b64(this.config.etcd_prefix+'/pool/stats/'+pool_id),
                 } });
-                save_new_pgs_txn(new_pg_config, etcd_request, this.state, this.config.etcd_prefix,
+                save_new_pgs_txn(new_pg_config, etcd_request, this.state, this.config,
                     this.etcd_watch_revision, pool_id, up_osds, osd_tree, prev_pgs, [], []);
             }
         }
@@ -669,7 +672,7 @@ class Mon
                 key: b64(this.config.etcd_prefix+'/pool/stats/'+pool_id),
                 value: b64(JSON.stringify(stats)),
             } });
-            save_new_pgs_txn(new_pg_config, etcd_request, this.state, this.config.etcd_prefix,
+            save_new_pgs_txn(new_pg_config, etcd_request, this.state, this.config,
                 this.etcd_watch_revision, pool_id, up_osds, osd_tree, real_prev_pgs, pool_res.pgs, pg_history);
         }
         new_pg_config.hash = tree_hash;

mon/pg_gen.js

@@ -10,6 +10,12 @@ const { scale_pg_count } = require('./pg_utils.js');
 const { make_hier_tree, filter_osds_by_root_node,
     filter_osds_by_tags, filter_osds_by_block_layout, get_affinity_osds } = require('./osd_tree.js');
 
+function is_clean(state)
+{
+    return (state instanceof Array) && !state.filter(s => s != 'active' && s != 'left_on_dead' &&
+        s != 'has_incomplete' && s != 'has_invalid').length;
+}
+
 function pick_primary(pool_id, pg_num, pool_config, osd_set, up_osds, aff_osds)
 {
     let alive_set;
@@ -44,6 +50,7 @@ function pick_primary(pool_id, pg_num, pool_config, osd_set, up_osds, aff_osds)
 function recheck_primary(state, global_config, up_osds, osd_tree)
 {
     let new_pg_config;
+    let held = false;
     for (const pool_id in state.config.pools)
     {
         const pool_cfg = state.config.pools[pool_id];
@@ -62,7 +69,16 @@ function recheck_primary(state, global_config, up_osds, osd_tree)
             const pg_cfg = state.pg.config.items[pool_id][pg_num];
             if (pg_cfg)
             {
-                const new_primary = pick_primary(pool_id, pg_num, state.config.pools[pool_id], pg_cfg.osd_set, up_osds, aff_osds);
+                const pg_state = (state.pg.state[pool_id]||{})[pg_num]||{};
+                let new_primary = pg_state.primary;
+                if (global_config.mon_hold_unclean_pg_primary && !is_clean(pg_state.state) && new_primary && up_osds[new_primary])
+                {
+                    held = true;
+                }
+                else
+                {
+                    new_primary = pick_primary(pool_id, pg_num, state.config.pools[pool_id], pg_cfg.osd_set, up_osds, aff_osds);
+                }
                 if (pg_cfg.primary != new_primary)
                 {
                     if (!new_pg_config)
@@ -91,19 +107,24 @@ function recheck_primary(state, global_config, up_osds, osd_tree)
             }
         }
     }
-    return new_pg_config;
+    return { new_pg_config, held };
 }
 
-function save_new_pgs_txn(save_to, request, state, etcd_prefix, etcd_watch_revision, pool_id, up_osds, osd_tree, prev_pgs, new_pgs, pg_history)
+function save_new_pgs_txn(save_to, request, state, config, etcd_watch_revision, pool_id, up_osds, osd_tree, prev_pgs, new_pgs, pg_history)
 {
     const aff_osds = get_affinity_osds(state.config.pools[pool_id] || {}, up_osds, osd_tree);
     const pg_items = {};
+    const cur_pgs = config.mon_hold_unclean_pg_primary && prev_pgs.length == new_pgs.length && state.pg.state[pool_id];
     new_pgs.map((osd_set, i) =>
     {
         osd_set = osd_set.map(osd_num => osd_num === LPOptimizer.NO_OSD ? 0 : osd_num);
+        const hold = (cur_pgs && cur_pgs[i+1] &&
+            cur_pgs[i+1].state && !is_clean(cur_pgs[i+1].state) &&
+            cur_pgs[i+1].primary && up_osds[cur_pgs[i+1].primary] &&
+            osd_set.filter(c => c == cur_pgs[i+1].primary).length > 0);
         pg_items[i+1] = {
             osd_set,
-            primary: pick_primary(pool_id, i+1, state.config.pools[pool_id], osd_set, up_osds, aff_osds),
+            primary: hold ? cur_pgs[i+1].primary : pick_primary(pool_id, i+1, state.config.pools[pool_id], osd_set, up_osds, aff_osds),
         };
         if (prev_pgs[i] && prev_pgs[i].join(' ') != osd_set.join(' ') &&
             prev_pgs[i].filter(osd_num => osd_num).length > 0)
@@ -123,7 +144,7 @@ function save_new_pgs_txn(save_to, request, state, etcd_prefix, etcd_watch_revis
         // FIXME: etcd has max_txn_ops limit, and it's 128 by default
         // Sooo we probably want to change our storage scheme for PG histories...
         request.compare.push({
-            key: b64(etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
+            key: b64(config.etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
             target: 'MOD',
             mod_revision: ''+etcd_watch_revision,
             result: 'LESS',
@@ -132,7 +153,7 @@ function save_new_pgs_txn(save_to, request, state, etcd_prefix, etcd_watch_revis
         {
             request.success.push({
                 requestPut: {
-                    key: b64(etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
+                    key: b64(config.etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
                     value: b64(JSON.stringify(pg_history[i])),
                 },
             });
@@ -141,7 +162,7 @@ function save_new_pgs_txn(save_to, request, state, etcd_prefix, etcd_watch_revis
         {
             request.success.push({
                 requestDeleteRange: {
-                    key: b64(etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
+                    key: b64(config.etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
                 },
             });
         }

mon/tests/primary-stickiness.js

@@ -0,0 +1,158 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 (see README.md for details)
+
+// Primary OSD selection must not move a PG away from a live primary while the
+// PG is unclean. Run: node mon/test-primary-stickiness.js
+
+const assert = require('assert');
+
+const { recheck_primary, save_new_pgs_txn } = require('../pg_gen.js');
+const LPOptimizer = require('../lp_optimizer/lp_optimizer.js');
+
+const HOSTS = 6, OSDS_PER_HOST = 16;
+const host_osds = (h) => Array.from({ length: OSDS_PER_HOST }, (_, i) => h*OSDS_PER_HOST+i+1);
+
+// EC 4+2 over 6 hosts, one OSD of each host per PG; every PG active+clean on its balanced primary
+function make_state(pg_count)
+{
+    const state = {
+        config: { pools: { 1: { name: 'ec', scheme: 'ec', pg_size: 6, parity_chunks: 2, pg_minsize: 4,
+            pg_count, failure_domain: 'host' } } },
+        pg: { config: { items: { 1: {} } }, state: { 1: {} } },
+    };
+    for (let pg = 1; pg <= pg_count; pg++)
+    {
+        const osd_set = [];
+        for (let h = 0; h < HOSTS; h++)
+            osd_set.push(host_osds((h + pg) % HOSTS)[(pg * 7 + h) % OSDS_PER_HOST]);
+        state.pg.config.items[1][pg] = { osd_set, primary: 0 };
+    }
+    const all_up = up_except([]);
+    const moved = recheck_primary(state, { mon_hold_unclean_pg_primary: true }, all_up, {}).new_pg_config;
+    state.pg.config = moved;
+    for (let pg = 1; pg <= pg_count; pg++)
+        state.pg.state[1][pg] = { primary: state.pg.config.items[1][pg].primary, state: [ 'active' ] };
+    return state;
+}
+
+function up_except(down)
+{
+    const up = {};
+    for (let h = 0; h < HOSTS; h++)
+        for (const osd of host_osds(h))
+            up[osd] = true;
+    for (const osd of down)
+        delete up[osd];
+    return up;
+}
+
+// OSDs report every PG that lost a member as degraded+has_unclean
+function mark_unclean(state, up_osds)
+{
+    for (const pg in state.pg.config.items[1])
+    {
+        const cfg = state.pg.config.items[1][pg];
+        if (cfg.osd_set.some(osd => !up_osds[osd]))
+            state.pg.state[1][pg] = { primary: cfg.primary, state: [ 'active', 'degraded', 'has_unclean' ] };
+    }
+}
+
+function moves(state, new_cfg)
+{
+    const res = [];
+    for (const pg in (new_cfg||{items:{1:{}}}).items[1])
+    {
+        const from = state.pg.config.items[1][pg].primary, to = new_cfg.items[1][pg].primary;
+        if (from != to)
+            res.push({ pg, from, to });
+    }
+    return res;
+}
+
+function test_node_failure_moves_only_dead_primaries()
+{
+    const state = make_state(1024);
+    const down = host_osds(4);
+    const up_osds = up_except(down);
+    mark_unclean(state, up_osds);
+    const recheck = recheck_primary(state, { mon_hold_unclean_pg_primary: true }, up_osds, {});
+    const mv = moves(state, recheck.new_pg_config);
+    const from_live = mv.filter(m => up_osds[m.from]);
+    console.log(`node down: ${mv.length} moves, ${from_live.length} away from a live primary, held = ${recheck.held}`);
+    assert(mv.length > 0, 'PGs whose primary died must move');
+    assert.equal(from_live.length, 0, 'a live primary must stay while its PG is unclean');
+    assert(mv.every(m => up_osds[m.to]));
+    assert(recheck.held);
+}
+
+// clean_state: a state the monitor counts as clean (see Mon.save_last_clean())
+function test_no_flap_back_when_osd_returns(clean_state)
+{
+    const state = make_state(256);
+    const pg = Object.keys(state.pg.config.items[1])[0];
+    const cfg = state.pg.config.items[1][pg];
+    const old_primary = cfg.primary;
+    // The primary crashes: the PG moves away from it
+    let up_osds = up_except([ old_primary ]);
+    mark_unclean(state, up_osds);
+    const after_crash = recheck_primary(state, { mon_hold_unclean_pg_primary: true }, up_osds, {}).new_pg_config;
+    const new_primary = after_crash.items[1][pg].primary;
+    assert.notEqual(new_primary, old_primary);
+    state.pg.config = after_crash;
+    state.pg.state[1][pg] = { primary: new_primary, state: [ 'active', 'degraded', 'has_unclean' ] };
+    // It restarts 2 s later while the PG is still unclean: no move back
+    up_osds = up_except([]);
+    const back = recheck_primary(state, { mon_hold_unclean_pg_primary: true }, up_osds, {}).new_pg_config;
+    assert(!back || back.items[1][pg].primary == new_primary, 'no move back while the PG is unclean');
+    // Once the PGs are clean on their new primaries, they return to the balanced ones
+    for (const p in state.pg.config.items[1])
+        state.pg.state[1][p] = { primary: state.pg.config.items[1][p].primary, state: clean_state };
+    const clean = recheck_primary(state, { mon_hold_unclean_pg_primary: true }, up_osds, {});
+    assert(clean.new_pg_config.items[1][pg].primary == old_primary);
+    assert(!clean.held);
+    console.log(`no flap back, back once ${clean_state.join('+')}: ok`);
+}
+
+function test_reoptimize_keeps_live_primary()
+{
+    const state = make_state(64);
+    const up_osds = up_except(host_osds(4));
+    mark_unclean(state, up_osds);
+    const prev_pgs = Object.values(state.pg.config.items[1]).map(c => c.osd_set);
+    const save_to = {}, request = { compare: [], success: [] };
+    save_new_pgs_txn(save_to, request, state, { etcd_prefix: '/vitastor', mon_hold_unclean_pg_primary: true },
+        1, 1, up_osds, {}, prev_pgs, prev_pgs.map(s => [ ...s ]), []);
+    const mv = moves(state, save_to);
+    assert(mv.length > 0, 'PGs whose primary died must move');
+    console.log(save_to);
+    console.log(mv);
+    assert.equal(mv.filter(m => up_osds[m.from]).length, 0, 're-optimisation must not move a live primary');
+    console.log(`re-optimize: ${mv.length} moves, all from dead primaries`);
+}
+
+// A clean PG with its whole current set up may move even if its new set has an empty slot
+function test_reoptimize_judges_current_set()
+{
+    const state = make_state(64);
+    const up_osds = up_except([]);
+    const prev_pgs = Object.values(state.pg.config.items[1]).map(c => c.osd_set);
+    const new_pgs = prev_pgs.map(s => s.map((osd, i) => i == 1 ? LPOptimizer.NO_OSD : osd));
+    const save_to = {}, request = { compare: [], success: [] };
+    save_new_pgs_txn(save_to, request, state, { etcd_prefix: '/vitastor', mon_hold_unclean_pg_primary: true },
+        1, 1, up_osds, {}, prev_pgs, new_pgs, []);
+    const kept_in_set = moves(state, save_to).filter(m => save_to.items[1][m.pg].osd_set.includes(m.from));
+    assert(kept_in_set.length > 0, 'a clean PG must not be held because of its new set');
+    console.log(`re-optimize, clean PGs: ${kept_in_set.length} rebalanced`);
+}
+
+async function run()
+{
+    test_node_failure_moves_only_dead_primaries();
+    test_no_flap_back_when_osd_returns([ 'active' ]);
+    test_no_flap_back_when_osd_returns([ 'active', 'left_on_dead' ]);
+    test_reoptimize_keeps_live_primary();
+    test_reoptimize_judges_current_set();
+    console.log('OK');
+}
+
+run().catch(e => { console.error(e); process.exit(1); });

src/blockstore/blockstore_heap.cpp

@@ -1187,16 +1187,12 @@ bool blockstore_heap_t::recheck_small_writes(std::function<void(bool is_data, ui
         {
             // Check this object synchronously
             heap_entry_t *bad_wr = NULL, *oldest_wr = NULL;
-            int checked = 0;
             auto wr = skip_commits(obj);
-            for (; wr && wr->type() == BS_HEAP_SMALL_WRITE && checked < recheck_count; wr = skip_commits(prev(wr)))
+            for (int i = 0; i < recheck_count; i++, wr = skip_commits(prev(wr)))
             {
-                if (!wr->small().len)
-                    continue;
                 fprintf(stderr, "Notice: rechecking %jx:%jx l%ju - %u bytes at %ju in buffer area\n",
                     wr->inode, wr->stripe, wr->lsn, wr->small().len, wr->small().location);
                 oldest_wr = wr;
-                checked++;
                 if (!calc_checksums(wr, buffer_area + wr->small().location, false))
                     bad_wr = wr;
             }

src/blockstore/blockstore_write.cpp

@@ -184,8 +184,7 @@ enospc:
         data->callback = [this, op](ring_data_t *data) { handle_write_event(data, op); };
         assert(loc+op->offset+op->len <= dsk.block_count*dsk.data_block_size);
         io_uring_prep_writev(sqe, dsk.data_fd, &data->iov, 1, dsk.data_offset + loc + op->offset);
-        // redirect_intent writes skip the data fsync (resume_2), so they must not take a slot
-        if (!dsk.disable_data_fsync && PRIV(op)->write_type == BS_HEAP_BIG_WRITE)
+        if (!dsk.disable_data_fsync)
         {
             // use PRIV->lsn for fsync_data_id
             PRIV(op)->lsn = ++data_fsync_next;

src/client/etcd_state_client_mock.cpp

@@ -322,18 +322,8 @@ void etcd_state_client_mock_t::mock_apply_changes(std::map<std::string, etcd_kv_
 {
     for (auto & kv: changes)
         parse_state(kv.second);
-    // A real client can't react to any change before its load finished: the
-    // watch starts after the range read. Stash changes until then, otherwise
-    // the OSD applies half-loaded configs (PGs start with epoch 0 and write a
-    // reset epoch back into etcd, clobbering the real one)
-    for (auto & kv: changes)
-        pending_changes[kv.first] = kv.second;
-    if (!pgs_loaded)
-        return;
-    auto pending = std::move(pending_changes);
-    pending_changes.clear();
-    if (pending.size() && on_change_hook != NULL)
-        on_change_hook(pending);
+    if (on_change_hook != NULL)
+        on_change_hook(changes);
 }
 
 void etcd_state_client_mock_t::load_global_config()
@@ -343,18 +333,5 @@ void etcd_state_client_mock_t::load_global_config()
 
 void etcd_state_client_mock_t::load_pgs()
 {
-    etcd_state_client_t::load_pgs([this](const std::string & err)
-    {
-        pgs_loaded = true;
-        // The range read already contains everything up to the load revision;
-        // replay only the newer changes, like a watch would from that revision
-        auto pending = std::move(pending_changes);
-        pending_changes.clear();
-        std::map<std::string, etcd_kv_t> fresh;
-        for (auto & kv: pending)
-            if (kv.second.mod_revision >= etcd_watch_revision_pg)
-                fresh[kv.first] = kv.second;
-        if (fresh.size() && on_change_hook != NULL)
-            on_change_hook(fresh);
-    });
+    etcd_state_client_t::load_pgs([this](const std::string & err) {});
 }

src/client/etcd_state_client_mock.h

@@ -62,10 +62,6 @@ struct etcd_state_client_mock_t: public etcd_state_client_t
     // The client doesn't own the store; the test does
     etcd_mock_t *store = NULL;
     bool paused = false;
-    // A real client starts receiving watch events only after its load finished;
-    // until then changes are stashed here and flushed on load completion
-    bool pgs_loaded = false;
-    std::map<std::string, etcd_kv_t> pending_changes;
     std::vector<etcd_mock_request_t> queue;
     std::string username;
     etcd_state_client_mock_t(etcd_mock_t & store);

src/cmd/cli.cpp

@@ -108,8 +108,8 @@ static const char* help_text =
     "  --writers-stopped allows optimised removal in case of a single 'slim' read-write\n"
     "  child and 'fat' removed parent: the child is merged into parent and parent is renamed\n"
     "  to child in that case. In other cases parent layers are always merged into children.\n"
-    "  --exact            Remove multiple images with given names.\n"
-    "  --matching         Remove multiple images with names matching given glob patterns.\n"
+    "  --exact            Remove multiple images with names matching given glob patterns.\n"
+    "  --matching         Remove multiple images with given names\n"
     "  --writers-stopped  Allow renaming inodes over their read/write children.\n"
     "  --down-ok          Continue deletion/merging even if some data will be left on unavailable OSDs.\n"
     "\n"
@@ -145,15 +145,12 @@ static const char* help_text =
     "  --min-offset, --max-offset\n"
     "      Restrict listing to specific offsets inside inodes.\n"
     "\n"
-    "vitastor-cli raw-ls --pool_id ID (--pg_num NUM|--offset NUM) [OPTIONS]\n"
+    "vitastor-cli raw-ls [OPTIONS]\n"
     "  Find object(s) in the cluster using raw secondary listing operations. Options:\n"
     "  [--min_inode NUM] [--max_inode NUM] [--offset NUM] [--pg_num NUM] [--pg_count COUNT]\n"
     "  [--pg_stripe_size NUM] [--osds 1,2,3,...]\n"
-    "  --pool_id (numeric pool ID) is required. --pg_num is required unless --offset is given,\n"
-    "  in which case the PG is calculated from the offset. --pg_count, --pg_stripe_size and\n"
-    "  --osds default to the pool configuration and are required if the pool does not exist.\n"
     "\n"
-    "vitastor-cli fix [--objects <objects>] --bad-osds <osds> [--part <part>] [--check no]\n"
+    "vitastor-cli fix [--objects <objects>] [--bad-osds <osds>] [--part <part>] [--check no]\n"
     "  Fix inconsistent objects in the cluster by deleting some copies.\n"
     "  --objects <objects>\n"
     "      Objects to fix, either in plain text or JSON format. If not specified,\n"
@@ -161,7 +158,7 @@ static const char* help_text =
     "      Plain text format: 0x<inode>:0x<stripe> <any delimiter> 0x<inode>:0x<stripe> ...\n"
     "      JSON format: [{\"inode\":\"0x...\",\"stripe\":\"0x...\"},...]\n"
     "  --bad-osds <osds>\n"
-    "      Required. Remove inconsistent copies/parts of objects from these OSDs, effectively\n"
+    "      Remove inconsistent copies/parts of objects from these OSDs, effectively\n"
     "      marking them bad and allowing Vitastor to recover objects from other copies.\n"
     "  --part <number>\n"
     "      Only remove EC part <number> (from 0 to pg_size-1), required for extreme\n"
@@ -173,14 +170,11 @@ static const char* help_text =
     "vitastor-cli alloc-osd\n"
     "  Allocate a new OSD number and reserve it by creating empty /osd/stats/<n> key.\n"
     "\n"
-    "vitastor-cli rm-osd [-f|--force] [--allow-data-loss] [--allow-up] [--dry-run] <osd_id> [osd_id...]\n"
+    "vitastor-cli rm-osd [--force] [--allow-data-loss] [--dry-run] <osd_id> [osd_id...]\n"
     "  Remove metadata and configuration for specified OSD(s) from etcd.\n"
-    "  Refuses to remove OSDs that are up or whose removal makes PGs degraded, offline or incomplete.\n"
-    "  -f|--force         Remove OSDs even if they are up or PGs become degraded or offline.\n"
-    "  --allow-data-loss  Remove OSDs even if PGs or objects become incomplete (DATA LOSS). Implies --force.\n"
-    "  --allow-up         Remove OSDs even if they are up.\n"
-    "  --dry-run          Only check if deletion is possible without data loss and\n"
-    "                     redundancy degradation.\n"
+    "  Refuses to remove OSDs with data without --force and --allow-data-loss.\n"
+    "  With --dry-run only checks if deletion is possible without data loss and\n"
+    "  redundancy degradation.\n"
     "\n"
     "vitastor-cli osd-tree [-l|--long]\n"
     "  Show current OSD tree, optionally with I/O statistics if -l is specified.\n"

src/test/test_blockstore.cpp

@@ -1406,124 +1406,6 @@ static void test_recheck_under_commit()
     free(buf);
 }
 
-// Zero-sized version bumps carry no data, so the startup recheck skips them - but they must not
-// consume the budget it has for the object's buffered writes. Otherwise the oldest small write
-// falls out of the check: when a volatile disk cache loses its data while the heap entry reaches
-// the platter, the entry survives recovery and the object returns the previous tenant's bytes
-// from the journal location
-static void test_recheck_under_bumps()
-{
-    printf("\n-- test_recheck_under_bumps\n");
-
-    bs_test_t test;
-    test.default_cfg();
-    // No block checksums: only the single data checksum every small write entry carries
-    // can catch the lost journal data
-    test.config["data_csum_type"] = "none";
-    test.config["csum_block_size"] = "0";
-    // A volatile write cache is what makes the data of an acknowledged write disappear
-    test.config["disable_data_fsync"] = "0";
-    test.config["immediate_commit"] = "none";
-    test.init();
-
-    object_id oid = { .inode = 1, .stripe = 0 };
-    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 128*1024);
-
-    blockstore_op_t op;
-    op.oid = oid;
-    op.buf = buf;
-
-    auto sync = [&]()
-    {
-        blockstore_op_t s;
-        s.opcode = BS_OP_SYNC;
-        test.exec_op(&s);
-        assert(s.retval == 0);
-    };
-    auto stabilize = [&](uint64_t version)
-    {
-        blockstore_op_t s;
-        s.opcode = BS_OP_STABLE;
-        s.len = 1;
-        s.buf = (uint8_t*)malloc_or_die(sizeof(obj_ver_id));
-        *((obj_ver_id*)s.buf) = (obj_ver_id){ .oid = oid, .version = version };
-        test.exec_op(&s);
-        assert(s.retval == 0);
-        free(s.buf);
-    };
-
-    printf("writing v1 0+128K, syncing and stabilizing it\n");
-    memset(buf, 0xaa, 128*1024);
-    op.opcode = BS_OP_WRITE;
-    op.version = 1;
-    op.offset = 0;
-    op.len = 128*1024;
-    test.exec_op(&op);
-    assert(op.retval == (int)op.len);
-    sync();
-    stabilize(1);
-
-    // Swallow the data of the next write: the disk acknowledges it from its volatile cache
-    // and then loses it, while the metadata entry does reach the platter
-    uint64_t buf_start = test.bs->dsk.journal_offset;
-    uint64_t buf_end = buf_start + test.bs->dsk.journal_len;
-    bool swallowed = false;
-    test.sqe_handler = [&](io_uring_sqe *sqe)
-    {
-        if (sqe->opcode == IORING_OP_WRITEV && sqe->fd == MOCK_DATA_FD &&
-            sqe->off >= buf_start && sqe->off < buf_end)
-        {
-            auto *data = (ring_data_t*)sqe->user_data;
-            data->res = data->iov.iov_len;
-            test.ringloop->mark_completed(data);
-            swallowed = true;
-            return true;
-        }
-        return false;
-    };
-
-    printf("writing v2 8192+4096, losing its data in the disk cache\n");
-    memset(buf, 0xcc, 4096);
-    op.version = 2;
-    op.offset = 8192;
-    op.len = 4096;
-    test.exec_op(&op);
-    assert(op.retval == (int)op.len);
-    assert(swallowed);
-    test.sqe_handler = nullptr;
-
-    printf("writing zero-sized v3 and v4 on top of v2\n");
-    op.version = 3;
-    op.offset = 8192;
-    op.len = 0;
-    test.exec_op(&op);
-    assert(op.retval == 0);
-    op.version = 4;
-    test.exec_op(&op);
-    assert(op.retval == 0);
-    sync();
-
-    printf("restarting the blockstore\n");
-    test.destroy_bs();
-    test.ringloop->reset();
-    test.init();
-
-    printf("v2 must be rolled back along with the bumps above it, leaving v1\n");
-    blockstore_op_t rd;
-    rd.opcode = BS_OP_READ;
-    rd.oid = oid;
-    rd.version = UINT64_MAX;
-    rd.offset = 0;
-    rd.len = 128*1024;
-    rd.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, rd.len);
-    test.exec_op(&rd);
-    assert(rd.retval == (int)rd.len);
-    assert(rd.version == 1);
-    assert(memcheck(rd.buf, 0xaa, 128*1024));
-    free(rd.buf);
-    free(buf);
-}
-
 // Compaction merges newer writes into the base data block in place, so after a crash the base
 // entry's checksums no longer match the block. The startup recheck may only skip the parts which
 // newer entries overwrite when one checksum block covers exactly one write granule. With a larger
@@ -2122,52 +2004,6 @@ static void test_no_intent_over_unwritten_compaction()
     free(op2.buf);
 }
 
-// A redirect_intent write must not leave its data fsync slot behind,
-// or every later big_write waits for that slot forever
-static void test_big_write_after_redirect_intent()
-{
-    printf("\n-- test_big_write_after_redirect_intent\n");
-
-    bs_test_t test;
-    test.default_cfg();
-    test.config["disable_data_fsync"] = "0";
-    test.config["immediate_commit"] = "none";
-    test.init();
-
-    // Stable write to a new object: redirect_intent
-    blockstore_op_t op;
-    op.opcode = BS_OP_WRITE_STABLE;
-    op.oid = { .inode = 1, .stripe = 0 };
-    op.version = 1;
-    op.offset = 0;
-    op.len = 16384;
-    op.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op.len);
-    memset(op.buf, 0xaa, op.len);
-    test.exec_op(&op);
-    assert(op.retval == op.len);
-    assert(!test.bs->data_fsyncs.size());
-
-    // Unstable write to a new object: big_write
-    bool done = false;
-    blockstore_op_t op2;
-    op2.opcode = BS_OP_WRITE;
-    op2.oid = { .inode = 2, .stripe = 0 };
-    op2.version = 1;
-    op2.offset = 0;
-    op2.len = 16384;
-    op2.buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, op2.len);
-    memset(op2.buf, 0xbb, op2.len);
-    op2.callback = [&](blockstore_op_t *op) { done = true; };
-    test.bs->enqueue_op(&op2);
-    for (int iters = 0; !done && iters < 1000; iters++)
-        test.ringloop->loop();
-    assert(done);
-    assert(op2.retval == op2.len);
-
-    free(op.buf);
-    free(op2.buf);
-}
-
 int main(int narg, char *args[])
 {
     test_simple();
@@ -2193,11 +2029,9 @@ int main(int narg, char *args[])
     test_compact_big_intent_survives_crash(16384);
     test_compact_big_intent_survives_crash(0);
     test_recheck_under_commit();
-    test_recheck_under_bumps();
     test_compact_over_rolled_back_big_write();
     test_zero_write_over_rollback_to_zero();
     test_compact_intent_write_in_partial_csum_block();
     test_no_intent_over_unwritten_compaction();
-    test_big_write_after_redirect_intent();
     return 0;
 }

src/test/test_pg_chaos.cpp

@@ -480,8 +480,6 @@ struct wire_op_t
     uint8_t *attr;
 };
 
-static void free_wire_op(wire_op_t & w);
-
 // The fleet of fake secondary OSDs: who exists, who has a live messenger
 // connection and which queued op gets answered on this round. The answering
 // itself is the peers' own behavior, see fake_peer_t::answer(); ch gives the
@@ -490,14 +488,6 @@ struct fake_cluster_t
 {
     chaos_t & ch;
     explicit fake_cluster_t(chaos_t & owner) : ch(owner) {}
-    ~fake_cluster_t()
-    {
-        // Stragglers still queued when the run ends have no instance left to
-        // apply them; release their bytes
-        for (auto & w: wire)
-            free_wire_op(w);
-        wire.clear();
-    }
     // Fake replicas including the spare, by OSD number. Survive OSD restarts -
     // an OSD crash doesn't touch the peers
     std::map<osd_num_t, fake_peer_t> peers;
@@ -1600,83 +1590,32 @@ void chaos_t::write_pg_history(const std::vector<osd_num_t> & old_set)
 {
     std::vector<osd_num_t> old_sorted = old_set;
     std::sort(old_sorted.begin(), old_sorted.end());
-    // The mon has no state except etcd itself: the OSD instance's st_cli view
-    // may be empty (fresh restart, load not finished), so epoch and sets are
-    // read from the store. The OSD rewrites this key too, so - like the real
-    // mon - commit with CAS on mod_revision and retry on conflict
-    std::string hist_key = "/vitastor/pg/history/"+std::to_string(POOL_ID)+"/"+std::to_string(PG_NUM);
-    for (int attempt = 0; ; attempt++)
-    {
-        auto stored = f.etcd_store.data.find(hist_key);
-        uint64_t cur_epoch = 0, next_scrub = 0, cur_rev = 0;
-        std::vector<std::vector<osd_num_t>> hist;
-        if (stored != f.etcd_store.data.end())
-        {
-            std::string err;
-            auto hist_json = json11::Json::parse(stored->second.value, err);
-            if (err != "")
-            {
-                fprintf(stderr, "can't parse pg history in etcd: %s\n", err.c_str());
-                abort();
-            }
-            cur_epoch = hist_json["epoch"].uint64_value();
-            next_scrub = hist_json["next_scrub"].uint64_value();
-            cur_rev = stored->second.mod_revision;
-            for (auto & s: hist_json["osd_sets"].array_items())
-            {
-                std::vector<osd_num_t> hs;
-                for (auto & o: s.array_items())
-                    hs.push_back((osd_num_t)o.uint64_value());
-                hist.push_back(hs);
-            }
-        }
-        auto it = std::lower_bound(hist.begin(), hist.end(), old_sorted);
-        if (it == hist.end() || *it != old_sorted)
-            hist.insert(it, old_sorted);
-        std::set<osd_num_t> all_peers(cur_set.begin(), cur_set.end());
-        for (auto & hs: hist)
-        {
-            for (osd_num_t o: hs)
-                all_peers.insert(o);
-        }
-        json11::Json::array sets_arr;
-        for (auto & hs: hist)
-            sets_arr.push_back(json11::Json(hs));
-        json11::Json::object value = {
-            { "epoch", (double)(cur_epoch+1) },
-            { "all_peers", json11::Json(std::vector<osd_num_t>(all_peers.begin(), all_peers.end())) },
-            { "osd_sets", sets_arr },
-        };
-        if (next_scrub)
-            value["next_scrub"] = (double)next_scrub;
-        json11::Json txn = json11::Json::object {
-            { "compare", json11::Json::array { json11::Json::object {
-                { "target", "MOD" },
-                { "key", base64_encode(hist_key) },
-                { "mod_revision", (double)(cur_rev ? cur_rev : 1) },
-                { "result", cur_rev ? "" : "LESS" },
-            } } },
-            { "success", json11::Json::array { json11::Json::object {
-                { "request_put", json11::Json::object {
-                    { "key", base64_encode(hist_key) },
-                    { "value", base64_encode(json11::Json(value).dump()) },
-                } } } } },
-        };
-        // The mock mutates the store synchronously inside the call, so the
-        // revision bump of the key tells whether the CAS won
-        f.etcd_store.etcd_call("/kv/txn", txn, [](std::string, json11::Json) {});
-        auto now = f.etcd_store.data.find(hist_key);
-        if (now != f.etcd_store.data.end() && now->second.mod_revision > cur_rev)
-            break;
-        if (attempt >= 100)
-        {
-            fprintf(stderr, "pg history CAS never won\n");
-            abort();
-        }
-    }
+    auto & pg_cfg = f.st_cli->pool_config[POOL_ID].pg_config[PG_NUM];
     // Mirror what a real OSD would report first: the PG is taken by us, so
     // our own last report (with cur_primary set) is already in the store
-    f.st_cli->pool_config[POOL_ID].pg_config[PG_NUM].cur_primary = 1;
+    pg_cfg.cur_primary = 1;
+    auto & hist = pg_cfg.target_history;
+    auto it = std::lower_bound(hist.begin(), hist.end(), old_sorted);
+    if (it == hist.end() || *it != old_sorted)
+        hist.insert(it, old_sorted);
+    std::set<osd_num_t> all_peers(cur_set.begin(), cur_set.end());
+    for (auto & hs: hist)
+    {
+        for (osd_num_t o: hs)
+            all_peers.insert(o);
+    }
+    json11::Json::array sets_arr;
+    for (auto & hs: hist)
+        sets_arr.push_back(json11::Json(hs));
+    json11::Json::object value = {
+        { "epoch", (double)(pg_cfg.epoch + 1) },
+        { "all_peers", json11::Json(std::vector<osd_num_t>(all_peers.begin(), all_peers.end())) },
+        { "osd_sets", sets_arr },
+    };
+    if (pg_cfg.next_scrub)
+        value["next_scrub"] = (double)pg_cfg.next_scrub;
+    f.st_cli->set("/vitastor/pg/history/"+std::to_string(POOL_ID)+"/"+std::to_string(PG_NUM),
+        json11::Json(value));
 }
 
 void chaos_t::connect_session()
@@ -1729,14 +1668,10 @@ void chaos_t::session_died()
     // Fold the acks the client really received before breaking the promise:
     // account them as ACKED and demote_unstable() then turns exactly those into
     // OPTIONAL - production order (the ack reached the client, THEN the
-    // connection died and the repeer rolls the version back).  This is the
-    // only demoting event in the model: the client reacts to its dead session,
-    // never to a server-side crash it cannot see.  With immediate_commit=all
-    // an ack is journaled durable, survives the repeer, so nothing demotes.
-    // Account only, never erase here: predicate rounds of run_until() judge on
-    // inflight and a done op hidden from them looks like an op that never
-    // completes; the sole eraser is reap(), which runs at step() entry and
-    // drain boundaries
+    // connection died and the repeer rolls the version back).  Account only,
+    // never erase here: predicate rounds of run_until() judge on inflight and
+    // a done op hidden from them looks like an op that never completes; the
+    // sole eraser is reap(), which runs at step() entry and drain boundaries
     for (auto *p: inflight)
         if (p->done)
             account(p);
@@ -2514,13 +2449,26 @@ void chaos_t::do_crash()
     // delete lands. Its OPTIONAL version above the last clean delete is
     // itself the open window that arms the repeat (redel_pending) - there is
     // nothing to carry over
-    // A crash itself demotes nothing: demotion follows the client's view, not
-    // the server's misfortune.  With immediate_commit=all every ack is durable,
-    // so a crash takes nothing that was acknowledged.  With a volatile cache the
-    // crash also kills the client session, and session_died() - the only
-    // demoting event - folds the lost acks into OPTIONAL the moment the model
-    // observes the dead session; a session that somehow survives keeps its
-    // promises
+    // Acked-uncommitted versions are at the replay's mercy only with a volatile
+    // cache: with immediate_commit=all every ack is journaled durable, a crash
+    // takes nothing that was acknowledged, and those versions keep their
+    // promise through the outage. Demote them once and for all - a later SYNC
+    // must not make a lost version forceful. If some chain still carries their
+    // bytes, world_check explains that world through the claim path; if the
+    // bytes are gone, nothing may demand them
+    if (use_fsync)
+    {
+        for (int obj = 0; obj < OBJ_COUNT; obj++)
+        {
+            for (auto & kv: objs[obj].vers)
+            {
+                if (kv.second.state == V_ACKED)
+                    kv.second.state = V_OPTIONAL;
+                // (SUBMITTED is skipped: its op hasn't landed; account()
+                // resolves it from the reply)
+            }
+        }
+    }
     // From here on the pre-crash acks are unverifiable: a SYNC on the fresh OSD
     // session proves nothing about them, so promotions freeze until the recovery
     // reads below have reconciled every object
@@ -2916,10 +2864,9 @@ void chaos_t::observe(int obj, const std::vector<uint8_t> & got, uint64_t max_se
     static const payload_t zero = {};
     // The strict verdict: the answer is entirely ACKED/SYNCED content. A chain
     // that legitimately lags behind the acknowledged one was rolled back by a
-    // crash or a repeer - which kills the sessions owning acked-unstable
-    // versions, and their death demotes the ACKED versions to OPTIONAL - so an
-    // ACKED/SYNCED la with only ACKED/SYNCED bytes shown can only come from the
-    // acknowledged chain itself: every granule la wrote
+    // crash or a repeer, and every such event demotes the ACKED versions to
+    // OPTIONAL - so an ACKED/SYNCED la with only ACKED/SYNCED bytes shown can
+    // only come from the acknowledged chain itself: every granule la wrote
     // must show it, anything else is committed data lost. A delete
     // acknowledged after the last write excuses absence; a delete still in
     // flight may already have applied on the primary and legitimately answers
@@ -3067,10 +3014,10 @@ void chaos_t::observe(int obj, const std::vector<uint8_t> & got, uint64_t max_se
                 abort();
             }
             // An ACKED version was established by an ack or a positive
-            // observation, and only a dead client session can undo it - the
-            // crash or repeer behind the death demotes the ACKED versions to
-            // OPTIONAL through session_died() before anything reads from the
-            // re-settled chain. Covered absent over its range while
+            // observation, and only an event can undo it - and every event
+            // (a crash over the volatile cache, a repeer rollback, a session
+            // kill) demotes ACKED versions to OPTIONAL before anything reads
+            // from the re-settled chain. Covered absent over its range while
             // still ACKED means established bytes vanished with no event in
             // between: a non-repeatable read
             if (d.state == V_ACKED)