messaging: Add RESTORE_TABLET RPC verb

The topology coordinator will need to call this verb against existing
tablet replicas to ask them restore tablet sstables. Here's the RPC verb
to do it.

It now returns an empty restore_result to make it "synchronous" -- the
co_await send_restore_tablets() won't resolve until client call
finishes.

Signed-off-by: Pavel Emelyanov <xemul@scylladb.com>

configure.py

@@ -1464,6 +1464,7 @@ idls = ['idl/gossip_digest.idl.hh',
         'idl/frozen_mutation.idl.hh',
         'idl/reconcilable_result.idl.hh',
         'idl/streaming.idl.hh',
+        'idl/sstables_loader.idl.hh',
         'idl/paging_state.idl.hh',
         'idl/frozen_schema.idl.hh',
         'idl/repair.idl.hh',

idl/CMakeLists.txt

@@ -53,6 +53,7 @@ set(idl_headers
   group0.idl.hh
   hinted_handoff.idl.hh
   sstables.idl.hh
+  sstables_loader.idl.hh
   storage_proxy.idl.hh
   storage_service.idl.hh
   strong_consistency/state_machine.idl.hh

idl/sstables_loader.idl.hh

@@ -0,0 +1,12 @@
+/*
+ * Copyright 2026-present ScyllaDB
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.1
+ */
+
+class restore_result {
+};
+
+verb [[]] restore_tablet (raft::server_id dst_id, locator::global_tablet_id gid) -> restore_result;

message/messaging_service.cc

@@ -24,6 +24,7 @@
 #include "service/storage_service.hh"
 #include "service/qos/service_level_controller.hh"
 #include "streaming/prepare_message.hh"
+#include "sstables_loader.hh"
 #include "gms/gossip_digest_syn.hh"
 #include "gms/gossip_digest_ack.hh"
 #include "gms/gossip_digest_ack2.hh"
@@ -139,6 +140,7 @@
 #include "idl/tasks.dist.impl.hh"
 #include "idl/forward_cql.dist.impl.hh"
 #include "gms/feature_service.hh"
+#include "idl/sstables_loader.dist.impl.hh"
 
 namespace netw {
 
@@ -734,6 +736,7 @@ static constexpr unsigned do_get_rpc_client_idx(messaging_verb verb) {
     case messaging_verb::TABLE_LOAD_STATS:
     case messaging_verb::WORK_ON_VIEW_BUILDING_TASKS:
     case messaging_verb::SNAPSHOT_WITH_TABLETS:
+    case messaging_verb::RESTORE_TABLET:
         return 1;
     case messaging_verb::CLIENT_ID:
     case messaging_verb::MUTATION:

message/messaging_service.hh

@@ -214,7 +214,8 @@ enum class messaging_verb : int32_t {
     RAFT_READ_BARRIER = 85,
     FORWARD_CQL_EXECUTE = 86,
     FORWARD_CQL_PREPARE = 87,
-    LAST = 88,
+    RESTORE_TABLET = 88,
+    LAST = 89,
 };
 
 } // namespace netw

sstables_loader.cc

@@ -33,6 +33,7 @@
 #include "service/storage_service.hh"
 #include "sstables_loader_helpers.hh"
 #include "db/system_distributed_keyspace.hh"
+#include "idl/sstables_loader.dist.hh"
 
 #include "sstables/object_storage_client.hh"
 #include "utils/rjson.hh"
@@ -868,9 +869,20 @@ sstables_loader::sstables_loader(sharded<replica::database>& db,
     , _sched_group(std::move(sg))
 {
     tm.register_module("sstables_loader", _task_manager_module);
+    ser::sstables_loader_rpc_verbs::register_restore_tablet(&_messaging, [this] (raft::server_id dst_id, locator::global_tablet_id gid) -> future<restore_result> {
+        return _ss.local().handle_raft_rpc(dst_id, [&sl = container(), gid] (auto& ss) {
+            return ss.do_tablet_operation(gid, "Restore", [&sl, gid] (locator::tablet_metadata_guard& guard) -> future<service::tablet_operation_result> {
+                co_await sl.local().download_tablet_sstables(gid, guard);
+                co_return service::tablet_operation_empty_result{};
+            }).then([] (auto res) {
+                return make_ready_future<restore_result>();
+            });
+        });
+    });
 }
 
 future<> sstables_loader::stop() {
+    co_await ser::sstables_loader_rpc_verbs::unregister(&_messaging),
     co_await _task_manager_module->stop();
 }
 

sstables_loader.hh

@@ -25,6 +25,8 @@ class database;
 }
 
 struct minimal_sst_info;
+struct restore_result {
+};
 
 namespace sstables { class storage_manager; }