repair: throw abort_requested_exception when abort is requested

If abort is requsted during bootstrap then a node should exit normally.
To achieve so, abort_requested_exception should be thrown as main
handles it gracefully.

In data_sync_repair_task_impl::run exceptions from all shards are
wrapped together into std::runtime_exception and so they aren't
handled as they are supposed to.

Throw abort_requested_exception when shutdown was requested.
Throw abort_requested_exception also if repair::task_manager_module::is_aborted,
so that force_terminate_all_repair_sessions acts the same regardless
the state of the repair.

To maintain consistency do the same for user_requested_repair_task_impl.

Fixes: scylladb#15710.

repair/repair.cc

@@ -1272,7 +1272,7 @@ future<> repair::user_requested_repair_task_impl::run() {
         });
 
         if (rs.get_repair_module().is_aborted(id.uuid())) {
-            throw std::runtime_error("aborted by user request");
+            throw abort_requested_exception();
         }
 
         auto ranges_parallelism = _ranges_parallelism;
@@ -1304,8 +1304,9 @@ future<> repair::user_requested_repair_task_impl::run() {
             }
             return make_ready_future<>();
         }).get();
-    }).handle_exception([id] (std::exception_ptr ep) {
+    }).handle_exception([id, &rs] (std::exception_ptr ep) {
         rlogger.warn("repair[{}]: user-requested repair failed: {}", id.uuid(), ep);
+        rs.get_repair_module().check_in_shutdown();
         return make_exception_future<>(ep);
     });
 }
@@ -1390,7 +1391,7 @@ future<> repair::data_sync_repair_task_impl::run() {
         std::vector<future<>> repair_results;
         repair_results.reserve(smp::count);
         if (rs.get_repair_module().is_aborted(id.uuid())) {
-            throw std::runtime_error("aborted by user request");
+            throw abort_requested_exception();
         }
         for (auto shard : boost::irange(unsigned(0), smp::count)) {
             auto f = rs.container().invoke_on(shard, [keyspace, table_ids, id, ranges, neighbors, reason, germs, parent_data = get_repair_uniq_id().task_info] (repair_service& local_repair) mutable -> future<> {
@@ -1428,12 +1429,13 @@ future<> repair::data_sync_repair_task_impl::run() {
         }).get();
     }).then([id, keyspace] {
         rlogger.info("repair[{}]: sync data for keyspace={}, status=succeeded", id.uuid(), keyspace);
-    }).handle_exception([&db, id, keyspace] (std::exception_ptr ep) {
+    }).handle_exception([&db, id, keyspace, &rs] (std::exception_ptr ep) {
         if (!db.has_keyspace(keyspace)) {
             rlogger.warn("repair[{}]: sync data for keyspace={}, status=failed: keyspace does not exist any more, ignoring it, {}", id.uuid(), keyspace, ep);
             return make_ready_future<>();
         }
         rlogger.warn("repair[{}]: sync data for keyspace={}, status=failed: {}", id.uuid(), keyspace,  ep);
+        rs.get_repair_module().check_in_shutdown();
         return make_exception_future<>(ep);
     });
 }