storage_service: merge_topology_snapshot: co_await to_mutation_gently

Perevent stalls from "unpacking" of large
canonical mutations seen with test_add_many_nodes_under_load
when called from `group0_state_machine::transfer_snapshot`:

```
  ++[1#1/44 14%] addr=0x395b2f total=569 count=6 avg=95: ?? ??:0
  | ++[2#1/2 56%] addr=0x3991e3 total=321 count=4 avg=80: ?? ??:0
  | ++          - addr=0x1587159:
  | |             std::__new_allocator<seastar::basic_sstring<signed char, unsigned int, 31u, false> >::allocate at /usr/bin/../lib/gcc/x86_64-redhat-linux/13/../../../../include/c++/13/bits/new_allocator.h:147
  | |             (inlined by) std::allocator<seastar::basic_sstring<signed char, unsigned int, 31u, false> >::allocate at /usr/bin/../lib/gcc/x86_64-redhat-linux/13/../../../../include/c++/13/bits/allocator.h:198
  | |             (inlined by) std::allocator_traits<std::allocator<seastar::basic_sstring<signed char, unsigned int, 31u, false> > >::allocate at /usr/bin/../lib/gcc/x86_64-redhat-linux/13/../../../../include/c++/13/bits/alloc_traits.h:482
  | |             (inlined by) std::_Vector_base<seastar::basic_sstring<signed char, unsigned int, 31u, false>, std::allocator<seastar::basic_sstring<signed char, unsigned int, 31u, false> > >::_M_allocate at /usr/bin/../lib/gcc/x86_64-redhat-linux/13/../../../../include/c++/13/bits/stl_vector.h:378
  | |             (inlined by) std::vector<seastar::basic_sstring<signed char, unsigned int, 31u, false>, std::allocator<seastar::basic_sstring<signed char, unsigned int, 31u, false> > >::reserve at /usr/bin/../lib/gcc/x86_64-redhat-linux/13/../../../../include/c++/13/bits/vector.tcc:79
  | |             (inlined by) ser::idl::serializers::internal::vector_serializer<std::vector<seastar::basic_sstring<signed char, unsigned int, 31u, false>, std::allocator<seastar::basic_sstring<signed char, unsigned int, 31u, false> > > >::read<seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator> > at ././serializer_impl.hh:226
  | |             (inlined by) ser::deserialize<std::vector<seastar::basic_sstring<signed char, unsigned int, 31u, false>, std::allocator<seastar::basic_sstring<signed char, unsigned int, 31u, false> > >, seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator> > at ././serializer.hh:264
  | |             (inlined by) ser::serializer<clustering_key_prefix>::read<seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator> >(seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator>&)::{lambda(auto:1&)#1}::operator()<seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator> > at ./build/dev/gen/idl/keys.dist.impl.hh:31
  | ++          - addr=0x1587085:
  | |             seastar::with_serialized_stream<seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator>, ser::serializer<clustering_key_prefix>::read<seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator> >(seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator>&)::{lambda(auto:1&)#1}, void, void> at ././seastar/include/seastar/core/simple-stream.hh:646
  | |             (inlined by) ser::serializer<clustering_key_prefix>::read<seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator> > at ./build/dev/gen/idl/keys.dist.impl.hh:28
  | |             (inlined by) ser::deserialize<clustering_key_prefix, seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator> > at ././serializer.hh:264
  | |             (inlined by) ser::deletable_row_view::key() const::{lambda(auto:1&)#1}::operator()<seastar::fragmented_memory_input_stream<bytes_ostream::fragment_iterator> const> at ./build/dev/gen/idl/mutation.dist.impl.hh:1268
  | | ++[3#1/1 100%] addr=0x15865a3 total=577 count=7 avg=82:
  | | |              seastar::memory_input_stream<bytes_ostream::fragment_iterator>::with_stream<ser::deletable_row_view::key() const::{lambda(auto:1&)#1}> at ././seastar/include/seastar/core/simple-stream.hh:491
  | | |              (inlined by) seastar::with_serialized_stream<seastar::memory_input_stream<bytes_ostream::fragment_iterator> const, ser::deletable_row_view::key() const::{lambda(auto:1&)#1}, void> at ././seastar/include/seastar/core/simple-stream.hh:639
  | | |              (inlined by) ser::deletable_row_view::key at ./build/dev/gen/idl/mutation.dist.impl.hh:1264
  | |   ++[4#1/1 100%] addr=0x157cf27 total=643 count=8 avg=80:
  | |   |              mutation_partition_view::do_accept<partition_builder> at ./mutation/mutation_partition_view.cc:212
  | |   ++           - addr=0x1516cac:
  | |   |              mutation_partition::apply at ./mutation/mutation_partition.cc:497
  | |     ++[5#1/1 100%] addr=0x14e4433 total=1765 count=22 avg=80:
  | |     |              canonical_mutation::to_mutation at ./mutation/canonical_mutation.cc:60
  | |       ++[6#1/2 98%] addr=0x2452a60 total=1732 count=21 avg=82:
  | |       |             service::storage_service::merge_topology_snapshot at ./service/storage_service.cc:761
  | |       ++          - addr=0x2858782:
  | |       |             service::group0_state_machine::transfer_snapshot at ./service/raft/group0_state_machine.cc:303
```

Signed-off-by: Benny Halevy <[email protected]>

service/storage_service.cc

@@ -40,6 +40,7 @@
 #include "replica/tablet_mutation_builder.hh"
 #include <seastar/core/smp.hh>
 #include "mutation/canonical_mutation.hh"
+#include "mutation/async_utils.hh"
 #include "seastar/core/on_internal_error.hh"
 #include "service/raft/group0_state_machine.hh"
 #include "service/raft/raft_group0_client.hh"
@@ -786,7 +787,7 @@ future<> storage_service::merge_topology_snapshot(raft_snapshot snp) {
             const auto max_size = _db.local().schema_commitlog()->max_record_size() / 2;
             for (auto i = it; i != snp.mutations.end(); i++) {
                 const auto& m = *i;
-                auto mut = m.to_mutation(s);
+                auto mut = co_await to_mutation_gently(m, s);
                 if (m.representation().size() <= max_size) {
                     muts_to_apply.push_back(std::move(mut));
                 } else {