[PATCH 06/12] vinyl: don't add dropped LSM trees to the scheduler during recovery

[Vladimir Davydov <vdavydov.dev@gmail.com>]

During local recovery we may encounter an LSM tree marked as dropped.
This means that the LSM tree was dropped before restart and hence will
be deleted before recovery completion. There's no need to add such trees
to the vinyl scheduler - it looks confusing and can potentially result
in mistakes when the code gets modified.
---
 src/box/vinyl.c        | 42 +++++++++++++++++++++++++++---------------
 src/box/vy_scheduler.c |  2 ++
 2 files changed, 29 insertions(+), 15 deletions(-)

diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index 8028fd2b..5965700b 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -749,18 +749,16 @@ vinyl_index_open(struct index *index)
 		diag_set(SystemError, "can not access vinyl data directory");
 		return -1;
 	}
-	int rc;
 	switch (env->status) {
 	case VINYL_ONLINE:
 		/*
 		 * The recovery is complete, simply
 		 * create a new index.
 		 */
-		rc = vy_lsm_create(lsm);
-		if (rc == 0) {
-			/* Make sure reader threads are up and running. */
-			vy_run_env_enable_coio(&env->run_env);
-		}
+		if (vy_lsm_create(lsm) != 0)
+			return -1;
+		/* Make sure reader threads are up and running. */
+		vy_run_env_enable_coio(&env->run_env);
 		break;
 	case VINYL_INITIAL_RECOVERY_REMOTE:
 	case VINYL_FINAL_RECOVERY_REMOTE:
@@ -769,7 +767,8 @@ vinyl_index_open(struct index *index)
 		 * exist locally, and we should create the
 		 * index directory from scratch.
 		 */
-		rc = vy_lsm_create(lsm);
+		if (vy_lsm_create(lsm) != 0)
+			return -1;
 		break;
 	case VINYL_INITIAL_RECOVERY_LOCAL:
 	case VINYL_FINAL_RECOVERY_LOCAL:
@@ -779,17 +778,30 @@ vinyl_index_open(struct index *index)
 		 * have already been created, so try to load
 		 * the index files from it.
 		 */
-		rc = vy_lsm_recover(lsm, env->recovery, &env->run_env,
-				    vclock_sum(env->recovery_vclock),
-				    env->status == VINYL_INITIAL_RECOVERY_LOCAL,
-				    env->force_recovery);
+		if (vy_lsm_recover(lsm, env->recovery, &env->run_env,
+				   vclock_sum(env->recovery_vclock),
+				   env->status == VINYL_INITIAL_RECOVERY_LOCAL,
+				   env->force_recovery) != 0)
+			return -1;
 		break;
 	default:
 		unreachable();
 	}
-	if (rc == 0)
+	/*
+	 * Add the new LSM tree to the scheduler so that it can
+	 * be dumped and compacted.
+	 *
+	 * Note, during local recovery an LSM tree may be marked
+	 * as dropped, which means that it will be dropped before
+	 * recovery is complete. In this case there's no need in
+	 * letting the scheduler know about it.
+	 */
+	if (!lsm->is_dropped)
 		vy_scheduler_add_lsm(&env->scheduler, lsm);
-	return rc;
+	else
+		assert(env->status == VINYL_INITIAL_RECOVERY_LOCAL ||
+		       env->status == VINYL_FINAL_RECOVERY_LOCAL);
+	return 0;
 }
 
 static void
@@ -911,8 +923,6 @@ vinyl_index_commit_drop(struct index *index, int64_t lsn)
 	struct vy_env *env = vy_env(index->engine);
 	struct vy_lsm *lsm = vy_lsm(index);
 
-	vy_scheduler_remove_lsm(&env->scheduler, lsm);
-
 	/*
 	 * We can't abort here, because the index drop request has
 	 * already been written to WAL. So if we fail to write the
@@ -924,6 +934,8 @@ vinyl_index_commit_drop(struct index *index, int64_t lsn)
 	if (env->status == VINYL_FINAL_RECOVERY_LOCAL && lsm->is_dropped)
 		return;
 
+	vy_scheduler_remove_lsm(&env->scheduler, lsm);
+
 	lsm->is_dropped = true;
 
 	vy_log_tx_begin();
diff --git a/src/box/vy_scheduler.c b/src/box/vy_scheduler.c
index 63ac948f..f431eb24 100644
--- a/src/box/vy_scheduler.c
+++ b/src/box/vy_scheduler.c
@@ -503,6 +503,7 @@ vy_scheduler_destroy(struct vy_scheduler *scheduler)
 void
 vy_scheduler_add_lsm(struct vy_scheduler *scheduler, struct vy_lsm *lsm)
 {
+	assert(!lsm->is_dropped);
 	assert(lsm->in_dump.pos == UINT32_MAX);
 	assert(lsm->in_compaction.pos == UINT32_MAX);
 	vy_dump_heap_insert(&scheduler->dump_heap, &lsm->in_dump);
@@ -513,6 +514,7 @@ vy_scheduler_add_lsm(struct vy_scheduler *scheduler, struct vy_lsm *lsm)
 void
 vy_scheduler_remove_lsm(struct vy_scheduler *scheduler, struct vy_lsm *lsm)
 {
+	assert(!lsm->is_dropped);
 	assert(lsm->in_dump.pos != UINT32_MAX);
 	assert(lsm->in_compaction.pos != UINT32_MAX);
 	vy_dump_heap_delete(&scheduler->dump_heap, &lsm->in_dump);
-- 
2.11.0