Store v2: Do not take a data fsync slot for redirect_intent writes

Since 0c888845 every big write with data fsync enabled takes a slot in
data_fsyncs, but only big_writes mark theirs completed, redirect_intent
writes skip the data fsync. Their slot stays forever and every later
big_write waits for it, so after the first stable write to a new object
all unstable big writes of the OSD hang.

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/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)