checkpoint: peek-then-pop in flush_pending_checkpoints

The queue holds serialized beespoint.ini blobs that snapshot_point()
has handed off; the queue owns the obligation to land them on disk
until a write succeeds.  The previous drain loop popped every
eligible entry up-front and kept only the last one's string on the
stack, so a throw from m_point_file.write() destroyed the only
remaining copy.  In the worst case (no further crawler progress
before shutdown) the unflushed checkpoint was silently lost.

Restructure as peek-then-pop:

  1. Coalesce by popping all earlier eligible entries -- those are
     intentionally superseded by later ones.  The eligibility check
     moves to the second entry so the latest eligible one survives at
     the front.
  2. Verify the surviving head is itself eligible; if not, return.
  3. Copy (do not move) the front entry's ini_string and release the
     lock.
  4. Write outside the lock.  On throw, the entry stays at the front
     for the next call to retry -- or, if a newer entry has been
     enqueued meanwhile, the next coalesce iteration discards this
     one in favour of the newer one.
  5. On success, re-lock and pop.  Safe because the queue is
     single-consumer (writeback thread and the shutdown flush do not
     overlap) and snapshot_point() only ever appends to the tail, so
     the front is still the entry we wrote.

The clean/dirty bookkeeping in snapshot_point() is unchanged: clean
still advances at enqueue, because "clean" means "no new serialized
data needs to be queued", not "all bytes are on disk".  Persistence
of the queued blobs is the queue's responsibility, which this commit
now fulfils across write failures.

Assisted-by: Claude-Code:claude-opus-4-7
Signed-off-by: Zygo Blaxell <bees@furryterror.org>

diff --git a/src/bees-roots.cc b/src/bees-roots.cc
index d8a06a56091b7a6e969656fb001ae28cbbeef017..746e5ab2b6564ac4a3504c649a22123991e875d2 100644 (file)
--- a/src/bees-roots.cc
+++ b/src/bees-roots.cc
@@ -1855,7 +1855,7 @@ void
 BeesRoots::flush_pending_checkpoints(bool force)
 {
        while (true) {
-               string last_ini;
+               string write_ini;
                {
                        unique_lock<mutex> lock(m_mutex);
                        if (m_pending_checkpoints.empty()) return;
@@ -1863,21 +1863,44 @@ BeesRoots::flush_pending_checkpoints(bool force)
                        const auto extent_count  = m_ctx->hash_table()->extent_count();
                        const auto writeback_now = m_ctx->hash_table()->writeback_extent_count();
 
-                       // Drain all entries eligible under writeback_now (or all if forced),
-                       // keeping only the last one's content to write.
-                       while (!m_pending_checkpoints.empty()) {
-                               const auto &entry = m_pending_checkpoints.front();
-                               if (!force && writeback_now - entry.writeback_count < extent_count) {
+                       // Coalesce: drop earlier eligible entries (superseded by later
+                       // ones).  writeback_count is captured monotonically at push_back
+                       // time, so eligibility is monotone in queue order: if the second
+                       // entry is eligible, the first is also eligible and superseded.
+                       while (m_pending_checkpoints.size() > 1) {
+                               const auto &second = m_pending_checkpoints[1];
+                               if (!force && writeback_now - second.writeback_count < extent_count) {
                                        break;
                                }
-                               last_ini = std::move(entry.ini_string);
                                m_pending_checkpoints.pop_front();
                        }
-               }  // lock released here
 
-               if (last_ini.empty()) return;   // nothing became eligible
-               m_point_file.write(last_ini);   // slow I/O outside the lock
-               // loop: more entries may be eligible now
+                       // Check whether the surviving head is itself eligible.
+                       const auto &front = m_pending_checkpoints.front();
+                       if (!force && writeback_now - front.writeback_count < extent_count) {
+                               return;
+                       }
+
+                       // Peek-then-pop: copy (do not move) the ini_string and leave the
+                       // entry in the queue.  If m_point_file.write() throws, the entry
+                       // stays for the next call to retry -- or, if a newer entry has
+                       // been enqueued meanwhile, the next coalesce iteration discards
+                       // this one in favour of the newer one.  Either way the queue
+                       // holds the obligation until a write succeeds.
+                       write_ini = front.ini_string;
+               }
+
+               m_point_file.write(write_ini);  // slow I/O outside the lock; may throw
+
+               // Write succeeded.  Pop the entry we just wrote.  The front is still
+               // that entry because flush_pending_checkpoints is single-consumer
+               // (the writeback thread, REPL save-state, and the shutdown flush do
+               // not overlap), and snapshot_point() only ever appends to the tail.
+               {
+                       unique_lock<mutex> lock(m_mutex);
+                       m_pending_checkpoints.pop_front();
+               }
+               // loop: more entries may now be eligible, or newer ones have arrived
        }
 }