service: migration_manager: allow using MIGRATION_REQUEST verb to fet…

…ch group 0 history table

The MIGRATION_REQUEST verb is currently used to pull the contents of
schema tables (in the form of mutations) when nodes synchronize schemas.
We will (ab)use the verb to fetch additional data, such as the contents
of the group 0 history table, for purposes of group 0 snapshot transfer.

We extend `schema_pull_options` with a flag specifying that the puller
requests the additional data associated with group 0 snapshots. This
flag is `false` by default, so existing schema pulls will do what they
did before. If the flag is `true`, the migration request handler will
include the contents of group 0 history table.

Note that if a request is set with the flag set to `true`, that means
the entire cluster must have enabled the Raft feature, which also means
that the handler knows of the flag.

idl/messaging_service.idl.hh

@@ -10,6 +10,7 @@ namespace netw {
 
 struct schema_pull_options {
     bool remote_supports_canonical_mutation_retval;
+    bool group0_snapshot_transfer [[version 4.7]] = false;
 };
 
 } // namespace netw

message/messaging_service.hh

@@ -171,6 +171,11 @@ struct serializer {};
 
 struct schema_pull_options {
     bool remote_supports_canonical_mutation_retval = true;
+
+    // We (ab)use `MIGRATION_REQUEST` verb to transfer raft group 0 snapshots,
+    // which contain additional data (besides schema tables mutations).
+    // When used inside group 0 snapshot transfer, this is `true`.
+    bool group0_snapshot_transfer = false;
 };
 
 class messaging_service : public seastar::async_sharded_service<messaging_service>, public peering_sharded_service<messaging_service> {

service/migration_manager.cc

@@ -130,6 +130,15 @@ void migration_manager::init_messaging_service()
         auto features = self._feat.cluster_schema_features();
         auto& proxy = self._storage_proxy.container();
         auto cm = co_await db::schema_tables::convert_schema_to_mutations(proxy, features);
+        if (self._raft_gr.is_enabled() && options->group0_snapshot_transfer) {
+            // if `group0_snapshot_transfer` is `true`, the sender must also understand canonical mutations
+            // (`group0_snapshot_transfer` was added more recently).
+            if (!cm_retval_supported) {
+                on_internal_error(mlogger,
+                    "migration request handler: group0 snapshot transfer requested, but canonical mutations not supported");
+            }
+            cm.emplace_back(co_await db::system_keyspace::get_group0_history(proxy));
+        }
         if (cm_retval_supported) {
             co_return rpc::tuple(std::vector<frozen_mutation>{}, std::move(cm));
         }