Fix vitastor-cli ls --tree hiding layers whose parent is filtered out

ls --tree built the tree starting from the roots of the selected layers,
so a layer whose parent was left out by a glob, -p or -n was never
reached and not printed at all: ls --tree testimg printed nothing when
testimg had a parent. Show such a layer as a root; its parent is still
shown in the PARENT column.

mon/etcd_schema.js

@@ -65,7 +65,6 @@ 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,7 +74,6 @@ 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;
@@ -200,7 +199,6 @@ 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)
@@ -254,7 +252,7 @@ class Mon
         {
             this.schedule_update_stats();
         }
-        if (changed || this.primaries_held && pg_states_changed)
+        if (changed)
         {
             this.schedule_recheck();
         }
@@ -566,8 +564,7 @@ class Mon
         else
         {
             // Nothing changed, but we still want to recheck the distribution of primaries
-            let { new_pg_config, held } = recheck_primary(this.state, this.config, up_osds, osd_tree, held);
-            this.primaries_held = held;
+            let new_pg_config = recheck_primary(this.state, this.config, up_osds, osd_tree);
             if (new_pg_config)
             {
                 const ok = await this.save_pg_config(new_pg_config);
@@ -629,7 +626,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,
+                save_new_pgs_txn(new_pg_config, etcd_request, this.state, this.config.etcd_prefix,
                     this.etcd_watch_revision, pool_id, up_osds, osd_tree, prev_pgs, [], []);
             }
         }
@@ -672,7 +669,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,
+            save_new_pgs_txn(new_pg_config, etcd_request, this.state, this.config.etcd_prefix,
                 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,12 +10,6 @@ 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;
@@ -50,7 +44,6 @@ 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];
@@ -69,16 +62,7 @@ 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 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);
-                }
+                const 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)
@@ -107,24 +91,19 @@ function recheck_primary(state, global_config, up_osds, osd_tree)
             }
         }
     }
-    return { new_pg_config, held };
+    return new_pg_config;
 }
 
-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)
+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)
 {
     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: hold ? cur_pgs[i+1].primary : pick_primary(pool_id, i+1, state.config.pools[pool_id], osd_set, up_osds, aff_osds),
+            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)
@@ -144,7 +123,7 @@ function save_new_pgs_txn(save_to, request, state, config, etcd_watch_revision,
         // 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(config.etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
+            key: b64(etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
             target: 'MOD',
             mod_revision: ''+etcd_watch_revision,
             result: 'LESS',
@@ -153,7 +132,7 @@ function save_new_pgs_txn(save_to, request, state, config, etcd_watch_revision,
         {
             request.success.push({
                 requestPut: {
-                    key: b64(config.etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
+                    key: b64(etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
                     value: b64(JSON.stringify(pg_history[i])),
                 },
             });
@@ -162,7 +141,7 @@ function save_new_pgs_txn(save_to, request, state, config, etcd_watch_revision,
         {
             request.success.push({
                 requestDeleteRange: {
-                    key: b64(config.etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
+                    key: b64(etcd_prefix+'/pg/history/'+pool_id+'/'+(i+1)),
                 },
             });
         }

mon/tests/primary-stickiness.js

@@ -1,158 +0,0 @@
-// 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,12 +1187,16 @@ 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 (int i = 0; i < recheck_count; i++, wr = skip_commits(prev(wr)))
+            for (; wr && wr->type() == BS_HEAP_SMALL_WRITE && checked < recheck_count; 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,7 +184,8 @@ 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);
-        if (!dsk.disable_data_fsync)
+        // 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)
         {
             // use PRIV->lsn for fsync_data_id
             PRIV(op)->lsn = ++data_fsync_next;

src/client/etcd_state_client_mock.cpp

@@ -322,8 +322,18 @@ void etcd_state_client_mock_t::mock_apply_changes(std::map<std::string, etcd_kv_
 {
     for (auto & kv: changes)
         parse_state(kv.second);
-    if (on_change_hook != NULL)
-        on_change_hook(changes);
+    // 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);
 }
 
 void etcd_state_client_mock_t::load_global_config()
@@ -333,5 +343,18 @@ 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) {});
+    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);
+    });
 }

src/client/etcd_state_client_mock.h

@@ -62,6 +62,10 @@ 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_ls.cpp

@@ -250,11 +250,17 @@ resume_1:
 
     json11::Json::array to_tree(const json11::Json::array & array)
     {
+        std::set<uint64_t> listed;
+        for (const auto & item: array)
+        {
+            listed.insert(item["inode_id"].uint64_value());
+        }
         std::map<uint64_t, json11::Json::array> children;
         for (const auto & item: array)
         {
             uint64_t parent_id = item["parent_inode_id"].uint64_value();
-            children[parent_id].push_back(item);
+            // Layers with a filtered out parent are shown as roots
+            children[listed.find(parent_id) != listed.end() ? parent_id : 0].push_back(item);
         }
         json11::Json::array tree;
         std::function<void(uint64_t, const std::string &)> add_children = [&](uint64_t parent_id, const std::string & prefix)

src/test/test_blockstore.cpp

@@ -1406,6 +1406,124 @@ 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
@@ -2004,6 +2122,52 @@ 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();
@@ -2029,9 +2193,11 @@ 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,6 +480,8 @@ 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
@@ -488,6 +490,14 @@ 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;
@@ -1590,32 +1600,83 @@ 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());
-    auto & pg_cfg = f.st_cli->pool_config[POOL_ID].pg_config[PG_NUM];
+    // 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();
+        }
+    }
     // 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
-    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));
+    f.st_cli->pool_config[POOL_ID].pg_config[PG_NUM].cur_primary = 1;
 }
 
 void chaos_t::connect_session()
@@ -1668,10 +1729,14 @@ 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).  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).  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
     for (auto *p: inflight)
         if (p->done)
             account(p);
@@ -2449,26 +2514,13 @@ 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
-    // 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)
-            }
-        }
-    }
+    // 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
     // 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
@@ -2864,9 +2916,10 @@ 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, 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
+    // 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
     // 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
@@ -3014,10 +3067,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 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
+            // 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
             // still ACKED means established bytes vanished with no event in
             // between: a non-repeatable read
             if (d.state == V_ACKED)

tests/run_tests.sh

@@ -122,6 +122,7 @@ OLD=1 ./test_corrupt_all.sh
 
 ./test_reweight_half.sh
 ./test_snapshot_pool2.sh
+./test_ls_tree.sh
 ./test_snapshot_read_bitmap.sh
 
 TEST_NAME=csum_32k_dmj OSD_ARGS="--data_csum_type crc32c --csum_block_size 32k --inmemory_metadata false --inmemory_journal false" OFFSET_ARGS=$OSD_ARGS ./test_heal.sh

tests/test_ls_tree.sh

@@ -0,0 +1,40 @@
+#!/bin/bash -ex
+
+# Test vitastor-cli ls --tree with filters which leave out parent layers
+
+. `dirname $0`/run_3osds.sh
+
+$VITASTOR_CLI create -s 128M testimg
+$VITASTOR_CLI snap-create testimg@snap1
+$VITASTOR_CLI create --parent testimg@snap1 testclone
+
+$VITASTOR_CLI create-pool testpool2 -s 3 -n 4 --failure_domain osd
+wait_pool_up 30 2 3 4
+$VITASTOR_CLI create --parent testimg@snap1 -p testpool2 testclone2
+
+# First column of each row without the header
+check_tree()
+{
+    local expected=$1
+    shift
+    local rows=$($VITASTOR_CLI ls --tree "$@" | tail -n +2 | sed 's/  .*//' | tr '\n' ';')
+    if [[ "$rows" != "$expected" ]]; then
+        format_error "ls --tree $* shows '$rows', expected '$expected'"
+    fi
+}
+
+$VITASTOR_CLI ls --tree
+
+check_tree 'testimg@snap1;├─ testclone;├─ testclone2;└─ testimg;'
+check_tree 'testimg@snap1;└─ testimg;' 'testimg*'
+# Layers whose parent is filtered out are shown as roots
+check_tree 'testimg;' testimg
+check_tree 'testclone;testimg;' testclone testimg
+check_tree 'testclone;' -n 1
+check_tree 'testclone2;' -p testpool2
+
+if ! $VITASTOR_CLI ls --tree testimg | grep -q ' testimg@snap1 *$'; then
+    format_error "ls --tree testimg should show testimg@snap1 as the parent"
+fi
+
+format_green OK