Don't let a blocked maintenance operation inhibit remaining maintenance queue

The old maintenance scheduler behavior is to only remove a bucket from the
priority DB if its maintenance operation was successfully started. Failing
to start an operation could happen from both max pending throttling as well
as operation/bucket-specific blocking behavior. Since the scheduler would
encounter the same bucket as the one previously blocked upon its next tick
invocation, a single blocked bucket would run the risk of head-of-line
stalling the rest of the remaining maintenance queue (assuming the ongoing
DB scan did not encounter any higher priority buckets).

This commit changes the following aspects of maintenance scheduling:
 * Always clear entries from the priority DB before trying to start an
   operation. A blocked operation will be retried the next time the
   regular bucket DB scan encounters the bucket.
 * Avoid trying to start (and clear) inherently doomed operations by
   _not_ trying to schedule any operations if it would be blocked due
   to too many pending maintenance operations anyway. Introduces a
   new `PendingWindowChecker` interface for this purpose.
 * Explicitly inhibit all maintenance scheduling if a pending cluster
   state is present. Operations are already _implicitly_ blocked from
   starting if there's a pending cluster state, but this would cause
   the priority DB from being pointlessly cleared.

1 files changed, 50 insertions, 33 deletions

diff --git a/storage/src/tests/distributor/maintenanceschedulertest.cpp b/storage/src/tests/distributor/maintenanceschedulertest.cpp
index a97ffeef24b..8f13e10fedc 100644
--- a/storage/src/tests/distributor/maintenanceschedulertest.cpp
+++ b/storage/src/tests/distributor/maintenanceschedulertest.cpp
@@ -18,60 +18,77 @@ using Priority = MaintenancePriority;
 using WaitTimeMs = MaintenanceScheduler::WaitTimeMs;
 
 struct MaintenanceSchedulerTest : Test {
-    std::unique_ptr<SimpleBucketPriorityDatabase> _priorityDb;
-    std::unique_ptr<MockMaintenanceOperationGenerator> _operationGenerator;
-    std::unique_ptr<MockOperationStarter> _operationStarter;
-    std::unique_ptr<MaintenanceScheduler> _scheduler;
+    SimpleBucketPriorityDatabase      _priority_db;
+    MockMaintenanceOperationGenerator _operation_generator;
+    MockOperationStarter              _operation_starter;
+    MockPendingWindowChecker          _pending_window_checker;
+    MaintenanceScheduler              _scheduler;
 
-    void SetUp() override;
+    MaintenanceSchedulerTest()
+        : _priority_db(),
+          _operation_generator(),
+          _operation_starter(),
+          _pending_window_checker(),
+          _scheduler(_operation_generator, _priority_db, _pending_window_checker, _operation_starter)
+    {}
 };
 
-void
-MaintenanceSchedulerTest::SetUp()
-{
-    _priorityDb = std::make_unique<SimpleBucketPriorityDatabase>();
-    _operationGenerator = std::make_unique<MockMaintenanceOperationGenerator>();
-    _operationStarter = std::make_unique<MockOperationStarter>();
-    _scheduler = std::make_unique<MaintenanceScheduler>(*_operationGenerator, *_priorityDb, *_operationStarter);
-}
-
 TEST_F(MaintenanceSchedulerTest, priority_cleared_after_scheduled) {
-    _priorityDb->setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::HIGHEST));
-    _scheduler->tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE);
-    EXPECT_EQ("", _priorityDb->toString());
+    _priority_db.setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::HIGHEST));
+    _scheduler.tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE);
+    EXPECT_EQ("", _priority_db.toString());
 }
 
 TEST_F(MaintenanceSchedulerTest, operation_is_scheduled) {
-    _priorityDb->setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::MEDIUM));
-    _scheduler->tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE);
+    _priority_db.setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::MEDIUM));
+    _scheduler.tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE);
     EXPECT_EQ("Bucket(BucketSpace(0x0000000000000001), BucketId(0x4000000000000001)), pri 100\n",
-              _operationStarter->toString());
+              _operation_starter.toString());
+}
+
+TEST_F(MaintenanceSchedulerTest, operation_is_not_scheduled_if_pending_ops_not_accepted) {
+    _priority_db.setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::MEDIUM));
+    _pending_window_checker.allow_operations(false);
+    _scheduler.tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE);
+    EXPECT_EQ("", _operation_starter.toString());
+    // Priority DB entry is not cleared
+    EXPECT_EQ("PrioritizedBucket(Bucket(BucketSpace(0x0000000000000001), BucketId(0x4000000000000001)), pri MEDIUM)\n",
+              _priority_db.toString());
 }
 
 TEST_F(MaintenanceSchedulerTest, no_operations_to_schedule) {
-    WaitTimeMs waitMs(_scheduler->tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE));
+    WaitTimeMs waitMs(_scheduler.tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE));
     EXPECT_EQ(WaitTimeMs(1), waitMs);
-    EXPECT_EQ("", _operationStarter->toString());
+    EXPECT_EQ("", _operation_starter.toString());
 }
 
 TEST_F(MaintenanceSchedulerTest, suppress_low_priorities_in_emergency_mode) {
-    _priorityDb->setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::VERY_HIGH));
-    _priorityDb->setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 2)), Priority::HIGHEST));
-    EXPECT_EQ(WaitTimeMs(0), _scheduler->tick(MaintenanceScheduler::RECOVERY_SCHEDULING_MODE));
-    EXPECT_EQ(WaitTimeMs(1), _scheduler->tick(MaintenanceScheduler::RECOVERY_SCHEDULING_MODE));
+    _priority_db.setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::VERY_HIGH));
+    _priority_db.setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 2)), Priority::HIGHEST));
+    EXPECT_EQ(WaitTimeMs(0), _scheduler.tick(MaintenanceScheduler::RECOVERY_SCHEDULING_MODE));
+    EXPECT_EQ(WaitTimeMs(1), _scheduler.tick(MaintenanceScheduler::RECOVERY_SCHEDULING_MODE));
     EXPECT_EQ("Bucket(BucketSpace(0x0000000000000001), BucketId(0x4000000000000002)), pri 0\n",
-              _operationStarter->toString());
+              _operation_starter.toString());
     EXPECT_EQ("PrioritizedBucket(Bucket(BucketSpace(0x0000000000000001), BucketId(0x4000000000000001)), pri VERY_HIGH)\n",
-              _priorityDb->toString());
+              _priority_db.toString());
+}
+
+TEST_F(MaintenanceSchedulerTest, priority_cleared_if_operation_not_started_inside_pending_window) {
+    _priority_db.setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::HIGH));
+    _operation_starter.setShouldStartOperations(false);
+    WaitTimeMs waitMs(_scheduler.tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE));
+    EXPECT_EQ(WaitTimeMs(1), waitMs);
+    EXPECT_EQ("", _priority_db.toString());
 }
 
-TEST_F(MaintenanceSchedulerTest, priority_not_cleared_if_operation_not_started) {
-    _priorityDb->setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::HIGH));
-    _operationStarter->setShouldStartOperations(false);
-    WaitTimeMs waitMs(_scheduler->tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE));
+TEST_F(MaintenanceSchedulerTest, priority_not_cleared_if_operation_not_started_outside_pending_window) {
+    _priority_db.setPriority(PrioritizedBucket(makeDocumentBucket(BucketId(16, 1)), Priority::HIGH));
+    _operation_starter.setShouldStartOperations(false);
+    _pending_window_checker.allow_operations(false);
+    WaitTimeMs waitMs(_scheduler.tick(MaintenanceScheduler::NORMAL_SCHEDULING_MODE));
     EXPECT_EQ(WaitTimeMs(1), waitMs);
     EXPECT_EQ("PrioritizedBucket(Bucket(BucketSpace(0x0000000000000001), BucketId(0x4000000000000001)), pri HIGH)\n",
-              _priorityDb->toString());
+              _priority_db.toString());
 }
 
 }