MB-51408: Don't miss closing the open checkpoint at memory recovery

There are a couple of DEV_ASSERTS in the CheckpointMemRecoveryTask that
wrap an important step of the checkpoint memory recovery logic: closing
the open checkpoint and creating a new one, if possible.

Recently we set DEV_ASSERTS=OFF for Neo, which caused the task missing
that step. Consequence is that we might enter unrecoverable TempOOM
states, eg:
 1. Store mutation A in vb1/checkpoint1 - no OOM
 2. Store mutation B in vb2/checkpoint1 - OOM
 3. CheckpointMemRecoveryTask runs and doesn't close the open
    checkpoints
 4. The same task can't expel anything as we never expel high-seqno
 5. Unrecoverable OOM

This patch fixes by reverting step (3) back into the code.
Covered by unit tests for both persistent/ephemeral.

Change-Id: Id51b837bd9500a26c89ae7999a8a4a60cd1930b5
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/172215
Tested-by: Build Bot <[email protected]>
Reviewed-by: James H <[email protected]>
Well-Formed: Restriction Checker

Author: paolococchi

engines/ep/src/checkpoint_config.h

@@ -55,6 +55,11 @@ class CheckpointConfig {
         return checkpointRemovalMode;
     }
 
+    // @todo: Test only. Remove as soon as param made dynamic in EPConfig.
+    void setCheckpointRemovalMode(CheckpointRemoval mode) {
+        checkpointRemovalMode = mode;
+    }
+
 protected:
     friend class EventuallyPersistentEngine;
     friend class SynchronousEPEngine;

engines/ep/src/checkpoint_remover.cc

@@ -183,6 +183,10 @@ CheckpointMemRecoveryTask::attemptCursorDropping() {
             case CheckpointRemoval::Eager: {
 #if CB_DEVELOPMENT_ASSERTS
                 Expects(manager.removeClosedUnrefCheckpoints().count == 0);
+#else
+                // MB-51408: We need to make the call for keeping executing the
+                // inner checkpoint creation logic - minimal fix for Neo.
+                manager.removeClosedUnrefCheckpoints();
 #endif
                 break;
             }
@@ -244,6 +248,10 @@ bool CheckpointMemRecoveryTask::runInner() {
         // This is not cheap to verify, as it requires scanning every
         // vbucket, so is only checked if dev asserts are on.
         Expects(attemptCheckpointRemoval().second == 0);
+#else
+        // MB-51408: We need to make the call for keeping executing the inner
+        // checkpoint creation logic - minimal fix for Neo.
+        attemptCheckpointRemoval();
 #endif
         break;
     }

engines/ep/tests/ep_testsuite.cc

@@ -1801,9 +1801,6 @@ static enum test_result test_vbucket_destroy_stats(EngineIface* h) {
     }
     wait_for_flusher_to_settle(h);
     testHarness->time_travel(65);
-    // store one more item to close the previous checkpoint, allowing it
-    // to be removed
-    wait_for_persisted_value(h, "padding-key", "some value", Vbid(1));
     wait_for_stat_change(h, "ep_items_rm_from_checkpoints", itemsRemoved);
 
     check(set_vbucket_state(h, Vbid(1), vbucket_state_dead),
@@ -2128,9 +2125,6 @@ static enum test_result test_mem_stats(EngineIface* h) {
     int itemsRemoved = get_int_stat(h, "ep_items_rm_from_checkpoints");
     wait_for_persisted_value(h, "key", value.c_str());
     testHarness->time_travel(65);
-    // store a second item after advancing time. This will trigger a checkpoint
-    // close, and will allow the previous checkpoint to be removed.
-    wait_for_persisted_value(h, "key2", value.c_str());
     if (isPersistentBucket(h)) {
         wait_for_stat_change(h, "ep_items_rm_from_checkpoints", itemsRemoved);
     }
@@ -4260,15 +4254,12 @@ static enum test_result test_disk_gt_ram_golden(EngineIface* h) {
     // Store some data and check post-set state.
     wait_for_persisted_value(h, "k1", "some value");
     testHarness->time_travel(65);
-    // store one more item to close the previous checkpoint, allowing it
-    // to be removed
-    wait_for_persisted_value(h, "padding-key", "some value");
     wait_for_stat_change(h, "ep_items_rm_from_checkpoints", itemsRemoved);
 
     checkeq(0,
             get_int_stat(h, "ep_bg_fetched"),
             "Should start with zero bg fetches");
-    checkeq((initial_enqueued + 2),
+    checkeq((initial_enqueued + 1),
             get_int_stat(h, "ep_total_enqueued"),
             "Should have additional item enqueued after store");
     int kv_size = get_int_stat(h, "ep_kv_size");
@@ -4288,7 +4279,7 @@ static enum test_result test_disk_gt_ram_golden(EngineIface* h) {
     checkeq(1,
             get_int_stat(h, "ep_bg_fetched"),
             "BG fetches should be one after reading an evicted key");
-    checkeq((initial_enqueued + 2),
+    checkeq((initial_enqueued + 1),
             get_int_stat(h, "ep_total_enqueued"),
             "Item should not be marked dirty after reading an evicted key");
 
@@ -4303,9 +4294,6 @@ static enum test_result test_disk_gt_ram_golden(EngineIface* h) {
             "Failed remove with value.");
     wait_for_stat_change(h, "ep_total_persisted", numStored);
     testHarness->time_travel(65);
-    // store one more item to close the previous checkpoint, allowing it
-    // to be removed
-    wait_for_persisted_value(h, "padding-key", "some value");
     wait_for_stat_change(h, "ep_items_rm_from_checkpoints", itemsRemoved);
 
     return SUCCESS;
@@ -4339,9 +4327,6 @@ static enum test_result test_disk_gt_ram_paged_rm(EngineIface* h) {
             "Failed remove with value.");
     wait_for_stat_change(h, "ep_total_persisted", numStored);
     testHarness->time_travel(65);
-    // store one more item to close the previous checkpoint, allowing it
-    // to be removed
-    wait_for_persisted_value(h, "padding-key", "some value");
     wait_for_stat_change(h, "ep_items_rm_from_checkpoints", itemsRemoved);
 
     return SUCCESS;

engines/ep/tests/ep_testsuite_checkpoint.cc

@@ -109,11 +109,6 @@ static enum test_result test_checkpoint_timeout(EngineIface* h) {
             store(h, nullptr, StoreSemantics::Set, "key", "value"),
             "Failed to store an item.");
     testHarness->time_travel(600);
-    // store another item after advancing time, should close the previous
-    // checkpoint
-    checkeq(cb::engine_errc::success,
-            store(h, nullptr, StoreSemantics::Set, "key2", "value"),
-            "Failed to store second item.");
     wait_for_stat_to_be(h, "vb_0:open_checkpoint_id", 2, "checkpoint");
     return SUCCESS;
 }

engines/ep/tests/ep_testsuite_dcp.cc

@@ -2002,18 +2002,18 @@ static enum test_result test_dcp_producer_stream_req_partial(EngineIface* h) {
             get_int_stat(h, "vb_0:open_checkpoint_id", "checkpoint");
     checkeq(1, initial_ckpt_id, "Expected to start at checkpoint ID 1");
 
-    // Create two 'full' checkpoints by storing 2 x 'chk_max_items' + 1
+    // Create two 'full' checkpoints by storing exactly 2 x 'chk_max_items'
     // into the VBucket.
     const auto max_ckpt_items = get_int_stat(h, "ep_chk_max_items");
 
-    write_items(h, max_ckpt_items + 1);
+    write_items(h, max_ckpt_items);
     wait_for_flusher_to_settle(h);
     wait_for_stat_to_be(h, "ep_items_rm_from_checkpoints", max_ckpt_items);
     checkeq(initial_ckpt_id + 1,
             get_int_stat(h, "vb_0:open_checkpoint_id", "checkpoint"),
             "Expected #checkpoints to increase by 1 after storing items");
 
-    write_items(h, max_ckpt_items + 1, max_ckpt_items);
+    write_items(h, max_ckpt_items, max_ckpt_items);
     wait_for_flusher_to_settle(h);
     wait_for_stat_to_be(h, "ep_items_rm_from_checkpoints", max_ckpt_items * 2);
     checkeq(initial_ckpt_id + 2,
@@ -2054,11 +2054,11 @@ static enum test_result test_dcp_producer_stream_req_partial(EngineIface* h) {
     ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
     ctx.seqno = {105, 209};
     ctx.snapshot = {105, 105};
-    ctx.exp_mutations = 97; // 105 to 200
+    ctx.exp_mutations = 95; // 105 to 200
     ctx.exp_deletions = 100; // 201 to 300
 
     if (isPersistentBucket(h)) {
-        ctx.exp_markers = 3;
+        ctx.exp_markers = 2;
     } else {
         // the ephemeral stream request won't be broken into two snapshots of
         // backfill from disk vs the checkpoint in memory
@@ -2257,41 +2257,29 @@ static test_result testDcpProducerExpiredItemBackfill(
 }
 
 static enum test_result test_dcp_producer_stream_req_backfill(EngineIface* h) {
-    const auto max_ckpt_items = get_int_stat(h, "ep_chk_max_items");
-
-    const int batch_items = max_ckpt_items;
-    const int num_items = batch_items * 2;
+    const int num_items = 400, batch_items = 200;
     for (int start_seqno = 0; start_seqno < num_items;
          start_seqno += batch_items) {
-        if (start_seqno == batch_items) {
-            // stop persistence after the first batch is on disk
+        if (200 == start_seqno) {
             wait_for_flusher_to_settle(h);
-            wait_for_stat_to_be(h,
-                                "vb_0:persistence_seqno",
-                                batch_items,
-                                "vbucket-details");
+            wait_for_stat_to_be(h, "ep_items_rm_from_checkpoints", 200);
             stop_persistence(h);
         }
         write_items(h, batch_items, start_seqno);
     }
 
-    // Wait for the first checkpoint to be removed.
-    // The second checkpoint is the open checkpoint (and for a persistent
-    // bucket, contains unpersisted items) and won't be removed
-    wait_for_stat_to_be_gte(h, "ep_items_rm_from_checkpoints", batch_items);
-
     auto* cookie = testHarness->create_cookie(h);
 
     DcpStreamCtx ctx;
     ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
-    ctx.seqno = {0, uint64_t(batch_items)};
+    ctx.seqno = {0, 200};
     // The idea here is that at backfill we get the full Disk/SeqList snapshot.
     // Persistence has been stopped at seqno 200, while Ephemeral stores all
     // seqnos in the SeqList.
     if (isEphemeralBucket(h)) {
-        ctx.exp_mutations = num_items;
+        ctx.exp_mutations = 400;
     } else {
-        ctx.exp_mutations = batch_items;
+        ctx.exp_mutations = 200;
     }
     ctx.exp_markers = 1;
 
@@ -2360,11 +2348,6 @@ static enum test_result test_dcp_producer_disk_backfill_buffer_limits(
     wait_for_flusher_to_settle(h);
     verify_curr_items(h, num_items, "Wrong amount of items");
 
-    testHarness->time_travel(65);
-    // store one more item after advancing time to close the previous
-    // checkpoint, allowing it to be removed
-    wait_for_persisted_value(h, "padding-key", "some value");
-
     /* Wait for the checkpoint to be removed so that upon DCP connection
        backfill is scheduled */
     wait_for_stat_to_be(h, "ep_items_rm_from_checkpoints", num_items);
@@ -2374,7 +2357,7 @@ static enum test_result test_dcp_producer_disk_backfill_buffer_limits(
     DcpStreamCtx ctx;
     ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
     ctx.seqno = {0, num_items};
-    ctx.exp_mutations = num_items + 1;
+    ctx.exp_mutations = 3;
     ctx.exp_markers = 1;
 
     TestDcpConsumer tdc("unittest", cookie, h);
@@ -2908,12 +2891,7 @@ static enum test_result test_dcp_producer_stream_cursor_movement(
                        cdc.dcpConsumer->producers.last_byseqno,
                        exp_items);
 
-    testHarness->time_travel(65);
-    // store one more item to close the previous checkpoint, allowing it
-    // to be removed
-    wait_for_persisted_value(h, "padding-key", "some value");
-
-    /* Wait for new open checkpoint to be added */
+    /* Wait for new open (empty) checkpoint to be added */
     wait_for_stat_to_be(
             h, "vb_0:open_checkpoint_id", curr_chkpt_id + 1, "checkpoint");
 
@@ -3774,11 +3752,6 @@ static enum test_result test_failover_scenario_two_with_dcp(EngineIface* h) {
     // Front-end operations (sets)
     write_items(h, 2, 1, "key_");
 
-    testHarness->time_travel(65);
-    // store one more item to close the previous checkpoint, allowing it
-    // to be removed
-    wait_for_persisted_value(h, "padding-key", "some value");
-
     // Wait for a new open checkpoint
     wait_for_stat_to_be(
             h, "vb_0:open_checkpoint_id", openCheckpointId + 1, "checkpoint");
@@ -6156,19 +6129,14 @@ static enum test_result test_dcp_last_items_purged(EngineIface* h) {
             get_int_stat(h, "vb_0:purge_seqno", "vbucket-seqno"),
             "purge_seqno didn't match expected value");
 
-    testHarness->time_travel(65);
-    // store one more item to close the previous checkpoint, allowing it
-    // to be removed
-    wait_for_persisted_value(h, "padding-key", "some value");
-
     wait_for_stat_to_be(h, "vb_0:open_checkpoint_id", 2, "checkpoint");
     wait_for_stat_to_be(h, "vb_0:num_checkpoints", 1, "checkpoint");
 
     /* Create a DCP stream */
     DcpStreamCtx ctx;
     ctx.vb_uuid = get_ull_stat(h, "vb_0:0:id", "failovers");
     ctx.seqno = {0, get_ull_stat(h, "vb_0:high_seqno", "vbucket-seqno")};
-    ctx.exp_mutations = 2;
+    ctx.exp_mutations = 1;
     ctx.exp_deletions = 1;
     ctx.exp_markers = 1;
     ctx.skip_estimate_check = true;
@@ -6241,11 +6209,6 @@ static enum test_result test_dcp_rollback_after_purge(EngineIface* h) {
             get_int_stat(h, "vb_0:purge_seqno", "vbucket-seqno"),
             "purge_seqno didn't match expected value");
 
-    testHarness->time_travel(65);
-    // store one more item to close the previous checkpoint, allowing it
-    // to be removed
-    wait_for_persisted_value(h, "padding-key", "some value");
-
     wait_for_stat_to_be(h, "vb_0:open_checkpoint_id", 2, "checkpoint");
     wait_for_stat_to_be(h, "vb_0:num_checkpoints", 1, "checkpoint");
 

engines/ep/tests/module_tests/checkpoint_remover_test.cc

@@ -390,6 +390,65 @@ TEST_P(CheckpointRemoverTest, expelWithoutCursor) {
     EXPECT_GT(stats.memFreedByCheckpointItemExpel, 0);
 }
 
+TEST_P(CheckpointRemoverTest, MemRecoveryByCheckpointCreation) {
+    setVBucketStateAndRunPersistTask(Vbid(0), vbucket_state_active);
+    setVBucketStateAndRunPersistTask(Vbid(1), vbucket_state_active);
+
+    auto& config = engine->getConfiguration();
+    config.setChkExpelEnabled(true);
+    config.setMaxSize(1024 * 1024 * 100);
+
+    // @todo: Convert to config.setCheckpointRemovalMode("eager") once the
+    //  param has been made dynamic
+    auto& ckptConfig = engine->getCheckpointConfig();
+    ckptConfig.setCheckpointRemovalMode(CheckpointRemoval::Eager);
+    ASSERT_TRUE(engine->getCheckpointConfig().isEagerCheckpointRemoval());
+
+    ASSERT_EQ(0, store->getRequiredCheckpointMemoryReduction());
+
+    // Compute paylaod size such that we enter a TempOOM phase when we store
+    // the second item.
+    const size_t valueSize =
+            config.getMaxSize() * config.getCheckpointMemoryRatio() *
+                    config.getCheckpointMemoryRecoveryUpperMark() / 2 +
+            1;
+    const auto value = std::string(valueSize, 'x');
+    // Store first item, no checkpoint OOM yet
+    store_item(Vbid(0), makeStoredDocKey("keyA"), value);
+    EXPECT_EQ(0, store->getRequiredCheckpointMemoryReduction());
+    // Store second item, Checkpoint OOM
+    store_item(Vbid(1), makeStoredDocKey("keyB"), value);
+    ASSERT_GT(store->getRequiredCheckpointMemoryReduction(), 0);
+
+    // Move the cursors to the end of the open checkpoint. Step required to
+    // allow checkpoint creation + cursor jumping into the new checkpoints in
+    // the next steps
+    flushVBucketToDiskIfPersistent(Vbid(0), 1);
+    flushVBucketToDiskIfPersistent(Vbid(1), 1);
+
+    const auto& stats = engine->getEpStats();
+    ASSERT_EQ(0, stats.itemsExpelledFromCheckpoints);
+    ASSERT_EQ(0, stats.itemsRemovedFromCheckpoints);
+
+    // Mem-recovery is expected to:
+    // 1. Create a new checkpoint on at least 1 vbucket
+    // 2. Move the cursors from the closed checkpoint to the open one
+    // 3. Remove the closed (and now unred) checkpoint
+    const auto remover = std::make_shared<CheckpointMemRecoveryTask>(
+            engine.get(),
+            engine->getEpStats(),
+            engine->getConfiguration().getChkRemoverStime(),
+            0);
+    remover->run();
+
+    // That allows to remove checkpoints and recover from OOM
+    // Before the fix, nothing removed from checkpoints and mem-reduction still
+    // required at this point
+    EXPECT_EQ(0, stats.itemsExpelledFromCheckpoints);
+    EXPECT_GT(stats.itemsRemovedFromCheckpoints, 0);
+    EXPECT_EQ(0, store->getRequiredCheckpointMemoryReduction());
+}
+
 // Test written for MB-36366. With the fix removed this test failed because
 // post expel, we continued onto cursor dropping.
 // MB-36447 - unreliable test, disabling for now