view_building: introduce task_uuid_generator

With the new `min_alive_uuid` saved in the group0 table,
we need to make sure that all new tasks are created with time uuid
greater than the value saved in `min_alive_uuid`.

This patch introduces the `task_uuid_generator` which ensures that
when we are generating multiple tasks in one group0 command, each task
will have an unique time uuid and each time uuid will be greater than
`min_alive_uuid`.

db/view/view_building_state.cc

@@ -8,6 +8,7 @@
  */
 
 #include "db/view/view_building_state.hh"
+#include "utils/UUID_gen.hh"
 
 namespace db {
 
@@ -128,6 +129,24 @@ std::map<dht::token, std::vector<view_building_task>> view_building_state::colle
     return tasks;
 }
 
+task_uuid_generator::task_uuid_generator(api::timestamp_type base_ts)
+        : _next_ts(base_ts) {}
+
+utils::UUID task_uuid_generator::operator()() {
+    return utils::UUID_gen::get_random_time_UUID_from_micros(
+            std::chrono::microseconds{_next_ts++});
+}
+
+task_uuid_generator view_building_state::make_task_uuid_generator(api::timestamp_type ts) const {
+    if (min_alive_uuid) {
+        auto lower_bound = utils::UUID_gen::micros_timestamp(*min_alive_uuid);
+        if (ts <= lower_bound) {
+            ts = lower_bound + 1;
+        }
+    }
+    return task_uuid_generator{ts};
+}
+
 }
 
 }

db/view/view_building_state.hh

@@ -14,6 +14,7 @@
 #include "db/view/view_build_status.hh"
 #include "locator/host_id.hh"
 #include "locator/tablets.hh"
+#include "mutation/timestamp.hh"
 #include "utils/UUID.hh"
 #include <fmt/base.h>
 #include "schema/schema_fwd.hh"
@@ -64,6 +65,16 @@ struct replica_tasks {
 using base_table_tasks = std::map<locator::tablet_replica, replica_tasks>;
 using building_tasks = std::map<table_id, base_table_tasks>;
 
+// Generates unique timeuuids with strictly increasing microsecond timestamps.
+// Each call to operator() returns a new timeuuid whose timestamp is one
+// microsecond greater than the previous one.
+class task_uuid_generator {
+    api::timestamp_type _next_ts;
+public:
+    explicit task_uuid_generator(api::timestamp_type base_ts);
+    utils::UUID operator()();
+};
+
 // Represents cluster-wide view building state (only for tablet-based views).
 // The state stores all unfinished view building tasks for all tablet-based views
 // and table_id of currently processed base table by view building coordinator.
@@ -82,6 +93,13 @@ struct view_building_state {
     std::vector<std::reference_wrapper<const view_building_task>> get_tasks_for_host(table_id base_id, locator::host_id host) const;
     std::map<dht::token, std::vector<view_building_task>> collect_tasks_by_last_token(table_id base_table_id) const;
     std::map<dht::token, std::vector<view_building_task>> collect_tasks_by_last_token(table_id base_table_id, const locator::tablet_replica& replica) const;
+
+    // Creates a generator that produces unique timeuuids suitable for view
+    // building task IDs. The generated uuids have strictly increasing
+    // microsecond timestamps starting from write_timestamp. If min_alive_uuid
+    // is set, all generated uuids are guaranteed to be greater than
+    // *min_alive_uuid in timeuuid order.
+    task_uuid_generator make_task_uuid_generator(api::timestamp_type write_timestamp) const;
 };
 
 // Represents global state of tablet-based views.

db/view/view_building_worker.cc

@@ -275,11 +275,12 @@ future<> view_building_worker::create_staging_sstable_tasks() {
 
     utils::chunked_vector<canonical_mutation> cmuts;
     auto guard = co_await _group0.client().start_operation(_as);
+    auto uuid_gen = _vb_state_machine.building_state.make_task_uuid_generator(guard.write_timestamp());
     auto my_host_id = _db.get_token_metadata().get_topology().my_host_id();
     for (auto& [table_id, sst_infos]: _sstables_to_register) {
         for (auto& sst_info: sst_infos) {
             view_building_task task {
-                utils::UUID_gen::get_time_UUID(), view_building_task::task_type::process_staging, false,
+                uuid_gen(), view_building_task::task_type::process_staging, false,
                 table_id, ::table_id{}, {my_host_id, sst_info.shard}, sst_info.last_token
             };
             auto mut = co_await _sys_ks.make_view_building_task_mutation(guard.write_timestamp(), task);

service/migration_manager.cc

@@ -793,16 +793,18 @@ static future<> add_view_building_tasks_mutations(storage_proxy& sp, view_ptr vi
 
     auto& db = sp.local_db();
     auto& sys_ks = sp.system_keyspace();
+    auto& vb_sm = sp.view_building_state_machine();
 
     auto base_id = view->view_info()->base_id();
     auto& base_cf = db.find_column_family(base_id);
     auto erm = base_cf.get_effective_replication_map();
     auto& tablet_map = erm->get_token_metadata().tablets().get_tablet_map(base_id);
+    auto uuid_gen = vb_sm.building_state.make_task_uuid_generator(ts);
 
     co_await tablet_map.for_each_tablet([&] (auto tid, const auto& tablet_info) -> future<> {
         auto last_token = tablet_map.get_last_token(tid);
         for (auto& replica: tablet_info.replicas) {
-            auto id = utils::UUID_gen::get_time_UUID();
+            auto id = uuid_gen();
             view_building_task task {
                 id, view_building_task::task_type::build_range, false,
                 base_id, view->id(), replica, last_token