src/blockstore/v1/flush.cpp

@@ -176,6 +176,22 @@ void journal_flusher_t::unshift_flush(obj_ver_id ov, bool force)
     }
 }
 
+// If the object is being flushed, let that flush requeue <ov> when it finishes. Queueing it now
+// would leave it in the queue after the flush erases it, and it would hide newer versions
+bool journal_flusher_t::requeue_after_flush(obj_ver_id ov)
+{
+    auto repeat_it = sync_to_repeat.find(ov.oid);
+    if (repeat_it == sync_to_repeat.end())
+    {
+        return false;
+    }
+    if (repeat_it->second < ov.version)
+    {
+        repeat_it->second = ov.version;
+    }
+    return true;
+}
+
 void journal_flusher_t::remove_flush(object_id oid)
 {
 #ifdef BLOCKSTORE_DEBUG
@@ -231,6 +247,50 @@ void journal_flusher_t::release_trim()
     trim_wanted--;
 }
 
+// The flusher puts back entries whose journal sector isn't written yet and stops if it finds
+// nothing else. Writing the sector makes them flushable, and nothing else may wake it then
+void journal_flusher_t::wakeup_on_journal_written()
+{
+    if (!dequeuing && flush_queue.size() && (flush_queue.size() >= flusher_start_threshold || trim_wanted > 0))
+    {
+        dequeuing = true;
+        bs->ringloop->wakeup();
+    }
+}
+
+// An object's newest stable version is reachable when the object is being flushed, or when
+// it's queued under a version that is still in dirty_db and not older than that one: the
+// flusher skips a queued version it can't find, and flushes only up to the queued version
+std::vector<obj_ver_id> journal_flusher_t::find_unflushable_stable()
+{
+    std::vector<obj_ver_id> lost;
+    auto & dirty_db = bs->dirty_db;
+    auto it = dirty_db.begin();
+    while (it != dirty_db.end())
+    {
+        object_id oid = it->first.oid;
+        uint64_t stable_ver = 0;
+        for (; it != dirty_db.end() && it->first.oid == oid; it++)
+        {
+            if (IS_STABLE(it->second.state))
+                stable_ver = it->first.version;
+        }
+        if (!stable_ver || sync_to_repeat.find(oid) != sync_to_repeat.end())
+            continue;
+        auto fv = flush_versions.find(oid);
+        if (fv != flush_versions.end() && fv->second >= stable_ver &&
+            dirty_db.find((obj_ver_id){ .oid = oid, .version = fv->second }) != dirty_db.end())
+            continue;
+        lost.push_back((obj_ver_id){ .oid = oid, .version = stable_ver });
+    }
+    return lost;
+}
+
+std::vector<obj_ver_id> blockstore_impl_t::find_unflushable_stable()
+{
+    return flusher->find_unflushable_stable();
+}
+
 void journal_flusher_t::dump_diagnostics()
 {
     const char *unflushable_type = "";

src/blockstore/v1/flush.h

@@ -132,9 +132,12 @@ public:
     void mark_trim_possible();
     void request_trim();
     void release_trim();
+    void wakeup_on_journal_written();
     void enqueue_flush(obj_ver_id oid);
     void unshift_flush(obj_ver_id oid, bool force);
+    bool requeue_after_flush(obj_ver_id ov);
     void remove_flush(object_id oid);
     void dump_diagnostics();
     bool is_mutated(uint64_t clean_loc);
+    std::vector<obj_ver_id> find_unflushable_stable();
 };

src/blockstore/v1/impl.h

@@ -346,6 +346,9 @@ public:
     // Print diagnostics to stdout
     void dump_diagnostics();
 
+    // Stable dirty versions that no flush will ever reach. Scans all of dirty_db: for tests only
+    std::vector<obj_ver_id> find_unflushable_stable();
+
     // Get diagnostic string for an operation
     std::string get_op_diag(blockstore_op_t *op);
 

src/blockstore/v1/write.cpp

@@ -353,7 +353,7 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
 void blockstore_impl_t::unshift_stable_flush(obj_ver_id ov)
 {
     auto dirty_it = dirty_db.find(ov);
-    if (dirty_it != dirty_db.end() && IS_STABLE(dirty_it->second.state))
+    if (dirty_it != dirty_db.end() && IS_STABLE(dirty_it->second.state) && !flusher->requeue_after_flush(ov))
     {
         flusher->unshift_flush(ov, true);
     }
@@ -925,6 +925,7 @@ void blockstore_impl_t::mark_not_dirty(int journal_sector)
         (new_ds + (new_ds >= journal.used_start ? 0 : journal.len)))
     {
         journal.dirty_start = new_ds;
+        flusher->wakeup_on_journal_written();
     }
 }
 

src/test/test_blockstore_chaos.cpp

@@ -66,6 +66,8 @@ struct chaos_probs_t
 };
 
 static chaos_probs_t probs;
+// Check the store's internal state after every drain, which takes full scans of it
+static bool hardened = false;
 
 // What a single step decided to do
 enum chaos_action_t
@@ -114,6 +116,28 @@ static void dump_extra(v1::blockstore_impl_t *bs)
 {
 }
 
+// A stable version no flush reaches pins the journal for good, and the store only shows that
+// as writes refused with EAGAIN - expected under pressure - so --hardened checks it explicitly
+static void check_flush_liveness(blockstore_impl_t *bs, uint64_t step_no)
+{
+}
+
+static void check_flush_liveness(v1::blockstore_impl_t *bs, uint64_t step_no)
+{
+    if (!hardened)
+        return;
+    auto lost = bs->find_unflushable_stable();
+    if (!lost.size())
+        return;
+    fprintf(stderr, "step %ju: %zu stable version(s) will never be flushed:", step_no, lost.size());
+    for (auto & ov: lost)
+        fprintf(stderr, " %jx:%jx v%ju", ov.oid.inode, ov.oid.stripe, ov.version);
+    fprintf(stderr, "\n");
+    bs->dump_diagnostics();
+    fflush(stdout);
+    abort();
+}
+
 // What we believe about one object.
 //
 // `submitted` is the highest version we ever enqueued a write for - not necessarily
@@ -1052,6 +1076,7 @@ struct chaos_t
     void do_restart()
     {
         drain();
+        check_flush_liveness(bs, step_no);
         restarts++;
         std::vector<std::vector<uint8_t>> before(OBJ_COUNT);
         std::vector<uint64_t> before_ver(OBJ_COUNT, 0);
@@ -1201,6 +1226,7 @@ struct chaos_t
             step();
         // Everything still in flight must complete, and everything durable must still be there
         drain();
+        check_flush_liveness(bs, step_no);
         for (int obj = 0; obj < OBJ_COUNT; obj++)
             check_object(obj);
     }
@@ -1283,6 +1309,9 @@ static const char *help_text =
     "  --steps <n>         Steps per seed (default %ju)\n"
     "  --config <n>        Force configuration <n> instead of deriving it from the seed\n"
     "  --max-inflight <n>  Operations in flight at once (default %u)\n"
+    "  --hardened          After every drain, also check the store's internal state, which\n"
+    "                      takes full scans of it: in store v1, that every stable version\n"
+    "                      gets flushed\n"
     "  -v, --trace         Trace every operation and every simulated disk request\n"
     "  -h, --help          Show this help\n"
     "\n"
@@ -1383,6 +1412,8 @@ int main(int narg, char *args[])
         }
         else if (!strcmp(opt, "-v") || !strcmp(opt, "--trace"))
             trace = true;
+        else if (!strcmp(opt, "--hardened"))
+            hardened = true;
         else if (!strcmp(opt, "--seed"))
         {
             take(v);