Use tablet_task_info fields for tablet task timing

For tablet_virtual_task, set creation_time to tablet_task_info::request_time
and start_time to tablet_task_info::sched_time, matching the actual semantics
of when the request was created vs when it was scheduled for execution.

Co-authored-by: Deexie <56607372+Deexie@users.noreply.github.com>

.github/CODEOWNERS

@@ -57,7 +57,6 @@ repair/* @tgrabiec @asias
 
 # SCHEMA MANAGEMENT
 db/schema_tables* @tgrabiec
-db/legacy_schema_migrator* @tgrabiec
 service/migration* @tgrabiec
 schema* @tgrabiec
 

SCYLLA_ASSERT_CONVERSION_SUMMARY.md

@@ -1,182 +0,0 @@
-# SCYLLA_ASSERT to scylla_assert() Conversion Summary
-
-## Objective
-
-Replace crash-inducing `SCYLLA_ASSERT` with exception-throwing `scylla_assert()` to prevent cluster-wide crashes and maintain availability.
-
-## What Was Done
-
-### 1. Infrastructure Implementation ✓
-
-Created new `scylla_assert()` macro in `utils/assert.hh`:
-- Based on `on_internal_error()` for exception-based error handling
-- Supports optional custom error messages via variadic arguments
-- Uses `seastar::format()` for string formatting
-- Compatible with C++23 standard (uses `__VA_OPT__`)
-
-**Key difference from SCYLLA_ASSERT:**
-```cpp
-// Old: Crashes the process immediately
-SCYLLA_ASSERT(condition);
-
-// New: Throws exception (or aborts based on config)
-scylla_assert(condition);
-scylla_assert(condition, "custom error message: {}", value);
-```
-
-### 2. Comprehensive Analysis ✓
-
-Analyzed entire codebase to identify safe vs unsafe conversion locations:
-
-**Statistics:**
-- Total SCYLLA_ASSERT usages: ~1307 (including tests)
-- Non-test usages: ~886
-- **Unsafe to convert**: 223 usages (25%)
-  - In noexcept functions: 187 usages across 50 files
-  - In destructors: 36 usages across 25 files
-- **Safe to convert**: ~668 usages (75%)
-- **Converted in this PR**: 112 usages (16.8% of safe conversions)
-
-### 3. Documentation ✓
-
-Created comprehensive documentation:
-
-1. **Conversion Guide** (`docs/dev/scylla_assert_conversion.md`)
-   - Explains safe vs unsafe contexts
-   - Provides conversion strategy
-   - Lists all completed conversions
-   - Includes testing guidance
-
-2. **Unsafe Locations Report** (`docs/dev/unsafe_scylla_assert_locations.md`)
-   - Detailed listing of 223 unsafe locations
-   - Organized by file with line numbers
-   - Separated into noexcept and destructor categories
-
-### 4. Sample Conversions ✓
-
-Converted 112 safe SCYLLA_ASSERT usages across 32 files as demonstration:
-
-| File | Conversions | Context |
-|------|------------|---------|
-| db/large_data_handler.{cc,hh} | 5 | Future-returning functions |
-| db/schema_applier.cc | 1 | Coroutine function |
-| db/system_distributed_keyspace.cc | 1 | Regular function |
-| db/commitlog/commitlog_replayer.cc | 1 | Coroutine function |
-| db/view/row_locking.cc | 2 | Regular function |
-| db/size_estimates_virtual_reader.cc | 1 | Lambda in coroutine |
-| db/corrupt_data_handler.cc | 2 | Lambdas in future-returning function |
-| raft/tracker.cc | 2 | Unreachable code (switch defaults) |
-| service/topology_coordinator.cc | 11 | Coroutine functions (topology operations) |
-| service/storage_service.cc | 28 | Critical node lifecycle operations |
-| sstables/* (22 files) | 58 | SSTable operations (read/write/compress/index) |
-
-All conversions were in **safe contexts** (non-noexcept, non-destructor functions). 3 assertions in storage_service.cc remain as SCYLLA_ASSERT (in noexcept functions).
-
-## Why These Cannot Be Converted
-
-### Unsafe Context #1: noexcept Functions (187 usages)
-
-**Problem**: Throwing from noexcept causes `std::terminate()`, same as crash.
-
-**Example** (from `locator/production_snitch_base.hh`):
-```cpp
-virtual bool prefer_local() const noexcept override {
-    SCYLLA_ASSERT(_backreference != nullptr);  // Cannot convert!
-    return _backreference->prefer_local();
-}
-```
-
-**Solution for these**: Keep as SCYLLA_ASSERT or use `on_fatal_internal_error()`.
-
-### Unsafe Context #2: Destructors (36 usages)
-
-**Problem**: Destructors are implicitly noexcept, throwing causes `std::terminate()`.
-
-**Example** (from `utils/file_lock.cc`):
-```cpp
-~file_lock() noexcept {
-    if (_fd.get() != -1) {
-        SCYLLA_ASSERT(_fd.get() != -1);  // Cannot convert!
-        auto r = ::flock(_fd.get(), LOCK_UN);
-        SCYLLA_ASSERT(r == 0);  // Cannot convert!
-    }
-}
-```
-
-**Solution for these**: Keep as SCYLLA_ASSERT.
-
-## Benefits of scylla_assert()
-
-1. **Prevents Cluster-Wide Crashes**
-   - Exception can be caught and handled gracefully
-   - Failed node doesn't bring down entire cluster
-
-2. **Maintains Availability**
-   - Service can continue with degraded functionality
-   - Better than complete crash
-
-3. **Better Error Reporting**
-   - Includes backtrace via `on_internal_error()`
-   - Supports custom error messages
-   - Configurable abort-on-error for testing
-
-4. **Backward Compatible**
-   - SCYLLA_ASSERT still exists for unsafe contexts
-   - Can be gradually adopted
-
-## Testing
-
-- Created manual test in `test/manual/test_scylla_assert.cc`
-- Verifies passing and failing assertions
-- Tests custom error messages
-- Code review passed with improvements made
-
-## Next Steps (Future Work)
-
-1. **Gradual Conversion**
-   - Convert remaining ~653 safe SCYLLA_ASSERT usages incrementally
-   - Prioritize high-impact code paths first
-
-2. **Review noexcept Functions**
-   - Evaluate if some can be made non-noexcept
-   - Consider using `on_fatal_internal_error()` where appropriate
-
-3. **Integration Testing**
-   - Run full test suite with conversions
-   - Monitor for any unexpected behavior
-   - Validate exception propagation
-
-4. **Automated Analysis Tool**
-   - Create tool to identify safe conversion candidates
-   - Generate conversion patches automatically
-   - Track conversion progress
-
-## Files Modified in This PR
-
-### Core Implementation
-- `utils/assert.hh` - Added scylla_assert() macro
-
-### Conversions
-- `db/large_data_handler.cc`
-- `db/large_data_handler.hh`
-- `db/schema_applier.cc`
-- `db/system_distributed_keyspace.cc`
-- `db/commitlog/commitlog_replayer.cc`
-- `db/view/row_locking.cc`
-- `db/size_estimates_virtual_reader.cc`
-- `db/corrupt_data_handler.cc`
-- `raft/tracker.cc`
-- `service/topology_coordinator.cc`
-- `service/storage_service.cc`
-- `sstables/` (22 files across trie/, mx/, and core sstables)
-
-### Documentation
-- `docs/dev/scylla_assert_conversion.md`
-- `docs/dev/unsafe_scylla_assert_locations.md`
-- `test/manual/test_scylla_assert.cc`
-
-## Conclusion
-
-This PR establishes the infrastructure and methodology for replacing SCYLLA_ASSERT with scylla_assert() to improve cluster availability. The sample conversions demonstrate the approach, while comprehensive documentation enables future work.
-
-**Key Achievement**: Provided a safe path forward for converting 75% (~668) of SCYLLA_ASSERT usages to exception-based assertions, while clearly documenting the 25% (~223) that must remain as crash-inducing assertions due to language constraints. Converted 112 usages as demonstration (16.8% of safe conversions), prioritizing critical files like storage_service.cc (node lifecycle) and all sstables files (data persistence), with ~556 remaining.

alternator/executor.cc

@@ -2223,12 +2223,12 @@ void validate_value(const rjson::value& v, const char* caller) {
 
 // The put_or_delete_item class builds the mutations needed by the PutItem and
 // DeleteItem operations - either as stand-alone commands or part of a list
-// of commands in BatchWriteItems.
+// of commands in BatchWriteItem.
 // put_or_delete_item splits each operation into two stages: Constructing the
 // object parses and validates the user input (throwing exceptions if there
 // are input errors). Later, build() generates the actual mutation, with a
 // specified timestamp. This split is needed because of the peculiar needs of
-// BatchWriteItems and LWT. BatchWriteItems needs all parsing to happen before
+// BatchWriteItem and LWT. BatchWriteItem needs all parsing to happen before
 // any writing happens (if one of the commands has an error, none of the
 // writes should be done). LWT makes it impossible for the parse step to
 // generate "mutation" objects, because the timestamp still isn't known.
@@ -2739,7 +2739,7 @@ future<executor::request_return_type> rmw_operation::execute(service::storage_pr
     auto read_command = needs_read_before_write ?
             previous_item_read_command(proxy, schema(), _ck, selection) :
             nullptr;
-    return proxy.cas(schema(), std::move(*cas_shard), shared_from_this(), read_command, to_partition_ranges(*schema(), _pk),
+    return proxy.cas(schema(), std::move(*cas_shard), *this, read_command, to_partition_ranges(*schema(), _pk),
             {timeout, std::move(permit), client_state, trace_state},
             db::consistency_level::LOCAL_SERIAL, db::consistency_level::LOCAL_QUORUM, timeout, timeout, true, std::move(cdc_opts)).then([this, read_command, &wcu_total] (bool is_applied) mutable {
         if (!is_applied) {
@@ -3026,17 +3026,20 @@ struct primary_key_equal {
 };
 
 // This is a cas_request subclass for applying given put_or_delete_items to
-// one partition using LWT as part as BatchWriteItems. This is a write-only
+// one partition using LWT as part as BatchWriteItem. This is a write-only
 // operation, not needing the previous value of the item (the mutation to be
 // done is known prior to starting the operation). Nevertheless, we want to
 // do this mutation via LWT to ensure that it is serialized with other LWT
 // mutations to the same partition.
+// 
+// The std::vector<put_or_delete_item> must remain alive until the
+// storage_proxy::cas() future is resolved.
 class put_or_delete_item_cas_request : public service::cas_request {
     schema_ptr schema;
-    std::vector<put_or_delete_item> _mutation_builders;
+    const std::vector<put_or_delete_item>& _mutation_builders;
 public:
-    put_or_delete_item_cas_request(schema_ptr s, std::vector<put_or_delete_item>&& b) :
-        schema(std::move(s)), _mutation_builders(std::move(b)) { }
+    put_or_delete_item_cas_request(schema_ptr s, const std::vector<put_or_delete_item>& b) :
+        schema(std::move(s)), _mutation_builders(b) { }
     virtual ~put_or_delete_item_cas_request() = default;
     virtual std::optional<mutation> apply(foreign_ptr<lw_shared_ptr<query::result>> qr, const query::partition_slice& slice, api::timestamp_type ts, cdc::per_request_options& cdc_opts) override {
         std::optional<mutation> ret;
@@ -3052,20 +3055,21 @@ public:
     }
 };
 
-static future<> cas_write(service::storage_proxy& proxy, schema_ptr schema, service::cas_shard cas_shard, dht::decorated_key dk, std::vector<put_or_delete_item>&& mutation_builders,
+static future<> cas_write(service::storage_proxy& proxy, schema_ptr schema, service::cas_shard cas_shard, const dht::decorated_key& dk, const std::vector<put_or_delete_item>& mutation_builders,
         service::client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit) {
     auto timeout = executor::default_timeout();
-    auto op = seastar::make_shared<put_or_delete_item_cas_request>(schema, std::move(mutation_builders));
+    auto op = std::make_unique<put_or_delete_item_cas_request>(schema, mutation_builders);
+    auto* op_ptr = op.get();
     auto cdc_opts = cdc::per_request_options{
         .alternator = true,
         .alternator_streams_increased_compatibility =
                 schema->cdc_options().enabled() && proxy.data_dictionary().get_config().alternator_streams_increased_compatibility(),
     };
-    return proxy.cas(schema, std::move(cas_shard), op, nullptr, to_partition_ranges(dk),
+    return proxy.cas(schema, std::move(cas_shard), *op_ptr, nullptr, to_partition_ranges(dk),
             {timeout, std::move(permit), client_state, trace_state},
             db::consistency_level::LOCAL_SERIAL, db::consistency_level::LOCAL_QUORUM,
-            timeout, timeout, true, std::move(cdc_opts)).discard_result();
-    // We discarded cas()'s future value ("is_applied") because BatchWriteItems
+            timeout, timeout, true, std::move(cdc_opts)).finally([op = std::move(op)]{}).discard_result();
+    // We discarded cas()'s future value ("is_applied") because BatchWriteItem
     // does not need to support conditional updates.
 }
 
@@ -3131,30 +3135,34 @@ static future<> do_batch_write(service::storage_proxy& proxy,
         // Multiple mutations may be destined for the same partition, adding
         // or deleting different items of one partition. Join them together
         // because we can do them in one cas() call.
-        std::unordered_map<schema_decorated_key, std::vector<put_or_delete_item>, schema_decorated_key_hash, schema_decorated_key_equal>
-            key_builders(1, schema_decorated_key_hash{}, schema_decorated_key_equal{});
+        using map_type = std::unordered_map<schema_decorated_key, 
+            std::vector<put_or_delete_item>, 
+            schema_decorated_key_hash, 
+            schema_decorated_key_equal>;
+        auto key_builders = std::make_unique<map_type>(1, schema_decorated_key_hash{}, schema_decorated_key_equal{});
         for (auto& b : mutation_builders) {
             auto dk = dht::decorate_key(*b.first, b.second.pk());
-            auto [it, added] = key_builders.try_emplace(schema_decorated_key{b.first, dk});
+            auto [it, added] = key_builders->try_emplace(schema_decorated_key{b.first, dk});
             it->second.push_back(std::move(b.second));
         }
-        return parallel_for_each(std::move(key_builders), [&proxy, &client_state, &stats, trace_state, ssg, permit = std::move(permit)] (auto& e) {
+        auto* key_builders_ptr = key_builders.get();
+        return parallel_for_each(*key_builders_ptr, [&proxy, &client_state, &stats, trace_state, ssg, permit = std::move(permit)] (const auto& e) {
             stats.write_using_lwt++;
             auto desired_shard = service::cas_shard(*e.first.schema, e.first.dk.token());
             if (desired_shard.this_shard()) {
-                return cas_write(proxy, e.first.schema, std::move(desired_shard), e.first.dk, std::move(e.second), client_state, trace_state, permit);
+                return cas_write(proxy, e.first.schema, std::move(desired_shard), e.first.dk, e.second, client_state, trace_state, permit);
             } else {
                 stats.shard_bounce_for_lwt++;
                 return proxy.container().invoke_on(desired_shard.shard(), ssg,
                             [cs = client_state.move_to_other_shard(),
-                             mb = e.second,
-                             dk = e.first.dk,
+                             &mb = e.second,
+                             &dk = e.first.dk,
                              ks = e.first.schema->ks_name(),
                              cf = e.first.schema->cf_name(),
                              gt =  tracing::global_trace_state_ptr(trace_state),
                              permit = std::move(permit)]
                             (service::storage_proxy& proxy) mutable {
-                    return do_with(cs.get(), [&proxy, mb = std::move(mb), dk = std::move(dk), ks = std::move(ks), cf = std::move(cf),
+                    return do_with(cs.get(), [&proxy, &mb, &dk, ks = std::move(ks), cf = std::move(cf),
                                               trace_state = tracing::trace_state_ptr(gt)]
                                               (service::client_state& client_state) mutable {
                         auto schema = proxy.data_dictionary().find_schema(ks, cf);
@@ -3168,11 +3176,11 @@ static future<> do_batch_write(service::storage_proxy& proxy,
                         //FIXME: Instead of passing empty_service_permit() to the background operation,
                         // the current permit's lifetime should be prolonged, so that it's destructed
                         // only after all background operations are finished as well.
-                        return cas_write(proxy, schema, std::move(cas_shard), dk, std::move(mb), client_state, std::move(trace_state), empty_service_permit());
+                        return cas_write(proxy, schema, std::move(cas_shard), dk, mb, client_state, std::move(trace_state), empty_service_permit());
                     });
                 }).finally([desired_shard = std::move(desired_shard)]{});
             }
-        });
+        }).finally([key_builders = std::move(key_builders)]{});
     }
 }
 

api/api-doc/task_manager.json

@@ -349,9 +349,13 @@
                "type":"long",
                "description":"The shard the task is running on"
             },
+            "creation_time":{
+               "type":"datetime",
+               "description":"The creation time of the task (when it was queued); extracted from the task_id UUID"
+            },
             "start_time":{
                "type":"datetime",
-               "description":"The start time of the task; unspecified (equal to epoch) when state == created"
+               "description":"The start time of the task (when execution began); unspecified (equal to epoch) when state == created"
             },
             "end_time":{
                "type":"datetime",
@@ -398,13 +402,17 @@
                "type":"boolean",
                "description":"Boolean flag indicating whether the task can be aborted"
             },
+            "creation_time":{
+               "type":"datetime",
+               "description":"The creation time of the task (when it was queued); extracted from the task_id UUID"
+            },
             "start_time":{
                "type":"datetime",
-               "description":"The start time of the task"
+               "description":"The start time of the task (when execution began); unspecified (equal to epoch) when state == created"
             },
             "end_time":{
                "type":"datetime",
-               "description":"The end time of the task (unspecified when the task is not completed)"
+               "description":"The end time of the task (when execution completed); unspecified (equal to epoch) when the task is not completed"
             },
             "error":{
                "type":"string",

api/task_manager.cc

@@ -55,6 +55,7 @@ tm::task_status make_status(tasks::task_status status, sharded<gms::gossiper>& g
     res.scope = status.scope;
     res.state = status.state;
     res.is_abortable = bool(status.is_abortable);
+    res.creation_time = get_time(status.creation_time);
     res.start_time = get_time(status.start_time);
     res.end_time = get_time(status.end_time);
     res.error = status.error;
@@ -83,6 +84,7 @@ tm::task_stats make_stats(tasks::task_stats stats) {
     res.table = stats.table;
     res.entity = stats.entity;
     res.shard = stats.shard;
+    res.creation_time = get_time(stats.creation_time);
     res.start_time = get_time(stats.start_time);
     res.end_time = get_time(stats.end_time);;
     return res;

auth/certificate_authenticator.cc

@@ -8,6 +8,7 @@
  */
 
 #include "auth/certificate_authenticator.hh"
+#include "auth/cache.hh"
 
 #include <boost/regex.hpp>
 #include <fmt/ranges.h>
@@ -34,13 +35,14 @@ static const class_registrator<auth::authenticator
     , cql3::query_processor&
     , ::service::raft_group0_client&
     , ::service::migration_manager&
+    , auth::cache&
     , utils::alien_worker&> cert_auth_reg(CERT_AUTH_NAME);
 
 enum class auth::certificate_authenticator::query_source {
     subject, altname
 };
 
-auth::certificate_authenticator::certificate_authenticator(cql3::query_processor& qp, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&)
+auth::certificate_authenticator::certificate_authenticator(cql3::query_processor& qp, ::service::raft_group0_client&, ::service::migration_manager&, auth::cache&, utils::alien_worker&)
     : _queries([&] {
         auto& conf = qp.db().get_config();
         auto queries = conf.auth_certificate_role_queries();

auth/certificate_authenticator.hh

@@ -26,13 +26,15 @@ class raft_group0_client;
 
 namespace auth {
 
+class cache;
+
 extern const std::string_view certificate_authenticator_name;
 
 class certificate_authenticator : public authenticator {
     enum class query_source;
     std::vector<std::pair<query_source, boost::regex>> _queries;
 public:
-    certificate_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&);
+    certificate_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&, utils::alien_worker&);
     ~certificate_authenticator();
 
     future<> start() override;

configure.py

@@ -1062,7 +1062,6 @@ scylla_core = (['message/messaging_service.cc',
                 'db/hints/resource_manager.cc',
                 'db/hints/sync_point.cc',
                 'db/large_data_handler.cc',
-                'db/legacy_schema_migrator.cc',
                 'db/marshal/type_parser.cc',
                 'db/per_partition_rate_limit_options.cc',
                 'db/rate_limiter.cc',

cql3/statements/alter_keyspace_statement.cc

@@ -165,8 +165,7 @@ cql3::statements::alter_keyspace_statement::prepare_schema_mutations(query_proce
 
             service::topology_mutation_builder builder(ts);
             service::topology_request_tracking_mutation_builder rtbuilder{global_request_id, qp.proxy().features().topology_requests_type_column};
-            rtbuilder.set("done", false)
-                     .set("start_time", db_clock::now());
+            rtbuilder.set("done", false);
             if (!qp.proxy().features().topology_global_request_queue) {
                 builder.set_global_topology_request(service::global_topology_request::keyspace_rf_change);
                 builder.set_global_topology_request_id(global_request_id);

cql3/statements/batch_statement.cc

@@ -331,7 +331,7 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::exe
     if (!cl_for_paxos) [[unlikely]] {
         return make_exception_future<shared_ptr<cql_transport::messages::result_message>>(std::move(cl_for_paxos).assume_error());
     }
-    seastar::shared_ptr<cas_request> request;
+    std::unique_ptr<cas_request> request;
     schema_ptr schema;
 
     db::timeout_clock::time_point now = db::timeout_clock::now();
@@ -354,9 +354,9 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::exe
         if (keys.empty()) {
             continue;
         }
-        if (request.get() == nullptr) {
+        if (!request) {
             schema = statement.s;
-            request = seastar::make_shared<cas_request>(schema, std::move(keys));
+            request = std::make_unique<cas_request>(schema, std::move(keys));
         } else if (keys.size() != 1 || keys.front().equal(request->key().front(), dht::ring_position_comparator(*schema)) == false) {
             throw exceptions::invalid_request_exception("BATCH with conditions cannot span multiple partitions");
         }
@@ -366,7 +366,7 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::exe
 
         request->add_row_update(statement, std::move(ranges), std::move(json_cache), statement_options);
     }
-    if (request.get() == nullptr) {
+    if (!request) {
         throw exceptions::invalid_request_exception(format("Unrestricted partition key in a conditional BATCH"));
     }
 
@@ -377,9 +377,10 @@ future<shared_ptr<cql_transport::messages::result_message>> batch_statement::exe
             );
     }
 
-    return qp.proxy().cas(schema, std::move(cas_shard), request, request->read_command(qp), request->key(),
+    auto* request_ptr = request.get();
+    return qp.proxy().cas(schema, std::move(cas_shard), *request_ptr, request->read_command(qp), request->key(),
             {read_timeout, qs.get_permit(), qs.get_client_state(), qs.get_trace_state()},
-            std::move(cl_for_paxos).assume_value(), cl_for_learn, batch_timeout, cas_timeout).then([this, request] (bool is_applied) {
+            std::move(cl_for_paxos).assume_value(), cl_for_learn, batch_timeout, cas_timeout).then([this, request = std::move(request)] (bool is_applied) {
         return request->build_cas_result_set(_metadata, _columns_of_cas_result_set, is_applied);
     });
 }

cql3/statements/modification_statement.cc

@@ -401,7 +401,8 @@ modification_statement::execute_with_condition(query_processor& qp, service::que
                     type.is_update() ? "update" : "deletion"));
     }
 
-    auto request = seastar::make_shared<cas_request>(s, std::move(keys));
+    auto request = std::make_unique<cas_request>(s, std::move(keys));
+    auto* request_ptr = request.get();
     // cas_request can be used for batches as well single statements; Here we have just a single
     // modification in the list of CAS commands, since we're handling single-statement execution.
     request->add_row_update(*this, std::move(ranges), std::move(json_cache), options);
@@ -427,9 +428,9 @@ modification_statement::execute_with_condition(query_processor& qp, service::que
         tablet_info = erm->check_locality(token);
     }
 
-    return qp.proxy().cas(s, std::move(cas_shard), request, request->read_command(qp), request->key(),
+    return qp.proxy().cas(s, std::move(cas_shard), *request_ptr, request->read_command(qp), request->key(),
             {read_timeout, qs.get_permit(), qs.get_client_state(), qs.get_trace_state()},
-            std::move(cl_for_paxos).assume_value(), cl_for_learn, statement_timeout, cas_timeout).then([this, request, tablet_replicas = std::move(tablet_info->tablet_replicas), token_range = tablet_info->token_range] (bool is_applied) {
+            std::move(cl_for_paxos).assume_value(), cl_for_learn, statement_timeout, cas_timeout).then([this, request = std::move(request), tablet_replicas = std::move(tablet_info->tablet_replicas), token_range = tablet_info->token_range] (bool is_applied) {
         auto result = request->build_cas_result_set(_metadata, _columns_of_cas_result_set, is_applied);
         result->add_tablet_info(tablet_replicas, token_range);
         return result;

db/CMakeLists.txt

@@ -10,7 +10,6 @@ target_sources(db
     schema_applier.cc
     schema_tables.cc
     cql_type_parser.cc
-    legacy_schema_migrator.cc
     commitlog/commitlog.cc
     commitlog/commitlog_replayer.cc
     commitlog/commitlog_entry.cc

db/commitlog/commitlog_replayer.cc

@@ -165,7 +165,7 @@ future<> db::commitlog_replayer::impl::init() {
 
 future<db::commitlog_replayer::impl::stats>
 db::commitlog_replayer::impl::recover(const commitlog::descriptor& d, const commitlog::replay_state& rpstate) const {
-    scylla_assert(_column_mappings.local_is_initialized());
+    SCYLLA_ASSERT(_column_mappings.local_is_initialized());
 
     replay_position rp{d};
     auto gp = min_pos(rp.shard_id());

db/corrupt_data_handler.cc

@@ -10,7 +10,6 @@
 #include "reader_concurrency_semaphore.hh"
 #include "replica/database.hh"
 #include "utils/UUID_gen.hh"
-#include "utils/assert.hh"
 
 static logging::logger corrupt_data_logger("corrupt_data");
 
@@ -76,14 +75,14 @@ future<corrupt_data_handler::entry_id> system_table_corrupt_data_handler::do_rec
 
     auto set_cell_raw = [this, &entry_row, &corrupt_data_schema, timestamp] (const char* cell_name, managed_bytes cell_value) {
         auto cdef = corrupt_data_schema->get_column_definition(cell_name);
-        scylla_assert(cdef);
+        SCYLLA_ASSERT(cdef);
 
         entry_row.cells().apply(*cdef, atomic_cell::make_live(*cdef->type, timestamp, cell_value, _entry_ttl));
     }; 
 
     auto set_cell = [this, &entry_row, &corrupt_data_schema, timestamp] (const char* cell_name, data_value cell_value) {
         auto cdef = corrupt_data_schema->get_column_definition(cell_name);
-        scylla_assert(cdef);
+        SCYLLA_ASSERT(cdef);
 
         entry_row.cells().apply(*cdef, atomic_cell::make_live(*cdef->type, timestamp, cell_value.serialize_nonnull(), _entry_ttl));
     };

db/large_data_handler.cc

@@ -39,7 +39,7 @@ large_data_handler::large_data_handler(uint64_t partition_threshold_bytes, uint6
 }
 
 future<large_data_handler::partition_above_threshold> large_data_handler::maybe_record_large_partitions(const sstables::sstable& sst, const sstables::key& key, uint64_t partition_size, uint64_t rows, uint64_t range_tombstones, uint64_t dead_rows) {
-    scylla_assert(running());
+    SCYLLA_ASSERT(running());
     partition_above_threshold above_threshold{partition_size > _partition_threshold_bytes, rows > _rows_count_threshold};
     static_assert(std::is_same_v<decltype(above_threshold.size), bool>);
     _stats.partitions_bigger_than_threshold += above_threshold.size; // increment if true
@@ -83,7 +83,7 @@ sstring large_data_handler::sst_filename(const sstables::sstable& sst) {
 }
 
 future<> large_data_handler::maybe_delete_large_data_entries(sstables::shared_sstable sst) {
-    scylla_assert(running());
+    SCYLLA_ASSERT(running());
     auto schema = sst->get_schema();
     auto filename = sst_filename(*sst);
     using ldt = sstables::large_data_type;
@@ -247,7 +247,7 @@ future<> cql_table_large_data_handler::record_large_rows(const sstables::sstable
 
 future<> cql_table_large_data_handler::delete_large_data_entries(const schema& s, sstring sstable_name, std::string_view large_table_name) const {
     auto sys_ks = _sys_ks.get_permit();
-    scylla_assert(sys_ks);
+    SCYLLA_ASSERT(sys_ks);
     const sstring req =
             seastar::format("DELETE FROM system.{} WHERE keyspace_name = ? AND table_name = ? AND sstable_name = ?",
                     large_table_name);

db/large_data_handler.hh

@@ -80,7 +80,7 @@ public:
 
     future<bool> maybe_record_large_rows(const sstables::sstable& sst, const sstables::key& partition_key,
             const clustering_key_prefix* clustering_key, uint64_t row_size) {
-        scylla_assert(running());
+        SCYLLA_ASSERT(running());
         if (row_size > _row_threshold_bytes) [[unlikely]] {
             return with_sem([&sst, &partition_key, clustering_key, row_size, this] {
                 return record_large_rows(sst, partition_key, clustering_key, row_size);
@@ -100,7 +100,7 @@ public:
 
     future<bool> maybe_record_large_cells(const sstables::sstable& sst, const sstables::key& partition_key,
             const clustering_key_prefix* clustering_key, const column_definition& cdef, uint64_t cell_size, uint64_t collection_elements) {
-        scylla_assert(running());
+        SCYLLA_ASSERT(running());
         if (cell_size > _cell_threshold_bytes || collection_elements > _collection_elements_count_threshold) [[unlikely]] {
             return with_sem([&sst, &partition_key, clustering_key, &cdef, cell_size, collection_elements, this] {
                 return record_large_cells(sst, partition_key, clustering_key, cdef, cell_size, collection_elements);

db/legacy_schema_migrator.cc

@@ -1,602 +0,0 @@
-/*
- * Modified by ScyllaDB
- * Copyright (C) 2017-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: (LicenseRef-ScyllaDB-Source-Available-1.0 and Apache-2.0)
- */
-
-// Since Scylla 2.0, we use system tables whose schemas were introduced in
-// Cassandra 3. If Scylla boots to find a data directory with system tables
-// with older schemas - produced by pre-2.0 Scylla or by pre-3.0 Cassandra,
-// we need to migrate these old tables to the new format.
-//
-// We provide here a function, db::legacy_schema_migrator::migrate(),
-// for a one-time migration from old to new system tables. The function
-// reads old system tables, write them back in the new format, and finally
-// delete the old system tables. Scylla's main should call this function and
-// wait for the returned future, before starting to serve the database.
-
-#include <boost/iterator/filter_iterator.hpp>
-#include <seastar/core/future-util.hh>
-#include <seastar/util/log.hh>
-#include <map>
-#include <unordered_set>
-#include <chrono>
-
-#include "replica/database.hh"
-#include "legacy_schema_migrator.hh"
-#include "system_keyspace.hh"
-#include "schema_tables.hh"
-#include "schema/schema_builder.hh"
-#include "service/storage_proxy.hh"
-#include "utils/rjson.hh"
-#include "cql3/query_processor.hh"
-#include "cql3/untyped_result_set.hh"
-#include "cql3/util.hh"
-#include "cql3/statements/property_definitions.hh"
-
-static seastar::logger mlogger("legacy_schema_migrator");
-
-namespace db {
-namespace legacy_schema_migrator {
-
-// local data carriers
-
-class migrator {
-public:
-    static const std::unordered_set<sstring> legacy_schema_tables;
-
-    migrator(sharded<service::storage_proxy>& sp, sharded<replica::database>& db, sharded<db::system_keyspace>& sys_ks, cql3::query_processor& qp)
-                    : _sp(sp), _db(db), _sys_ks(sys_ks), _qp(qp) {
-    }
-    migrator(migrator&&) = default;
-
-    typedef db_clock::time_point time_point;
-
-    // TODO: we don't support triggers.
-    // this is a placeholder.
-    struct trigger {
-        time_point timestamp;
-        sstring name;
-        std::unordered_map<sstring, sstring> options;
-    };
-
-    struct table {
-        time_point timestamp;
-        schema_ptr metadata;
-        std::vector<trigger> triggers;
-    };
-
-    struct type {
-        time_point timestamp;
-        user_type metadata;
-    };
-
-    struct function {
-        time_point timestamp;
-        sstring ks_name;
-        sstring fn_name;
-        std::vector<sstring> arg_names;
-        std::vector<sstring> arg_types;
-        sstring return_type;
-        bool called_on_null_input;
-        sstring language;
-        sstring body;
-    };
-
-    struct aggregate {
-        time_point timestamp;
-        sstring ks_name;
-        sstring fn_name;
-        std::vector<sstring> arg_names;
-        std::vector<sstring> arg_types;
-        sstring return_type;
-        sstring final_func;
-        sstring initcond;
-        sstring state_func;
-        sstring state_type;
-    };
-
-    struct keyspace {
-        time_point timestamp;
-        sstring name;
-        bool durable_writes;
-        std::map<sstring, sstring> replication_params;
-
-        std::vector<table> tables;
-        std::vector<type> types;
-        std::vector<function> functions;
-        std::vector<aggregate> aggregates;
-    };
-
-    class unsupported_feature : public std::runtime_error {
-    public:
-        using runtime_error::runtime_error;
-    };
-
-    static sstring fmt_query(const char* fmt, const char* table) {
-        return fmt::format(fmt::runtime(fmt), db::system_keyspace::NAME, table);
-    }
-
-    typedef ::shared_ptr<cql3::untyped_result_set> result_set_type;
-    typedef const cql3::untyped_result_set::row row_type;
-
-    future<> read_table(keyspace& dst, sstring cf_name, time_point timestamp) {
-        auto fmt = "SELECT * FROM {}.{} WHERE keyspace_name = ? AND columnfamily_name = ?";
-        auto tq = fmt_query(fmt, db::system_keyspace::legacy::COLUMNFAMILIES);
-        auto cq = fmt_query(fmt, db::system_keyspace::legacy::COLUMNS);
-        auto zq = fmt_query(fmt, db::system_keyspace::legacy::TRIGGERS);
-
-        typedef std::tuple<future<result_set_type>, future<result_set_type>, future<result_set_type>, future<db::schema_tables::legacy::schema_mutations>> result_tuple;
-
-        return when_all(_qp.execute_internal(tq, { dst.name, cf_name }, cql3::query_processor::cache_internal::yes),
-                        _qp.execute_internal(cq, { dst.name, cf_name }, cql3::query_processor::cache_internal::yes),
-                        _qp.execute_internal(zq, { dst.name, cf_name }, cql3::query_processor::cache_internal::yes),
-                        db::schema_tables::legacy::read_table_mutations(_sp, dst.name, cf_name, db::system_keyspace::legacy::column_families()))
-                    .then([&dst, cf_name, timestamp](result_tuple&& t) {
-
-            result_set_type tables = std::get<0>(t).get();
-            result_set_type columns = std::get<1>(t).get();
-            result_set_type triggers = std::get<2>(t).get();
-            db::schema_tables::legacy::schema_mutations sm = std::get<3>(t).get();
-
-            row_type& td = tables->one();
-
-            auto ks_name = td.get_as<sstring>("keyspace_name");
-            auto cf_name = td.get_as<sstring>("columnfamily_name");
-            auto id = table_id(td.get_or("cf_id", generate_legacy_id(ks_name, cf_name).uuid()));
-
-            schema_builder builder(dst.name, cf_name, id);
-
-            builder.with_version(sm.digest());
-
-            cf_type cf = sstring_to_cf_type(td.get_or("type", sstring("standard")));
-            if (cf == cf_type::super) {
-                fail(unimplemented::cause::SUPER);
-            }
-
-            auto comparator = td.get_as<sstring>("comparator");
-            bool is_compound = cell_comparator::check_compound(comparator);
-            builder.set_is_compound(is_compound);
-            cell_comparator::read_collections(builder, comparator);
-
-            bool filter_sparse = false;
-
-            data_type default_validator = {};
-            if (td.has("default_validator")) {
-                default_validator = db::schema_tables::parse_type(td.get_as<sstring>("default_validator"));
-                if (default_validator->is_counter()) {
-                    builder.set_is_counter(true);
-                }
-                builder.set_default_validation_class(default_validator);
-            }
-
-            /*
-             * Determine whether or not the table is *really* dense
-             * We cannot trust is_dense value of true (see CASSANDRA-11502, that fixed the issue for 2.2 only, and not retroactively),
-             * but we can trust is_dense value of false.
-             */
-            auto is_dense = td.get_opt<bool>("is_dense");
-            if (!is_dense || *is_dense) {
-                is_dense = [&] {
-                    /*
-                     * As said above, this method is only here because we need to deal with thrift upgrades.
-                     * Once a CF has been "upgraded", i.e. we've rebuilt and save its CQL3 metadata at least once,
-                     * then we'll have saved the "is_dense" value and will be good to go.
-                     *
-                     * But non-upgraded thrift CF (and pre-7744 CF) will have no value for "is_dense", so we need
-                     * to infer that information without relying on it in that case. And for the most part this is
-                     * easy, a CF that has at least one REGULAR definition is not dense. But the subtlety is that not
-                     * having a REGULAR definition may not mean dense because of CQL3 definitions that have only the
-                     * PRIMARY KEY defined.
-                     *
-                     * So we need to recognize those special case CQL3 table with only a primary key. If we have some
-                     * clustering columns, we're fine as said above. So the only problem is that we cannot decide for
-                     * sure if a CF without REGULAR columns nor CLUSTERING_COLUMN definition is meant to be dense, or if it
-                     * has been created in CQL3 by say:
-                     *    CREATE TABLE test (k int PRIMARY KEY)
-                     * in which case it should not be dense. However, we can limit our margin of error by assuming we are
-                     * in the latter case only if the comparator is exactly CompositeType(UTF8Type).
-                     */
-                    std::optional<column_id> max_cl_idx;
-                    const cql3::untyped_result_set::row * regular = nullptr;
-                    for (auto& row : *columns) {
-                        auto kind_str = row.get_as<sstring>("type");
-                        if (kind_str == "compact_value") {
-                            continue;
-                        }
-
-                        auto kind = db::schema_tables::deserialize_kind(kind_str);
-
-                        if (kind == column_kind::regular_column) {
-                            if (regular != nullptr) {
-                                return false;
-                            }
-                            regular = &row;
-                            continue;
-                        }
-                        if (kind == column_kind::clustering_key) {
-                            max_cl_idx = std::max(column_id(row.get_or("component_index", 0)), max_cl_idx.value_or(column_id()));
-                        }
-                    }
-
-                    auto is_cql3_only_pk_comparator = [](const sstring& comparator) {
-                        if (!cell_comparator::check_compound(comparator)) {
-                            return false;
-                        }
-                        // CMH. We don't have composites, nor a parser for it. This is a simple way of c
-                        // checking the same.
-                        auto comma = comparator.find(',');
-                        if (comma != sstring::npos) {
-                            return false;
-                        }
-                        auto off = comparator.find('(');
-                        auto end = comparator.find(')');
-
-                        return comparator.compare(off, end - off, utf8_type->name()) == 0;
-                    };
-
-                    if (max_cl_idx) {
-                        auto n = std::count(comparator.begin(), comparator.end(), ','); // num comp - 1
-                        return *max_cl_idx == n;
-                    }
-
-                    if (regular) {
-                        return false;
-                    }
-
-                    return !is_cql3_only_pk_comparator(comparator);
-
-                }();
-
-                // now, if switched to sparse, remove redundant compact_value column and the last clustering column,
-                // directly copying CASSANDRA-11502 logic. See CASSANDRA-11315.
-
-                filter_sparse = !*is_dense;
-            }
-            builder.set_is_dense(*is_dense);
-
-            auto is_cql = !*is_dense && is_compound;
-            auto is_static_compact = !*is_dense && !is_compound;
-
-            // org.apache.cassandra.schema.LegacySchemaMigrator#isEmptyCompactValueColumn
-            auto is_empty_compact_value = [](const cql3::untyped_result_set::row& column_row) {
-                auto kind_str = column_row.get_as<sstring>("type");
-                // Cassandra only checks for "compact_value", but Scylla generates "regular" instead (#2586)
-                return (kind_str == "compact_value" || kind_str == "regular")
-                       && column_row.get_as<sstring>("column_name").empty();
-            };
-
-            for (auto& row : *columns) {
-                auto kind_str = row.get_as<sstring>("type");
-                auto kind = db::schema_tables::deserialize_kind(kind_str);
-                auto component_index = kind > column_kind::clustering_key ? 0 : column_id(row.get_or("component_index", 0));
-                auto name = row.get_or<sstring>("column_name", sstring());
-                auto validator = db::schema_tables::parse_type(row.get_as<sstring>("validator"));
-
-                if (is_empty_compact_value(row)) {
-                    continue;
-                }
-
-                if (filter_sparse) {
-                    if (kind_str == "compact_value") {
-                        continue;
-                    }
-                    if (kind == column_kind::clustering_key) {
-                        if (cf == cf_type::super && component_index != 0) {
-                            continue;
-                        }
-                        if (cf != cf_type::super && !is_compound) {
-                            continue;
-                        }
-                    }
-                }
-
-                std::optional<index_metadata_kind> index_kind;
-                sstring index_name;
-                index_options_map options;
-                if (row.has("index_type")) {
-                    index_kind = schema_tables::deserialize_index_kind(row.get_as<sstring>("index_type"));
-                }
-                if (row.has("index_name")) {
-                    index_name = row.get_as<sstring>("index_name");
-                }
-                if (row.has("index_options")) {
-                    sstring index_options_str = row.get_as<sstring>("index_options");
-                    options = rjson::parse_to_map<index_options_map>(std::string_view(index_options_str));
-                    sstring type;
-                    auto i = options.find("index_keys");
-                    if (i != options.end()) {
-                        options.erase(i);
-                        type = "KEYS";
-                    }
-                    i = options.find("index_keys_and_values");
-                    if (i != options.end()) {
-                        options.erase(i);
-                        type = "KEYS_AND_VALUES";
-                    }
-                    if (type.empty()) {
-                        if (validator->is_collection() && validator->is_multi_cell()) {
-                            type = "FULL";
-                        } else {
-                            type = "VALUES";
-                        }
-                    }
-                    auto column = cql3::util::maybe_quote(name);
-                    options["target"] = validator->is_collection()
-                                    ? type + "(" + column + ")"
-                                    : column;
-                }
-                if (index_kind) {
-                    // Origin assumes index_name is always set, so let's do the same
-                    builder.with_index(index_metadata(index_name, options, *index_kind, index_metadata::is_local_index::no));
-                }
-
-                data_type column_name_type = [&] {
-                    if (is_static_compact && kind == column_kind::regular_column) {
-                        return db::schema_tables::parse_type(comparator);
-                    }
-                    return utf8_type;
-                }();
-                auto column_name = [&] {
-                    try {
-                        return column_name_type->from_string(name);
-                    } catch (marshal_exception&) {
-                        // #2597: Scylla < 2.0 writes names in serialized form, try to recover
-                        column_name_type->validate(to_bytes_view(name));
-                        return to_bytes(name);
-                    }
-                }();
-                builder.with_column_ordered(column_definition(std::move(column_name), std::move(validator), kind, component_index));
-            }
-
-            if (is_static_compact) {
-                builder.set_regular_column_name_type(db::schema_tables::parse_type(comparator));
-            }
-
-            if (td.has("gc_grace_seconds")) {
-                builder.set_gc_grace_seconds(td.get_as<int32_t>("gc_grace_seconds"));
-            }
-            if (td.has("min_compaction_threshold")) {
-                builder.set_min_compaction_threshold(td.get_as<int32_t>("min_compaction_threshold"));
-            }
-            if (td.has("max_compaction_threshold")) {
-                builder.set_max_compaction_threshold(td.get_as<int32_t>("max_compaction_threshold"));
-            }
-            if (td.has("comment")) {
-                builder.set_comment(td.get_as<sstring>("comment"));
-            }
-            if (td.has("memtable_flush_period_in_ms")) {
-                builder.set_memtable_flush_period(td.get_as<int32_t>("memtable_flush_period_in_ms"));
-            }
-            if (td.has("caching")) {
-                builder.set_caching_options(caching_options::from_sstring(td.get_as<sstring>("caching")));
-            }
-            if (td.has("default_time_to_live")) {
-                builder.set_default_time_to_live(gc_clock::duration(td.get_as<int32_t>("default_time_to_live")));
-            }
-            if (td.has("speculative_retry")) {
-                builder.set_speculative_retry(td.get_as<sstring>("speculative_retry"));
-            }
-            if (td.has("compaction_strategy_class")) {
-                auto strategy = td.get_as<sstring>("compaction_strategy_class");
-                try {
-                    builder.set_compaction_strategy(compaction::compaction_strategy::type(strategy));
-                } catch (const exceptions::configuration_exception& e) {
-                    // If compaction strategy class isn't supported, fallback to incremental.
-                    mlogger.warn("Falling back to incremental compaction strategy after the problem: {}", e.what());
-                    builder.set_compaction_strategy(compaction::compaction_strategy_type::incremental);
-                }
-            }
-            if (td.has("compaction_strategy_options")) {
-                sstring strategy_options_str = td.get_as<sstring>("compaction_strategy_options");
-                builder.set_compaction_strategy_options(rjson::parse_to_map<std::map<sstring, sstring>>(std::string_view(strategy_options_str)));
-            }
-            auto comp_param = td.get_as<sstring>("compression_parameters");
-            compression_parameters cp(rjson::parse_to_map<std::map<sstring, sstring>>(std::string_view(comp_param)));
-            builder.set_compressor_params(cp);
-
-            if (td.has("min_index_interval")) {
-                builder.set_min_index_interval(td.get_as<int32_t>("min_index_interval"));
-            } else if (td.has("index_interval")) { // compatibility
-                builder.set_min_index_interval(td.get_as<int32_t>("index_interval"));
-            }
-            if (td.has("max_index_interval")) {
-                builder.set_max_index_interval(td.get_as<int32_t>("max_index_interval"));
-            }
-            if (td.has("bloom_filter_fp_chance")) {
-                builder.set_bloom_filter_fp_chance(td.get_as<double>("bloom_filter_fp_chance"));
-            } else {
-                builder.set_bloom_filter_fp_chance(builder.get_bloom_filter_fp_chance());
-            }
-            if (td.has("dropped_columns")) {
-                auto map = td.get_map<sstring, int64_t>("dropped_columns");
-                for (auto&& e : map) {
-                    builder.without_column(e.first, api::timestamp_type(e.second));
-                };
-            }
-
-            // ignore version. we're transient
-            if (!triggers->empty()) {
-                throw unsupported_feature("triggers");
-            }
-
-            dst.tables.emplace_back(table{timestamp, builder.build() });
-        });
-    }
-
-    future<> read_tables(keyspace& dst) {
-        auto query = fmt_query("SELECT columnfamily_name, writeTime(type) AS timestamp FROM {}.{} WHERE keyspace_name = ?",
-                        db::system_keyspace::legacy::COLUMNFAMILIES);
-        return _qp.execute_internal(query, {dst.name}, cql3::query_processor::cache_internal::yes).then([this, &dst](result_set_type result) {
-            return parallel_for_each(*result, [this, &dst](row_type& row) {
-                return read_table(dst, row.get_as<sstring>("columnfamily_name"), row.get_as<time_point>("timestamp"));
-            }).finally([result] {});
-        });
-    }
-
-    future<time_point> read_type_timestamp(keyspace& dst, sstring type_name) {
-        // TODO: Unfortunately there is not a single REGULAR column in system.schema_usertypes, so annoyingly we cannot
-        // use the writeTime() CQL function, and must resort to a lower level.
-        // Origin digs up the actual cells of target partition and gets timestamp from there.
-        // We should do the same, but g-dam that's messy. Lets give back dung value for now.
-        return make_ready_future<time_point>(dst.timestamp);
-    }
-
-    future<> read_types(keyspace& dst) {
-        auto query = fmt_query("SELECT * FROM {}.{} WHERE keyspace_name = ?", db::system_keyspace::legacy::USERTYPES);
-        return _qp.execute_internal(query, {dst.name}, cql3::query_processor::cache_internal::yes).then([this, &dst](result_set_type result) {
-            return parallel_for_each(*result, [this, &dst](row_type& row) {
-                auto name = row.get_blob_unfragmented("type_name");
-                auto columns = row.get_list<bytes>("field_names");
-                auto types = row.get_list<sstring>("field_types");
-                std::vector<data_type> field_types;
-                for (auto&& value : types) {
-                    field_types.emplace_back(db::schema_tables::parse_type(value));
-                }
-                auto ut = user_type_impl::get_instance(dst.name, name, columns, field_types, false);
-                return read_type_timestamp(dst, value_cast<sstring>(utf8_type->deserialize(name))).then([ut = std::move(ut), &dst](time_point timestamp) {
-                    dst.types.emplace_back(type{timestamp, ut});
-                });
-            }).finally([result] {});
-        });
-    }
-
-    future<> read_functions(keyspace& dst) {
-        auto query = fmt_query("SELECT * FROM {}.{} WHERE keyspace_name = ?", db::system_keyspace::legacy::FUNCTIONS);
-        return _qp.execute_internal(query, {dst.name}, cql3::query_processor::cache_internal::yes).then([](result_set_type result) {
-            if (!result->empty()) {
-                throw unsupported_feature("functions");
-            }
-        });
-    }
-
-    future<> read_aggregates(keyspace& dst) {
-        auto query = fmt_query("SELECT * FROM {}.{} WHERE keyspace_name = ?", db::system_keyspace::legacy::AGGREGATES);
-        return _qp.execute_internal(query, {dst.name}, cql3::query_processor::cache_internal::yes).then([](result_set_type result) {
-            if (!result->empty()) {
-                throw unsupported_feature("aggregates");
-            }
-        });
-    }
-
-    future<keyspace> read_keyspace(sstring ks_name, bool durable_writes, sstring strategy_class, sstring strategy_options, time_point timestamp) {
-        auto map = rjson::parse_to_map<std::map<sstring, sstring>>(std::string_view(strategy_options));
-        map.emplace("class", std::move(strategy_class));
-        auto ks = ::make_lw_shared<keyspace>(keyspace{timestamp, std::move(ks_name), durable_writes, std::move(map) });
-
-        return read_tables(*ks).then([this, ks] {
-            //Collection<Type> types = readTypes(keyspaceName);
-            return read_types(*ks);
-        }).then([this, ks] {
-            return read_functions(*ks);
-        }).then([this, ks] {
-            return read_aggregates(*ks);
-        }).then([ks] {
-            return make_ready_future<keyspace>(std::move(*ks));
-        });
-    }
-
-    future<> read_all_keyspaces() {
-        static auto ks_filter = [](row_type& row) {
-            auto ks_name = row.get_as<sstring>("keyspace_name");
-            return ks_name != db::system_keyspace::NAME && ks_name != db::schema_tables::v3::NAME;
-        };
-
-        auto query = fmt_query("SELECT keyspace_name, durable_writes, strategy_options, strategy_class, writeTime(durable_writes) AS timestamp FROM {}.{}",
-                        db::system_keyspace::legacy::KEYSPACES);
-
-        return _qp.execute_internal(query, cql3::query_processor::cache_internal::yes).then([this](result_set_type result) {
-            auto i = boost::make_filter_iterator(ks_filter, result->begin(), result->end());
-            auto e = boost::make_filter_iterator(ks_filter, result->end(), result->end());
-            return parallel_for_each(i, e, [this](row_type& row) {
-                return read_keyspace(row.get_as<sstring>("keyspace_name")
-                                , row.get_as<bool>("durable_writes")
-                                , row.get_as<sstring>("strategy_class")
-                                , row.get_as<sstring>("strategy_options")
-                                , row.get_as<db_clock::time_point>("timestamp")
-                                ).then([this](keyspace ks) {
-                    _keyspaces.emplace_back(std::move(ks));
-                   });
-            }).finally([result] {});
-        });
-    }
-
-    future<> drop_legacy_tables() {
-        mlogger.info("Dropping legacy schema tables");
-        auto with_snapshot = !_keyspaces.empty();
-        for (const sstring& cfname : legacy_schema_tables) {
-            co_await replica::database::legacy_drop_table_on_all_shards(_db, _sys_ks, db::system_keyspace::NAME, cfname, with_snapshot);
-        }
-    }
-
-    future<> store_keyspaces_in_new_schema_tables() {
-        mlogger.info("Moving {} keyspaces from legacy schema tables to the new schema keyspace ({})",
-                        _keyspaces.size(), db::schema_tables::v3::NAME);
-
-        utils::chunked_vector<mutation> mutations;
-
-        for (auto& ks : _keyspaces) {
-            auto ksm = ::make_lw_shared<keyspace_metadata>(ks.name
-                            , ks.replication_params["class"] // TODO, make ksm like c3?
-                            , cql3::statements::property_definitions::to_extended_map(ks.replication_params)
-                            , std::nullopt
-                            , std::nullopt
-                            , ks.durable_writes);
-
-            // we want separate time stamps for tables/types, so cannot bulk them into the ksm.
-            for (auto&& m : db::schema_tables::make_create_keyspace_mutations(schema_features::full(), ksm, ks.timestamp.time_since_epoch().count(), false)) {
-                mutations.emplace_back(std::move(m));
-            }
-            for (auto& t : ks.tables) {
-                db::schema_tables::add_table_or_view_to_schema_mutation(t.metadata, t.timestamp.time_since_epoch().count(), true, mutations);
-            }
-            for (auto& t : ks.types) {
-                db::schema_tables::add_type_to_schema_mutation(t.metadata, t.timestamp.time_since_epoch().count(), mutations);
-            }
-        }
-        return _qp.proxy().mutate_locally(std::move(mutations), tracing::trace_state_ptr());
-    }
-
-    future<> flush_schemas() {
-        auto& db = _qp.db().real_database().container();
-        return replica::database::flush_tables_on_all_shards(db, db::schema_tables::all_table_infos(schema_features::full()));
-    }
-
-    future<> migrate() {
-        return read_all_keyspaces().then([this]() {
-            // write metadata to the new schema tables
-            return store_keyspaces_in_new_schema_tables()
-                                                .then(std::bind(&migrator::flush_schemas, this))
-                                                .then(std::bind(&migrator::drop_legacy_tables, this))
-                                                .then([] { mlogger.info("Completed migration of legacy schema tables"); });
-        });
-    }
-
-    sharded<service::storage_proxy>& _sp;
-    sharded<replica::database>& _db;
-    sharded<db::system_keyspace>& _sys_ks;
-    cql3::query_processor& _qp;
-    std::vector<keyspace> _keyspaces;
-};
-
-const std::unordered_set<sstring> migrator::legacy_schema_tables = {
-                db::system_keyspace::legacy::KEYSPACES,
-                db::system_keyspace::legacy::COLUMNFAMILIES,
-                db::system_keyspace::legacy::COLUMNS,
-                db::system_keyspace::legacy::TRIGGERS,
-                db::system_keyspace::legacy::USERTYPES,
-                db::system_keyspace::legacy::FUNCTIONS,
-                db::system_keyspace::legacy::AGGREGATES,
-};
-
-}
-}
-
-future<>
-db::legacy_schema_migrator::migrate(sharded<service::storage_proxy>& sp, sharded<replica::database>& db, sharded<db::system_keyspace>& sys_ks, cql3::query_processor& qp) {
-    return do_with(migrator(sp, db, sys_ks, qp), std::bind(&migrator::migrate, std::placeholders::_1));
-}
-

db/legacy_schema_migrator.hh

@@ -1,37 +0,0 @@
-/*
- * Modified by ScyllaDB
- * Copyright (C) 2017-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: (LicenseRef-ScyllaDB-Source-Available-1.0 and Apache-2.0)
- */
-
-#pragma once
-
-#include <seastar/core/future.hh>
-#include <seastar/core/sharded.hh>
-
-#include "seastarx.hh"
-
-namespace replica {
-class database;
-}
-
-namespace cql3 {
-class query_processor;
-}
-
-namespace service {
-class storage_proxy;
-}
-
-namespace db {
-class system_keyspace;
-
-namespace legacy_schema_migrator {
-
-future<> migrate(sharded<service::storage_proxy>&, sharded<replica::database>& db, sharded<db::system_keyspace>& sys_ks, cql3::query_processor&);
-
-}
-}

db/partition_snapshot_row_cursor.hh

@@ -542,6 +542,7 @@ public:
     // Returns the range tombstone for the key range adjacent to the cursor's position from the side of smaller keys.
     // Excludes the range for the row itself. That information is returned by range_tombstone_for_row().
     // It's possible that range_tombstone() is empty and range_tombstone_for_row() is not empty.
+    // Note that this is different from the meaning of rows_entry::range_tombstone(), which includes the row itself.
     tombstone range_tombstone() const { return _range_tombstone; }
 
     // Can be called when cursor is pointing at a row.

db/row_cache.cc

@@ -1287,6 +1287,15 @@ row_cache::row_cache(schema_ptr s, snapshot_source src, cache_tracker& tracker,
     , _partitions(dht::raw_token_less_comparator{})
     , _underlying(src())
     , _snapshot_source(std::move(src))
+    , _update_section(abstract_formatter([this] (fmt::context& ctx) {
+        fmt::format_to(ctx.out(), "cache.update {}.{}", _schema->ks_name(), _schema->cf_name());
+    }))
+    , _populate_section(abstract_formatter([this] (fmt::context& ctx) {
+        fmt::format_to(ctx.out(), "cache.populate {}.{}", _schema->ks_name(), _schema->cf_name());
+    }))
+    , _read_section(abstract_formatter([this] (fmt::context& ctx) {
+        fmt::format_to(ctx.out(), "cache.read {}.{}", _schema->ks_name(), _schema->cf_name());
+    }))
 {
   try {
     with_allocator(_tracker.allocator(), [this, cont] {

db/schema_applier.cc

@@ -1121,7 +1121,7 @@ future<> schema_applier::commit() {
     // Run func first on shard 0
     // to allow "seeding" of the effective_replication_map
     // with a new e_r_m instance.
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     commit_on_shard(sharded_db.local());
     co_await sharded_db.invoke_on_others([this] (replica::database& db) {
         commit_on_shard(db);

db/schema_tables.cc

@@ -404,10 +404,7 @@ const std::unordered_set<table_id>& schema_tables_holding_schema_mutations() {
                 computed_columns(),
                 dropped_columns(),
                 indexes(),
-                scylla_tables(),
-                db::system_keyspace::legacy::column_families(),
-                db::system_keyspace::legacy::columns(),
-                db::system_keyspace::legacy::triggers()}) {
+                scylla_tables()}) {
             SCYLLA_ASSERT(s->clustering_key_size() > 0);
             auto&& first_column_name = s->clustering_column_at(0).name_as_text();
             SCYLLA_ASSERT(first_column_name == "table_name"
@@ -2840,26 +2837,6 @@ void check_no_legacy_secondary_index_mv_schema(replica::database& db, const view
 }
 
 
-namespace legacy {
-
-table_schema_version schema_mutations::digest() const {
-    md5_hasher h;
-    const db::schema_features no_features;
-    db::schema_tables::feed_hash_for_schema_digest(h, _columnfamilies, no_features);
-    db::schema_tables::feed_hash_for_schema_digest(h, _columns, no_features);
-    return table_schema_version(utils::UUID_gen::get_name_UUID(h.finalize()));
-}
-
-future<schema_mutations> read_table_mutations(sharded<service::storage_proxy>& proxy,
-    sstring keyspace_name, sstring table_name, schema_ptr s)
-{
-    mutation cf_m = co_await read_schema_partition_for_table(proxy, s, keyspace_name, table_name);
-    mutation col_m = co_await read_schema_partition_for_table(proxy, db::system_keyspace::legacy::columns(), keyspace_name, table_name);
-    co_return schema_mutations{std::move(cf_m), std::move(col_m)};
-}
-
-} // namespace legacy
-
 static auto GET_COLUMN_MAPPING_QUERY = format("SELECT column_name, clustering_order, column_name_bytes, kind, position, type FROM system.{} WHERE cf_id = ? AND schema_version = ?",
     db::schema_tables::SCYLLA_TABLE_SCHEMA_HISTORY);
 

db/schema_tables.hh

@@ -155,24 +155,6 @@ schema_ptr scylla_table_schema_history();
 const std::unordered_set<table_id>& schema_tables_holding_schema_mutations();
 }
 
-namespace legacy {
-
-class schema_mutations {
-    mutation _columnfamilies;
-    mutation _columns;
-public:
-    schema_mutations(mutation columnfamilies, mutation columns)
-        : _columnfamilies(std::move(columnfamilies))
-        , _columns(std::move(columns))
-    { }
-    table_schema_version digest() const;
-};
-
-future<schema_mutations> read_table_mutations(sharded<service::storage_proxy>& proxy,
-    sstring keyspace_name, sstring table_name, schema_ptr s);
-
-}
-
 struct qualified_name {
     sstring keyspace_name;
     sstring table_name;

db/size_estimates_virtual_reader.cc

@@ -187,7 +187,7 @@ static future<std::vector<token_range>> get_local_ranges(replica::database& db,
         auto ranges = db.get_token_metadata().get_primary_ranges_for(std::move(tokens));
         std::vector<token_range> local_ranges;
         auto to_bytes = [](const std::optional<dht::token_range::bound>& b) {
-            scylla_assert(b);
+            SCYLLA_ASSERT(b);
             return utf8_type->decompose(b->value().to_sstring());
         };
         // We merge the ranges to be compatible with how Cassandra shows it's size estimates table.

db/system_distributed_keyspace.cc

@@ -231,7 +231,7 @@ static schema_ptr get_current_service_levels(data_dictionary::database db) {
 }
 
 static schema_ptr get_updated_service_levels(data_dictionary::database db, bool workload_prioritization_enabled) {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     auto schema = get_current_service_levels(db);
     schema_builder b(schema);
     for (const auto& col : new_service_levels_columns(workload_prioritization_enabled)) {

db/system_keyspace.cc

@@ -847,8 +847,6 @@ schema_ptr system_keyspace::corrupt_data() {
     return corrupt_data;
 }
 
-static constexpr auto schema_gc_grace = std::chrono::duration_cast<std::chrono::seconds>(days(7)).count();
-
 /*static*/ schema_ptr system_keyspace::scylla_local() {
     static thread_local auto scylla_local = [] {
         schema_builder builder(generate_legacy_id(NAME, SCYLLA_LOCAL), NAME, SCYLLA_LOCAL,
@@ -1360,289 +1358,6 @@ schema_ptr system_keyspace::role_permissions() {
     return schema;
 }
 
-schema_ptr system_keyspace::legacy::hints() {
-    static thread_local auto schema = [] {
-        schema_builder builder(generate_legacy_id(NAME, HINTS), NAME, HINTS,
-        // partition key
-        {{"target_id", uuid_type}},
-        // clustering key
-        {{"hint_id", timeuuid_type}, {"message_version", int32_type}},
-        // regular columns
-        {{"mutation", bytes_type}},
-        // static columns
-        {},
-        // regular column name type
-        utf8_type,
-        // comment
-        "*DEPRECATED* hints awaiting delivery"
-       );
-       builder.set_gc_grace_seconds(0);
-       builder.set_compaction_strategy(compaction::compaction_strategy_type::incremental);
-       builder.set_compaction_strategy_options({{"enabled", "false"}});
-       builder.with(schema_builder::compact_storage::yes);
-       builder.with_hash_version();
-       return builder.build();
-    }();
-    return schema;
-}
-
-schema_ptr system_keyspace::legacy::batchlog() {
-    static thread_local auto schema = [] {
-        schema_builder builder(generate_legacy_id(NAME, BATCHLOG), NAME, BATCHLOG,
-        // partition key
-        {{"id", uuid_type}},
-        // clustering key
-        {},
-        // regular columns
-        {{"data", bytes_type}, {"version", int32_type}, {"written_at", timestamp_type}},
-        // static columns
-        {},
-        // regular column name type
-        utf8_type,
-        // comment
-        "*DEPRECATED* batchlog entries"
-       );
-       builder.set_gc_grace_seconds(0);
-       builder.set_compaction_strategy(compaction::compaction_strategy_type::incremental);
-       builder.set_compaction_strategy_options({{"min_threshold", "2"}});
-       builder.with(schema_builder::compact_storage::no);
-       builder.with_hash_version();
-       return builder.build();
-    }();
-    return schema;
-}
-
-schema_ptr system_keyspace::legacy::keyspaces() {
-    static thread_local auto schema = [] {
-        schema_builder builder(generate_legacy_id(NAME, KEYSPACES), NAME, KEYSPACES,
-        // partition key
-        {{"keyspace_name", utf8_type}},
-        // clustering key
-        {},
-        // regular columns
-        {
-         {"durable_writes", boolean_type},
-         {"strategy_class", utf8_type},
-         {"strategy_options", utf8_type}
-        },
-        // static columns
-        {},
-        // regular column name type
-        utf8_type,
-        // comment
-        "*DEPRECATED* keyspace definitions"
-       );
-       builder.set_gc_grace_seconds(schema_gc_grace);
-       builder.with(schema_builder::compact_storage::yes);
-       builder.with_hash_version();
-       return builder.build();
-    }();
-    return schema;
-}
-
-schema_ptr system_keyspace::legacy::column_families() {
-    static thread_local auto schema = [] {
-        schema_builder builder(generate_legacy_id(NAME, COLUMNFAMILIES), NAME, COLUMNFAMILIES,
-        // partition key
-        {{"keyspace_name", utf8_type}},
-        // clustering key
-        {{"columnfamily_name", utf8_type}},
-        // regular columns
-        {
-         {"bloom_filter_fp_chance", double_type},
-         {"caching", utf8_type},
-         {"cf_id", uuid_type},
-         {"comment", utf8_type},
-         {"compaction_strategy_class", utf8_type},
-         {"compaction_strategy_options", utf8_type},
-         {"comparator", utf8_type},
-         {"compression_parameters", utf8_type},
-         {"default_time_to_live", int32_type},
-         {"default_validator", utf8_type},
-         {"dropped_columns",  map_type_impl::get_instance(utf8_type, long_type, true)},
-         {"gc_grace_seconds", int32_type},
-         {"is_dense", boolean_type},
-         {"key_validator", utf8_type},
-         {"max_compaction_threshold", int32_type},
-         {"max_index_interval", int32_type},
-         {"memtable_flush_period_in_ms", int32_type},
-         {"min_compaction_threshold", int32_type},
-         {"min_index_interval", int32_type},
-         {"speculative_retry", utf8_type},
-         {"subcomparator", utf8_type},
-         {"type", utf8_type},
-         // The following 4 columns are only present up until 2.1.8 tables
-         {"key_aliases", utf8_type},
-         {"value_alias", utf8_type},
-         {"column_aliases", utf8_type},
-         {"index_interval", int32_type},},
-        // static columns
-        {},
-        // regular column name type
-        utf8_type,
-        // comment
-        "*DEPRECATED* table definitions"
-       );
-       builder.set_gc_grace_seconds(schema_gc_grace);
-       builder.with(schema_builder::compact_storage::no);
-       builder.with_hash_version();
-       return builder.build();
-    }();
-    return schema;
-}
-
-schema_ptr system_keyspace::legacy::columns() {
-    static thread_local auto schema = [] {
-        schema_builder builder(generate_legacy_id(NAME, COLUMNS), NAME, COLUMNS,
-        // partition key
-        {{"keyspace_name", utf8_type}},
-        // clustering key
-        {{"columnfamily_name", utf8_type}, {"column_name", utf8_type}},
-        // regular columns
-        {
-            {"component_index", int32_type},
-            {"index_name", utf8_type},
-            {"index_options", utf8_type},
-            {"index_type", utf8_type},
-            {"type", utf8_type},
-            {"validator", utf8_type},
-        },
-        // static columns
-        {},
-        // regular column name type
-        utf8_type,
-        // comment
-        "column definitions"
-        );
-        builder.set_gc_grace_seconds(schema_gc_grace);
-        builder.with(schema_builder::compact_storage::no);
-        builder.with_hash_version();
-        return builder.build();
-    }();
-    return schema;
-}
-
-schema_ptr system_keyspace::legacy::triggers() {
-    static thread_local auto schema = [] {
-        schema_builder builder(generate_legacy_id(NAME, TRIGGERS), NAME, TRIGGERS,
-        // partition key
-        {{"keyspace_name", utf8_type}},
-        // clustering key
-        {{"columnfamily_name", utf8_type}, {"trigger_name", utf8_type}},
-        // regular columns
-        {
-            {"trigger_options",  map_type_impl::get_instance(utf8_type, utf8_type, true)},
-        },
-        // static columns
-        {},
-        // regular column name type
-        utf8_type,
-        // comment
-        "trigger definitions"
-        );
-        builder.set_gc_grace_seconds(schema_gc_grace);
-        builder.with(schema_builder::compact_storage::no);
-        builder.with_hash_version();
-        return builder.build();
-    }();
-    return schema;
-}
-
-schema_ptr system_keyspace::legacy::usertypes() {
-    static thread_local auto schema = [] {
-        schema_builder builder(generate_legacy_id(NAME, USERTYPES), NAME, USERTYPES,
-        // partition key
-        {{"keyspace_name", utf8_type}},
-        // clustering key
-        {{"type_name", utf8_type}},
-        // regular columns
-        {
-            {"field_names", list_type_impl::get_instance(utf8_type, true)},
-            {"field_types", list_type_impl::get_instance(utf8_type, true)},
-        },
-        // static columns
-        {},
-        // regular column name type
-        utf8_type,
-        // comment
-        "user defined type definitions"
-        );
-        builder.set_gc_grace_seconds(schema_gc_grace);
-        builder.with(schema_builder::compact_storage::no);
-        builder.with_hash_version();
-        return builder.build();
-    }();
-    return schema;
-}
-
-schema_ptr system_keyspace::legacy::functions() {
-    /**
-     * Note: we have our own "legacy" version of this table (in schema_tables),
-     * but it is (afaik) not used, and differs slightly from the origin one.
-     * This is based on the origin schema, since we're more likely to encounter
-     * installations of that to migrate, rather than our own (if we dont use the table).
-     */
-    static thread_local auto schema = [] {
-        schema_builder builder(generate_legacy_id(NAME, FUNCTIONS), NAME, FUNCTIONS,
-        // partition key
-        {{"keyspace_name", utf8_type}},
-        // clustering key
-        {{"function_name", utf8_type},{"signature", list_type_impl::get_instance(utf8_type, false)}},
-        // regular columns
-        {
-            {"argument_names", list_type_impl::get_instance(utf8_type, true)},
-            {"argument_types", list_type_impl::get_instance(utf8_type, true)},
-            {"body", utf8_type},
-            {"language", utf8_type},
-            {"return_type", utf8_type},
-            {"called_on_null_input", boolean_type},
-        },
-        // static columns
-        {},
-        // regular column name type
-        utf8_type,
-        // comment
-        "*DEPRECATED* user defined type definitions"
-        );
-        builder.set_gc_grace_seconds(schema_gc_grace);
-        builder.with(schema_builder::compact_storage::no);
-        builder.with_hash_version();
-        return builder.build();
-    }();
-    return schema;
-}
-
-schema_ptr system_keyspace::legacy::aggregates() {
-    static thread_local auto schema = [] {
-        schema_builder builder(generate_legacy_id(NAME, AGGREGATES), NAME, AGGREGATES,
-        // partition key
-        {{"keyspace_name", utf8_type}},
-        // clustering key
-        {{"aggregate_name", utf8_type},{"signature", list_type_impl::get_instance(utf8_type, false)}},
-        // regular columns
-        {
-            {"argument_types", list_type_impl::get_instance(utf8_type, true)},
-            {"final_func", utf8_type},
-            {"initcond", bytes_type},
-            {"return_type", utf8_type},
-            {"state_func", utf8_type},
-            {"state_type", utf8_type},
-        },
-        // static columns
-        {},
-        // regular column name type
-        utf8_type,
-        // comment
-        "*DEPRECATED* user defined aggregate definition"
-        );
-        builder.set_gc_grace_seconds(schema_gc_grace);
-        builder.with(schema_builder::compact_storage::no);
-        builder.with_hash_version();
-        return builder.build();
-    }();
-    return schema;
-}
-
 schema_ptr system_keyspace::dicts() {
     static thread_local auto schema = [] {
         auto id = generate_legacy_id(NAME, DICTS);
@@ -2615,13 +2330,6 @@ std::vector<schema_ptr> system_keyspace::all_tables(const db::config& cfg) {
     if (cfg.check_experimental(db::experimental_features_t::feature::KEYSPACE_STORAGE_OPTIONS)) {
         r.insert(r.end(), {sstables_registry()});
     }
-    // legacy schema
-    r.insert(r.end(), {
-                    // TODO: once we migrate hints/batchlog and add converter
-                    // legacy::hints(), legacy::batchlog(),
-                    legacy::keyspaces(), legacy::column_families(),
-                    legacy::columns(), legacy::triggers(), legacy::usertypes(),
-                    legacy::functions(), legacy::aggregates(), });
 
     return r;
 }

db/system_keyspace.hh

@@ -241,28 +241,6 @@ public:
         static schema_ptr cdc_local();
     };
 
-    struct legacy {
-        static constexpr auto HINTS = "hints";
-        static constexpr auto BATCHLOG = "batchlog";
-        static constexpr auto KEYSPACES = "schema_keyspaces";
-        static constexpr auto COLUMNFAMILIES = "schema_columnfamilies";
-        static constexpr auto COLUMNS = "schema_columns";
-        static constexpr auto TRIGGERS = "schema_triggers";
-        static constexpr auto USERTYPES = "schema_usertypes";
-        static constexpr auto FUNCTIONS = "schema_functions";
-        static constexpr auto AGGREGATES = "schema_aggregates";
-
-        static schema_ptr keyspaces();
-        static schema_ptr column_families();
-        static schema_ptr columns();
-        static schema_ptr triggers();
-        static schema_ptr usertypes();
-        static schema_ptr functions();
-        static schema_ptr aggregates();
-        static schema_ptr hints();
-        static schema_ptr batchlog();
-    };
-
     // Partition estimates for a given range of tokens.
     struct range_estimates {
         schema_ptr schema;

db/view/row_locking.cc

@@ -153,14 +153,14 @@ row_locker::unlock(const dht::decorated_key* pk, bool partition_exclusive,
             mylog.error("column_family::local_base_lock_holder::~local_base_lock_holder() can't find lock for partition", *pk);
             return;
         }
-        scylla_assert(&pli->first == pk);
+        SCYLLA_ASSERT(&pli->first == pk);
         if (cpk) {
             auto rli = pli->second._row_locks.find(*cpk);
             if (rli == pli->second._row_locks.end()) {
                 mylog.error("column_family::local_base_lock_holder::~local_base_lock_holder() can't find lock for row", *cpk);
                 return;
             }
-            scylla_assert(&rli->first == cpk);
+            SCYLLA_ASSERT(&rli->first == cpk);
             mylog.debug("releasing {} lock for row {} in partition {}", (row_exclusive ? "exclusive" : "shared"), *cpk, *pk);
             auto& lock = rli->second;
             if (row_exclusive) {

docs/dev/scylla_assert_conversion.md

@@ -1,198 +0,0 @@
-# SCYLLA_ASSERT to scylla_assert() Conversion Guide
-
-## Overview
-
-This document tracks the conversion of `SCYLLA_ASSERT` to the new `scylla_assert()` macro based on `on_internal_error()`. The new macro throws exceptions instead of crashing the process, preventing cluster-wide crashes and loss of availability.
-
-## Status Summary
-
-- **Total SCYLLA_ASSERT usages**: ~1307 (including tests)
-- **Non-test usages**: ~886
-- **Unsafe conversions (noexcept)**: ~187
-- **Unsafe conversions (destructors)**: ~36
-- **Safe conversions possible**: ~668
-- **Converted so far**: 112
-
-## Safe vs Unsafe Contexts
-
-### Safe to Convert ✓
-- Regular functions (non-noexcept)
-- Coroutine functions (returning `future<T>`)
-- Member functions without noexcept specifier
-- Functions where exception propagation is acceptable
-
-### Unsafe to Convert ✗
-1. **noexcept functions** - throwing exceptions from noexcept causes `std::terminate()`
-2. **Destructors** - destructors are implicitly noexcept
-3. **noexcept lambdas and callbacks**
-4. **Code with explicit exception-safety requirements** that cannot handle exceptions
-
-## Files with Unsafe Conversions
-
-### Files with SCYLLA_ASSERT in noexcept contexts (examples)
-
-1. **reader_concurrency_semaphore.cc**
-   - Lines with noexcept functions containing SCYLLA_ASSERT
-   - Must remain as SCYLLA_ASSERT
-
-2. **db/large_data_handler.cc**
-   - Line 86: `maybe_delete_large_data_entries()` - marked noexcept but contains SCYLLA_ASSERT
-   - Analysis shows this is actually safe (not truly noexcept)
-
-3. **db/row_cache.cc**
-   - Multiple SCYLLA_ASSERT usages in noexcept member functions
-
-4. **db/schema_tables.cc**
-   - SCYLLA_ASSERT in noexcept contexts
-
-5. **raft/server.cc**
-   - Multiple noexcept functions with SCYLLA_ASSERT
-
-### Files with SCYLLA_ASSERT in destructors
-
-1. **reader_concurrency_semaphore.cc**
-   - Line 1116: SCYLLA_ASSERT in destructor
-
-2. **api/column_family.cc**
-   - Line 102: SCYLLA_ASSERT in destructor
-
-3. **utils/logalloc.cc**
-   - Line 1991: SCYLLA_ASSERT in destructor
-
-4. **utils/file_lock.cc**
-   - Lines 34, 36: SCYLLA_ASSERT in destructor
-
-5. **utils/disk_space_monitor.cc**
-   - Line 66: SCYLLA_ASSERT in destructor
-
-## Conversion Strategy
-
-### Phase 1: Infrastructure (Completed)
-- Created `scylla_assert()` macro in `utils/assert.hh`
-- Uses `on_internal_error()` for exception-based error handling
-- Supports optional message parameters
-
-### Phase 2: Safe Conversions
-Convert SCYLLA_ASSERT to scylla_assert in contexts where:
-- Function is not noexcept
-- Not in a destructor
-- Exception propagation is safe
-
-### Phase 3: Document Remaining Uses
-For contexts that cannot be converted:
-- Add comments explaining why SCYLLA_ASSERT must remain
-- Consider alternative approaches (e.g., using `on_fatal_internal_error()` in noexcept)
-
-## Converted Files
-
-### Completed Conversions
-
-1. **db/large_data_handler.cc** (3 conversions)
-   - Line 42: `maybe_record_large_partitions()`
-   - Line 86: `maybe_delete_large_data_entries()`
-   - Line 250: `delete_large_data_entries()`
-
-2. **db/large_data_handler.hh** (2 conversions)
-   - Line 83: `maybe_record_large_rows()`
-   - Line 103: `maybe_record_large_cells()`
-
-3. **db/schema_applier.cc** (1 conversion)
-   - Line 1124: `commit()` coroutine
-
-4. **db/system_distributed_keyspace.cc** (1 conversion)
-   - Line 234: `get_updated_service_levels()`
-
-5. **db/commitlog/commitlog_replayer.cc** (1 conversion)
-   - Line 168: `recover()` coroutine
-
-6. **db/view/row_locking.cc** (2 conversions)
-   - Line 156: `unlock()` - partition lock check
-   - Line 163: `unlock()` - row lock check
-
-7. **db/size_estimates_virtual_reader.cc** (1 conversion)
-   - Line 190: Lambda in `get_local_ranges()`
-
-8. **db/corrupt_data_handler.cc** (2 conversions)
-   - Line 78: `set_cell_raw` lambda
-   - Line 85: `set_cell` lambda
-
-9. **raft/tracker.cc** (2 conversions)
-   - Line 49: Switch default case with descriptive error
-   - Line 90: Switch default case with descriptive error
-
-10. **service/topology_coordinator.cc** (11 conversions)
-    - Line 363: Node lookup assertion in `retake_node()`
-    - Line 2313: Bootstrapping state ring check
-    - Line 2362: Replacing state ring check
-    - Line 2365: Normal nodes lookup assertion
-    - Line 2366: Node ring and state validation
-    - Line 3025: Join request ring check
-    - Line 3036: Leave request ring check
-    - Line 3049: Remove request ring check
-    - Line 3061: Replace request ring check
-    - Line 3166: Transition nodes empty check
-    - Line 4016: Barrier validation in `stop()`
-
-11. **service/storage_service.cc** (28 conversions, 3 unsafe kept as SCYLLA_ASSERT)
-    - Lines 603, 691, 857, 901, 969: Core service operations
-    - Lines 1523, 1575, 1844, 2086, 2170, 2195: Bootstrap and join operations
-    - Lines 2319, 2352, 2354: Replacement operations
-    - Lines 3003, 3028, 3228: Cluster join and drain operations
-    - Lines 3995, 4047, 4353: Decommission and removenode operations
-    - Lines 4473, 5787, 5834, 5958: CDC and topology change operations
-    - Lines 6490, 6491: Tablet streaming operations
-    - Line 7512: Join node response handler
-    - **Unsafe (kept as SCYLLA_ASSERT)**: Lines 3398, 5760, 5775 (noexcept functions)
-
-12. **sstables/** (58 conversions across 22 files)
-    - **sstables/trie/bti_node_reader.cc** (6): Node reading operations
-    - **sstables/mx/writer.cc** (6): MX format writing
-    - **sstables/sstable_set.cc** (5): SSTable set management
-    - **sstables/compressor.cc** (5): Compression/decompression
-    - **sstables/trie/trie_writer.hh** (4): Trie writing
-    - **sstables/downsampling.hh** (4): Downsampling operations
-    - **sstables/storage.{cc,hh}** (6): Storage operations
-    - **sstables/sstables_manager.{cc,hh}** (6): SSTable lifecycle management
-    - **sstables/trie/writer_node.{hh,impl.hh}** (4): Trie node writing
-    - **sstables/trie/bti_key_translation.cc** (2): Key translation
-    - **sstables/sstable_directory.cc** (2): Directory management
-    - **sstables/trie/trie_writer.cc** (1): Trie writer implementation
-    - **sstables/trie/trie_traversal.hh** (1): Trie traversal
-    - **sstables/sstables.cc** (1): Core SSTable operations
-    - **sstables/partition_index_cache.hh** (1): Index caching
-    - **sstables/generation_type.hh** (1): Generation management
-    - **sstables/compress.{cc,hh}** (2): Compression utilities
-    - **sstables/exceptions.hh** (1): Comment update
-
-## Testing
-
-### Manual Testing
-Created `test/manual/test_scylla_assert.cc` to verify:
-- Passing assertions succeed
-- Failing assertions throw exceptions
-- Custom messages are properly formatted
-
-### Integration Testing
-- Run existing test suite with converted assertions
-- Verify no regressions in error handling
-- Confirm exception propagation works correctly
-
-## Future Work
-
-1. **Automated Analysis Tool**
-   - Create tool to identify safe vs unsafe conversion contexts
-   - Generate reports of remaining conversions
-
-2. **Gradual Conversion**
-   - Convert additional safe usages incrementally
-   - Monitor for any unexpected issues
-
-3. **noexcept Review**
-   - Review functions marked noexcept that contain SCYLLA_ASSERT
-   - Consider if they should use `on_fatal_internal_error()` instead
-
-## References
-
-- `utils/assert.hh` - Implementation of both SCYLLA_ASSERT and scylla_assert
-- `utils/on_internal_error.hh` - Exception-based error handling infrastructure
-- GitHub Issue: [Link to original issue tracking this work]

docs/dev/task_manager.md

@@ -45,6 +45,22 @@ immediately after it's finished.
 
 A flag which determines if a task can be aborted through API.
 
+# Task timing fields
+
+Tasks have three timing fields that track different stages of their lifecycle:
+
+- `creation_time` - When the task was created/queued. This is extracted from the task's
+  UUID (which is a timeuuid) and represents the moment the task request was submitted.
+- `start_time` - When the task actually began executing. For tasks that are queued, this
+  will be unspecified (equal to epoch) until execution starts. For node operations
+  like decommission, this is set when the request is picked up for execution by the
+  topology coordinator.
+- `end_time` - When the task completed (successfully or with an error). This is
+  unspecified (equal to epoch) until the task finishes.
+
+The difference between `creation_time` and `start_time` represents the time a task
+spent waiting in the queue before execution began.
+
 # Type vs scope vs kind
 
 `type` of a task describes what operation is covered by a task,

docs/dev/unsafe_scylla_assert_locations.md

@@ -1,614 +0,0 @@
-# Unsafe SCYLLA_ASSERT Locations
-
-This document lists specific locations where SCYLLA_ASSERT cannot be safely converted to scylla_assert().
-
-## Summary
-
-- Files with noexcept SCYLLA_ASSERT: 50
-- Files with destructor SCYLLA_ASSERT: 25
-- Total unsafe SCYLLA_ASSERT in noexcept: 187
-- Total unsafe SCYLLA_ASSERT in destructors: 36
-
-## SCYLLA_ASSERT in noexcept Functions
-
-### auth/cache.cc
-
-- Line 118: `SCYLLA_ASSERT(this_shard_id() == 0);`
-
-Total: 1 usages
-
-### db/cache_mutation_reader.hh
-
-- Line 309: `SCYLLA_ASSERT(sr->is_static_row());`
-
-Total: 1 usages
-
-### db/commitlog/commitlog.cc
-
-- Line 531: `SCYLLA_ASSERT(!*this);`
-- Line 544: `SCYLLA_ASSERT(!*this);`
-- Line 662: `SCYLLA_ASSERT(_iter != _end);`
-- Line 1462: `SCYLLA_ASSERT(i->second >= count);`
-
-Total: 4 usages
-
-### db/hints/manager.hh
-
-- Line 167: `SCYLLA_ASSERT(_ep_managers.empty());`
-
-Total: 1 usages
-
-### db/partition_snapshot_row_cursor.hh
-
-- Line 384: `SCYLLA_ASSERT(_latest_it);`
-
-Total: 1 usages
-
-### db/row_cache.cc
-
-- Line 1365: `SCYLLA_ASSERT(it->is_last_dummy());`
-
-Total: 1 usages
-
-### db/schema_tables.cc
-
-- Line 774: `SCYLLA_ASSERT(this_shard_id() == 0);`
-
-Total: 1 usages
-
-### db/view/view.cc
-
-- Line 3623: `SCYLLA_ASSERT(thread::running_in_thread());`
-
-Total: 1 usages
-
-### gms/gossiper.cc
-
-- Line 876: `SCYLLA_ASSERT(ptr->pid == _permit_id);`
-
-Total: 1 usages
-
-### locator/production_snitch_base.hh
-
-- Line 77: `SCYLLA_ASSERT(_backreference != nullptr);`
-- Line 82: `SCYLLA_ASSERT(_backreference != nullptr);`
-- Line 87: `SCYLLA_ASSERT(_backreference != nullptr);`
-
-Total: 3 usages
-
-### locator/topology.cc
-
-- Line 135: `SCYLLA_ASSERT(_shard == this_shard_id());`
-
-Total: 1 usages
-
-### mutation/counters.hh
-
-- Line 314: `SCYLLA_ASSERT(_cell.is_live());`
-- Line 315: `SCYLLA_ASSERT(!_cell.is_counter_update());`
-
-Total: 2 usages
-
-### mutation/mutation_partition_v2.hh
-
-- Line 271: `SCYLLA_ASSERT(s.version() == _schema_version);`
-
-Total: 1 usages
-
-### mutation/partition_version.cc
-
-- Line 364: `SCYLLA_ASSERT(!_snapshot->is_locked());`
-- Line 701: `SCYLLA_ASSERT(!rows.empty());`
-- Line 703: `SCYLLA_ASSERT(last_dummy.is_last_dummy());`
-- Line 746: `SCYLLA_ASSERT(!_snapshot->is_locked());`
-- Line 770: `SCYLLA_ASSERT(at_latest_version());`
-- Line 777: `SCYLLA_ASSERT(at_latest_version());`
-
-Total: 6 usages
-
-### mutation/partition_version.hh
-
-- Line 211: `SCYLLA_ASSERT(_schema);`
-- Line 217: `SCYLLA_ASSERT(_schema);`
-- Line 254: `SCYLLA_ASSERT(!_version->_backref);`
-- Line 282: `SCYLLA_ASSERT(_version);`
-- Line 286: `SCYLLA_ASSERT(_version);`
-- Line 290: `SCYLLA_ASSERT(_version);`
-- Line 294: `SCYLLA_ASSERT(_version);`
-
-Total: 7 usages
-
-### mutation/partition_version_list.hh
-
-- Line 36: `SCYLLA_ASSERT(!_head->is_referenced_from_entry());`
-- Line 42: `SCYLLA_ASSERT(!_tail->is_referenced_from_entry());`
-- Line 70: `SCYLLA_ASSERT(!_head->is_referenced_from_entry());`
-
-Total: 3 usages
-
-### mutation/range_tombstone_list.cc
-
-- Line 412: `SCYLLA_ASSERT (it != rt_list.end());`
-- Line 422: `SCYLLA_ASSERT (it != rt_list.end());`
-
-Total: 2 usages
-
-### raft/server.cc
-
-- Line 1720: `SCYLLA_ASSERT(_non_joint_conf_commit_promise);`
-
-Total: 1 usages
-
-### reader_concurrency_semaphore.cc
-
-- Line 109: `SCYLLA_ASSERT(_permit == o._permit);`
-- Line 432: `SCYLLA_ASSERT(_need_cpu_branches);`
-- Line 455: `SCYLLA_ASSERT(_awaits_branches);`
-- Line 1257: `SCYLLA_ASSERT(!_stopped);`
-- Line 1585: `SCYLLA_ASSERT(_stats.need_cpu_permits);`
-- Line 1587: `SCYLLA_ASSERT(_stats.need_cpu_permits >= _stats.awaits_permits);`
-- Line 1593: `SCYLLA_ASSERT(_stats.need_cpu_permits >= _stats.awaits_permits);`
-- Line 1598: `SCYLLA_ASSERT(_stats.awaits_permits);`
-
-Total: 8 usages
-
-### readers/multishard.cc
-
-- Line 296: `SCYLLA_ASSERT(!_irh);`
-
-Total: 1 usages
-
-### repair/repair.cc
-
-- Line 1073: `SCYLLA_ASSERT(table_names().size() == table_ids.size());`
-
-Total: 1 usages
-
-### replica/database.cc
-
-- Line 3299: `SCYLLA_ASSERT(!_cf_lock.try_write_lock()); // lock should be acquired before the`
-- Line 3304: `SCYLLA_ASSERT(!_cf_lock.try_write_lock()); // lock should be acquired before the`
-
-Total: 2 usages
-
-### replica/database.hh
-
-- Line 1971: `SCYLLA_ASSERT(_user_sstables_manager);`
-- Line 1976: `SCYLLA_ASSERT(_system_sstables_manager);`
-
-Total: 2 usages
-
-### replica/dirty_memory_manager.cc
-
-- Line 67: `SCYLLA_ASSERT(!child->_heap_handle);`
-
-Total: 1 usages
-
-### replica/dirty_memory_manager.hh
-
-- Line 261: `SCYLLA_ASSERT(_shutdown_requested);`
-
-Total: 1 usages
-
-### replica/memtable.cc
-
-- Line 563: `SCYLLA_ASSERT(_mt._flushed_memory <= static_cast<int64_t>(_mt.occupancy().total_`
-- Line 860: `SCYLLA_ASSERT(!reclaiming_enabled());`
-
-Total: 2 usages
-
-### replica/table.cc
-
-- Line 2829: `SCYLLA_ASSERT(!trange.start()->is_inclusive() && trange.end()->is_inclusive());`
-
-Total: 1 usages
-
-### schema/schema.hh
-
-- Line 1022: `SCYLLA_ASSERT(_schema->is_view());`
-
-Total: 1 usages
-
-### schema/schema_registry.cc
-
-- Line 257: `SCYLLA_ASSERT(_state >= state::LOADED);`
-- Line 262: `SCYLLA_ASSERT(_state >= state::LOADED);`
-- Line 329: `SCYLLA_ASSERT(o._cpu_of_origin == current);`
-
-Total: 3 usages
-
-### service/direct_failure_detector/failure_detector.cc
-
-- Line 628: `SCYLLA_ASSERT(alive != endpoint_liveness.marked_alive);`
-
-Total: 1 usages
-
-### service/storage_service.cc
-
-- Line 3398: `SCYLLA_ASSERT(this_shard_id() == 0);`
-- Line 5760: `SCYLLA_ASSERT(this_shard_id() == 0);`
-- Line 5775: `SCYLLA_ASSERT(this_shard_id() == 0);`
-- Line 5787: `SCYLLA_ASSERT(this_shard_id() == 0);`
-
-Total: 4 usages
-
-### sstables/generation_type.hh
-
-- Line 132: `SCYLLA_ASSERT(bool(gen));`
-
-Total: 1 usages
-
-### sstables/partition_index_cache.hh
-
-- Line 62: `SCYLLA_ASSERT(!ready());`
-
-Total: 1 usages
-
-### sstables/sstables_manager.hh
-
-- Line 244: `SCYLLA_ASSERT(_sstables_registry && "sstables_registry is not plugged");`
-
-Total: 1 usages
-
-### sstables/storage.hh
-
-- Line 86: `SCYLLA_ASSERT(false && "Changing directory not implemented");`
-- Line 89: `SCYLLA_ASSERT(false && "Direct links creation not implemented");`
-- Line 92: `SCYLLA_ASSERT(false && "Direct move not implemented");`
-
-Total: 3 usages
-
-### sstables_loader.cc
-
-- Line 735: `SCYLLA_ASSERT(p);`
-
-Total: 1 usages
-
-### tasks/task_manager.cc
-
-- Line 56: `SCYLLA_ASSERT(inserted);`
-- Line 76: `SCYLLA_ASSERT(child->get_status().progress_units == progress_units);`
-- Line 454: `SCYLLA_ASSERT(this_shard_id() == 0);`
-
-Total: 3 usages
-
-### tools/schema_loader.cc
-
-- Line 281: `SCYLLA_ASSERT(p);`
-
-Total: 1 usages
-
-### utils/UUID.hh
-
-- Line 59: `SCYLLA_ASSERT(is_timestamp());`
-
-Total: 1 usages
-
-### utils/bptree.hh
-
-- Line 289: `SCYLLA_ASSERT(n.is_leftmost());`
-- Line 301: `SCYLLA_ASSERT(n.is_rightmost());`
-- Line 343: `SCYLLA_ASSERT(leaf->is_leaf());`
-- Line 434: `SCYLLA_ASSERT(d->attached());`
-- Line 453: `SCYLLA_ASSERT(n._num_keys > 0);`
-- Line 505: `SCYLLA_ASSERT(n->is_leftmost());`
-- Line 511: `SCYLLA_ASSERT(n->is_rightmost());`
-- Line 517: `SCYLLA_ASSERT(n->is_root());`
-- Line 557: `SCYLLA_ASSERT(!is_end());`
-- Line 566: `SCYLLA_ASSERT(!is_end());`
-- Line 613: `SCYLLA_ASSERT(n->_num_keys > 0);`
-- Line 833: `SCYLLA_ASSERT(_left->_num_keys > 0);`
-- Line 926: `SCYLLA_ASSERT(rl == rb);`
-- Line 927: `SCYLLA_ASSERT(rl <= nr);`
-- Line 1037: `SCYLLA_ASSERT(is_leaf());`
-- Line 1042: `SCYLLA_ASSERT(is_leaf());`
-- Line 1047: `SCYLLA_ASSERT(is_leaf());`
-- Line 1052: `SCYLLA_ASSERT(is_leaf());`
-- Line 1062: `SCYLLA_ASSERT(t->_right == this);`
-- Line 1083: `SCYLLA_ASSERT(t->_left == this);`
-- Line 1091: `SCYLLA_ASSERT(t->_right == this);`
-- Line 1103: `SCYLLA_ASSERT(false);`
-- Line 1153: `SCYLLA_ASSERT(i <= _num_keys);`
-- Line 1212: `SCYLLA_ASSERT(off <= _num_keys);`
-- Line 1236: `SCYLLA_ASSERT(from._num_keys > 0);`
-- Line 1389: `SCYLLA_ASSERT(!is_root());`
-- Line 1450: `SCYLLA_ASSERT(_num_keys == NodeSize);`
-- Line 1563: `SCYLLA_ASSERT(_num_keys < NodeSize);`
-- Line 1577: `SCYLLA_ASSERT(i != 0 || left_kid_sorted(k, less));`
-- Line 1647: `SCYLLA_ASSERT(nodes.empty());`
-- Line 1684: `SCYLLA_ASSERT(_num_keys > 0);`
-- Line 1686: `SCYLLA_ASSERT(p._kids[i].n == this);`
-- Line 1788: `SCYLLA_ASSERT(_num_keys == 0);`
-- Line 1789: `SCYLLA_ASSERT(is_root() || !is_leaf() || (get_prev() == this && get_next() == th`
-- Line 1821: `SCYLLA_ASSERT(_parent->_kids[i].n == &other);`
-- Line 1841: `SCYLLA_ASSERT(i <= _num_keys);`
-- Line 1856: `SCYLLA_ASSERT(!_nodes.empty());`
-- Line 1938: `SCYLLA_ASSERT(!attached());`
-- Line 1943: `SCYLLA_ASSERT(attached());`
-
-Total: 39 usages
-
-### utils/cached_file.hh
-
-- Line 104: `SCYLLA_ASSERT(!_use_count);`
-
-Total: 1 usages
-
-### utils/compact-radix-tree.hh
-
-- Line 1026: `SCYLLA_ASSERT(check_capacity(head, ni));`
-- Line 1027: `SCYLLA_ASSERT(!_data.has(ni));`
-- Line 1083: `SCYLLA_ASSERT(next_cap > head._capacity);`
-- Line 1149: `SCYLLA_ASSERT(capacity != 0);`
-- Line 1239: `SCYLLA_ASSERT(i < Size);`
-- Line 1240: `SCYLLA_ASSERT(_idx[i] == unused_node_index);`
-- Line 1470: `SCYLLA_ASSERT(kid != nullptr);`
-- Line 1541: `SCYLLA_ASSERT(ret.first != nullptr);`
-- Line 1555: `SCYLLA_ASSERT(leaf_depth >= depth);`
-- Line 1614: `SCYLLA_ASSERT(n->check_prefix(key, depth));`
-- Line 1850: `SCYLLA_ASSERT(_root.is(nil_root));`
-
-Total: 11 usages
-
-### utils/cross-shard-barrier.hh
-
-- Line 134: `SCYLLA_ASSERT(w.has_value());`
-
-Total: 1 usages
-
-### utils/double-decker.hh
-
-- Line 200: `SCYLLA_ASSERT(!hint.match);`
-- Line 366: `SCYLLA_ASSERT(nb == end._bucket);`
-
-Total: 2 usages
-
-### utils/intrusive-array.hh
-
-- Line 217: `SCYLLA_ASSERT(!is_single_element());`
-- Line 218: `SCYLLA_ASSERT(pos < max_len);`
-- Line 225: `SCYLLA_ASSERT(pos > 0);`
-- Line 238: `SCYLLA_ASSERT(train_len < max_len);`
-- Line 329: `SCYLLA_ASSERT(idx < max_len); // may the force be with us...`
-
-Total: 5 usages
-
-### utils/intrusive_btree.hh
-
-- Line 148: `SCYLLA_ASSERT(to.num_keys == 0);`
-- Line 157: `SCYLLA_ASSERT(!attached());`
-- Line 227: `SCYLLA_ASSERT(n->is_inline());`
-- Line 232: `SCYLLA_ASSERT(n->is_inline());`
-- Line 288: `SCYLLA_ASSERT(n.is_root());`
-- Line 294: `SCYLLA_ASSERT(n.is_leftmost());`
-- Line 302: `SCYLLA_ASSERT(n.is_rightmost());`
-- Line 368: `SCYLLA_ASSERT(_root->is_leaf());`
-- Line 371: `SCYLLA_ASSERT(_inline.empty());`
-- Line 601: `SCYLLA_ASSERT(n->is_leaf());`
-- Line 673: `SCYLLA_ASSERT(!is_end());`
-- Line 674: `SCYLLA_ASSERT(h->attached());`
-- Line 677: `SCYLLA_ASSERT(_idx < cur.n->_base.num_keys);`
-- Line 679: `SCYLLA_ASSERT(_hook->attached());`
-- Line 690: `SCYLLA_ASSERT(!is_end());`
-- Line 764: `SCYLLA_ASSERT(n->num_keys > 0);`
-- Line 994: `SCYLLA_ASSERT(!_it.is_end());`
-- Line 1178: `SCYLLA_ASSERT(is_leaf());`
-- Line 1183: `SCYLLA_ASSERT(is_root());`
-- Line 1261: `SCYLLA_ASSERT(!is_root());`
-- Line 1268: `SCYLLA_ASSERT(p->_base.num_keys > 0 && p->_kids[0] == this);`
-- Line 1275: `SCYLLA_ASSERT(p->_base.num_keys > 0 && p->_kids[p->_base.num_keys] == this);`
-- Line 1286: `SCYLLA_ASSERT(false);`
-- Line 1291: `SCYLLA_ASSERT(!nb->is_inline());`
-- Line 1296: `SCYLLA_ASSERT(!nb->is_inline());`
-- Line 1338: `SCYLLA_ASSERT(_base.num_keys == 0);`
-- Line 1373: `SCYLLA_ASSERT(!(is_leftmost() || is_rightmost()));`
-- Line 1378: `SCYLLA_ASSERT(p->_kids[i] != this);`
-- Line 1396: `SCYLLA_ASSERT(!is_leaf());`
-- Line 1537: `SCYLLA_ASSERT(src != _base.num_keys); // need more keys for the next leaf`
-- Line 1995: `SCYLLA_ASSERT(_parent.n->_base.num_keys > 0);`
-- Line 2135: `SCYLLA_ASSERT(is_leaf());`
-- Line 2144: `SCYLLA_ASSERT(_base.num_keys != 0);`
-- Line 2160: `SCYLLA_ASSERT(_base.num_keys != 0);`
-- Line 2172: `SCYLLA_ASSERT(!empty());`
-- Line 2198: `SCYLLA_ASSERT(leaf == ret->is_leaf());`
-
-Total: 36 usages
-
-### utils/loading_shared_values.hh
-
-- Line 203: `SCYLLA_ASSERT(!_set.size());`
-
-Total: 1 usages
-
-### utils/logalloc.cc
-
-- Line 544: `SCYLLA_ASSERT(!_background_reclaimer);`
-- Line 926: `SCYLLA_ASSERT(idx < _segments.size());`
-- Line 933: `SCYLLA_ASSERT(idx < _segments.size());`
-- Line 957: `SCYLLA_ASSERT(i != _segments.end());`
-- Line 1323: `SCYLLA_ASSERT(_lsa_owned_segments_bitmap.test(idx_from_segment(seg)));`
-- Line 1366: `SCYLLA_ASSERT(desc._region);`
-- Line 1885: `SCYLLA_ASSERT(desc._buf_pointers.empty());`
-- Line 1911: `SCYLLA_ASSERT(&desc == old_ptr->_desc);`
-- Line 2105: `SCYLLA_ASSERT(seg);`
-- Line 2116: `SCYLLA_ASSERT(seg);`
-- Line 2341: `SCYLLA_ASSERT(pool.current_emergency_reserve_goal() >= n_segments);`
-
-Total: 11 usages
-
-### utils/logalloc.hh
-
-- Line 307: `SCYLLA_ASSERT(this_shard_id() == _cpu);`
-
-Total: 1 usages
-
-### utils/reusable_buffer.hh
-
-- Line 60: `SCYLLA_ASSERT(_refcount == 0);`
-
-Total: 1 usages
-
-
-## SCYLLA_ASSERT in Destructors
-
-### api/column_family.cc
-
-- Line 102: `SCYLLA_ASSERT(this_shard_id() == 0);`
-
-Total: 1 usages
-
-### cdc/generation.cc
-
-- Line 846: `SCYLLA_ASSERT(_stopped);`
-
-Total: 1 usages
-
-### cdc/log.cc
-
-- Line 173: `SCYLLA_ASSERT(_stopped);`
-
-Total: 1 usages
-
-### compaction/compaction_manager.cc
-
-- Line 1074: `SCYLLA_ASSERT(_state == state::none || _state == state::stopped);`
-
-Total: 1 usages
-
-### db/hints/internal/hint_endpoint_manager.cc
-
-- Line 188: `SCYLLA_ASSERT(stopped());`
-
-Total: 1 usages
-
-### mutation/partition_version.cc
-
-- Line 347: `SCYLLA_ASSERT(!_snapshot->is_locked());`
-
-Total: 1 usages
-
-### reader_concurrency_semaphore.cc
-
-- Line 1116: `SCYLLA_ASSERT(!_stats.waiters);`
-- Line 1125: `SCYLLA_ASSERT(_inactive_reads.empty() && !_close_readers_gate.get_count() && !_p`
-
-Total: 2 usages
-
-### repair/row_level.cc
-
-- Line 3647: `SCYLLA_ASSERT(_state == state::none || _state == state::stopped);`
-
-Total: 1 usages
-
-### replica/cell_locking.hh
-
-- Line 371: `SCYLLA_ASSERT(_partitions.empty());`
-
-Total: 1 usages
-
-### replica/distributed_loader.cc
-
-- Line 305: `SCYLLA_ASSERT(_sstable_directories.empty());`
-
-Total: 1 usages
-
-### schema/schema_registry.cc
-
-- Line 45: `SCYLLA_ASSERT(!_schema);`
-
-Total: 1 usages
-
-### service/direct_failure_detector/failure_detector.cc
-
-- Line 378: `SCYLLA_ASSERT(_ping_fiber.available());`
-- Line 379: `SCYLLA_ASSERT(_notify_fiber.available());`
-- Line 701: `SCYLLA_ASSERT(_shard_workers.empty());`
-- Line 702: `SCYLLA_ASSERT(_destroy_subscriptions.available());`
-- Line 703: `SCYLLA_ASSERT(_update_endpoint_fiber.available());`
-- Line 707: `SCYLLA_ASSERT(!_impl);`
-
-Total: 6 usages
-
-### service/load_broadcaster.hh
-
-- Line 37: `SCYLLA_ASSERT(_stopped);`
-
-Total: 1 usages
-
-### service/paxos/paxos_state.cc
-
-- Line 323: `SCYLLA_ASSERT(_stopped);`
-
-Total: 1 usages
-
-### service/storage_proxy.cc
-
-- Line 281: `SCYLLA_ASSERT(_stopped);`
-- Line 3207: `SCYLLA_ASSERT(!_remote);`
-
-Total: 2 usages
-
-### service/tablet_allocator.cc
-
-- Line 3288: `SCYLLA_ASSERT(_stopped);`
-
-Total: 1 usages
-
-### sstables/compressor.cc
-
-- Line 1271: `SCYLLA_ASSERT(thread::running_in_thread());`
-
-Total: 1 usages
-
-### sstables/sstables_manager.cc
-
-- Line 58: `SCYLLA_ASSERT(_closing);`
-- Line 59: `SCYLLA_ASSERT(_active.empty());`
-- Line 60: `SCYLLA_ASSERT(_undergoing_close.empty());`
-
-Total: 3 usages
-
-### sstables/sstables_manager.hh
-
-- Line 188: `SCYLLA_ASSERT(_storage != nullptr);`
-
-Total: 1 usages
-
-### utils/cached_file.hh
-
-- Line 477: `SCYLLA_ASSERT(_cache.empty());`
-
-Total: 1 usages
-
-### utils/disk_space_monitor.cc
-
-- Line 66: `SCYLLA_ASSERT(_poller_fut.available());`
-
-Total: 1 usages
-
-### utils/file_lock.cc
-
-- Line 34: `SCYLLA_ASSERT(_fd.get() != -1);`
-- Line 36: `SCYLLA_ASSERT(r == 0);`
-
-Total: 2 usages
-
-### utils/logalloc.cc
-
-- Line 1991: `SCYLLA_ASSERT(desc.is_empty());`
-- Line 1996: `SCYLLA_ASSERT(segment_pool().descriptor(_active).is_empty());`
-
-Total: 2 usages
-
-### utils/lru.hh
-
-- Line 41: `SCYLLA_ASSERT(!_lru_link.is_linked());`
-
-Total: 1 usages
-
-### utils/replicator.hh
-
-- Line 221: `SCYLLA_ASSERT(_stopped);`
-
-Total: 1 usages
-

docs/operating-scylla/nodetool-commands/setlogginglevel.rst

@@ -110,7 +110,6 @@ To display the log classes (output changes with each version so your display may
    keys
    keyspace_utils
    large_data
-   legacy_schema_migrator
    lister
    load_balancer
    load_broadcaster

docs/operating-scylla/nodetool-commands/tasks/list.rst

@@ -42,21 +42,21 @@ For single list:
 
 .. code-block:: shell
 
-    task_id                              type   kind scope    state sequence_number keyspace table entity shard start_time           end_time
-    5116ddb6-85b5-4c3e-94fb-72128f15d7b4 repair node keyspace done  3               abc                   0     2025-01-16T16:12:11Z 2025-01-16T16:12:13Z
+    task_id                              type   kind scope    state sequence_number keyspace table entity shard creation_time        start_time           end_time
+    5116ddb6-85b5-4c3e-94fb-72128f15d7b4 repair node keyspace done  3               abc                   0     2025-01-16T16:12:08Z 2025-01-16T16:12:11Z 2025-01-16T16:12:13Z
 
 With repetition:
 
 .. code-block:: shell
 
-    task_id                              type   kind scope    state   sequence_number keyspace table entity shard start_time           end_time
-    d8926ee7-0faf-47b7-bfeb-82477e0c7b33 repair node keyspace running 5               abc                   0     2025-01-16T16:12:57Z
-    1e028cb8-31a3-45ed-8728-af7a1ab586f6 repair node keyspace done    4               abc                   0     2025-01-16T16:12:45Z 2025-01-16T16:12:47Z
+    task_id                              type   kind scope    state   sequence_number keyspace table entity shard creation_time        start_time           end_time
+    d8926ee7-0faf-47b7-bfeb-82477e0c7b33 repair node keyspace running 5               abc                   0     2025-01-16T16:12:54Z 2025-01-16T16:12:57Z
+    1e028cb8-31a3-45ed-8728-af7a1ab586f6 repair node keyspace done    4               abc                   0     2025-01-16T16:12:42Z 2025-01-16T16:12:45Z 2025-01-16T16:12:47Z
 
-    task_id                              type   kind scope    state   sequence_number keyspace table entity shard start_time           end_time
-    1e535f9b-97fa-4788-a956-8f3216a6ea8d repair node keyspace created 6               abc                   0
-    d8926ee7-0faf-47b7-bfeb-82477e0c7b33 repair node keyspace running 5               abc                   0     2025-01-16T16:12:57Z
-    1e028cb8-31a3-45ed-8728-af7a1ab586f6 repair node keyspace done    4               abc                   0     2025-01-16T16:12:45Z 2025-01-16T16:12:47Z
+    task_id                              type   kind scope    state   sequence_number keyspace table entity shard creation_time        start_time           end_time
+    1e535f9b-97fa-4788-a956-8f3216a6ea8d repair node keyspace created 6               abc                   0     2025-01-16T16:13:02Z
+    d8926ee7-0faf-47b7-bfeb-82477e0c7b33 repair node keyspace running 5               abc                   0     2025-01-16T16:12:54Z 2025-01-16T16:12:57Z
+    1e028cb8-31a3-45ed-8728-af7a1ab586f6 repair node keyspace done    4               abc                   0     2025-01-16T16:12:42Z 2025-01-16T16:12:45Z 2025-01-16T16:12:47Z
 
 See also
 --------

docs/operating-scylla/nodetool-commands/tasks/status.rst

@@ -25,6 +25,7 @@ Example output
    scope: keyspace
    state: running
    is_abortable: true
+   creation_time: 2024-07-29T15:48:50Z
    start_time: 2024-07-29T15:48:55Z
    end_time:
    error:

docs/operating-scylla/nodetool-commands/tasks/tree.rst

@@ -26,22 +26,22 @@ For single task:
 
 .. code-block:: shell
 
-   id                                   type   kind scope    state is_abortable start_time           end_time             error parent_id                            sequence_number shard keyspace table entity progress_units total completed children_ids
-   be5559ea-bc5a-428c-b8ce-d14eac7a1765 repair node keyspace done  true         2024-07-29T16:06:46Z 2024-07-29T16:06:46Z       none                                 1               0     abc                   ranges         4     4         [{task_id: 542e38cb-9ad4-40aa-9010-de2630004e55, node: 127.0.0.1 }, {task_id: 8974ebcc-1e87-4040-88fe-f2438261f7fb, node: 127.0.0.1 }]
-   542e38cb-9ad4-40aa-9010-de2630004e55 repair node shard    done  false        2024-07-29T16:06:46Z 2024-07-29T16:06:46Z       be5559ea-bc5a-428c-b8ce-d14eac7a1765 1               0     abc                   ranges         2     2         []
-   8974ebcc-1e87-4040-88fe-f2438261f7fb repair node shard    done  false        2024-07-29T16:06:46Z 2024-07-29T16:06:46Z       be5559ea-bc5a-428c-b8ce-d14eac7a1765 1               1     abc                   ranges         2     2         []
+   id                                   type   kind scope    state is_abortable creation_time        start_time           end_time             error parent_id                            sequence_number shard keyspace table entity progress_units total completed children_ids
+   be5559ea-bc5a-428c-b8ce-d14eac7a1765 repair node keyspace done  true         2024-07-29T16:06:43Z 2024-07-29T16:06:46Z 2024-07-29T16:06:46Z       none                                 1               0     abc                   ranges         4     4         [{task_id: 542e38cb-9ad4-40aa-9010-de2630004e55, node: 127.0.0.1 }, {task_id: 8974ebcc-1e87-4040-88fe-f2438261f7fb, node: 127.0.0.1 }]
+   542e38cb-9ad4-40aa-9010-de2630004e55 repair node shard    done  false        2024-07-29T16:06:43Z 2024-07-29T16:06:46Z 2024-07-29T16:06:46Z       be5559ea-bc5a-428c-b8ce-d14eac7a1765 1               0     abc                   ranges         2     2         []
+   8974ebcc-1e87-4040-88fe-f2438261f7fb repair node shard    done  false        2024-07-29T16:06:43Z 2024-07-29T16:06:46Z 2024-07-29T16:06:46Z       be5559ea-bc5a-428c-b8ce-d14eac7a1765 1               1     abc                   ranges         2     2         []
 
 For all tasks:
 
 .. code-block:: shell
 
-   id                                   type                   kind    scope    state is_abortable start_time           end_time             error parent_id                            sequence_number shard keyspace table entity progress_units total completed children_ids
-   16eafb1e-8b2e-48e6-bd7a-432ca3d8b9fc repair                 node    keyspace done  true         2024-07-29T16:34:46Z 2024-07-29T16:34:46Z       none                                 1               0     abc                   ranges         4     4         [{task_id: e0aa1aa4-58ca-4bfb-b3e6-74e5f3a0f6ee, node: 127.0.0.1 }, {task_id: 49eb5797-b67e-46b0-9365-4460f7cf988a, node: 127.0.0.1 }]
-   e0aa1aa4-58ca-4bfb-b3e6-74e5f3a0f6ee repair                 node    shard    done  false        2024-07-29T16:34:46Z 2024-07-29T16:34:46Z       16eafb1e-8b2e-48e6-bd7a-432ca3d8b9fc 1               0     abc                   ranges         2     2         []
-   49eb5797-b67e-46b0-9365-4460f7cf988a repair                 node    shard    done  false        2024-07-29T16:34:46Z 2024-07-29T16:34:46Z       16eafb1e-8b2e-48e6-bd7a-432ca3d8b9fc 1               1     abc                   ranges         2     2         []
-   82d7b2a4-146e-4a72-ba93-c66d5b4e9867 offstrategy compaction node    keyspace done  true         2024-07-29T16:34:16Z 2024-07-29T16:34:16Z       none                                 954             0     abc                                  1     1         [{task_id: 9818277b-238d-4298-a56b-c0d2153bf140, node: 127.0.0.1 }, {task_id: c1eb0701-ad7a-45ff-956f-7b8d671fc5db, node: 127.0.0.1 }
-   9818277b-238d-4298-a56b-c0d2153bf140 offstrategy compaction node    shard    done  false        2024-07-29T16:34:16Z 2024-07-29T16:34:16Z       82d7b2a4-146e-4a72-ba93-c66d5b4e9867 954             0     abc                                  1     1         []
-   c1eb0701-ad7a-45ff-956f-7b8d671fc5db offstrategy compaction node    shard    done  false        2024-07-29T16:34:16Z 2024-07-29T16:34:16Z       82d7b2a4-146e-4a72-ba93-c66d5b4e9867 954             1     abc                                  1     1         []
+   id                                   type                   kind    scope    state is_abortable creation_time        start_time           end_time             error parent_id                            sequence_number shard keyspace table entity progress_units total completed children_ids
+   16eafb1e-8b2e-48e6-bd7a-432ca3d8b9fc repair                 node    keyspace done  true         2024-07-29T16:34:43Z 2024-07-29T16:34:46Z 2024-07-29T16:34:46Z       none                                 1               0     abc                   ranges         4     4         [{task_id: e0aa1aa4-58ca-4bfb-b3e6-74e5f3a0f6ee, node: 127.0.0.1 }, {task_id: 49eb5797-b67e-46b0-9365-4460f7cf988a, node: 127.0.0.1 }]
+   e0aa1aa4-58ca-4bfb-b3e6-74e5f3a0f6ee repair                 node    shard    done  false        2024-07-29T16:34:43Z 2024-07-29T16:34:46Z 2024-07-29T16:34:46Z       16eafb1e-8b2e-48e6-bd7a-432ca3d8b9fc 1               0     abc                   ranges         2     2         []
+   49eb5797-b67e-46b0-9365-4460f7cf988a repair                 node    shard    done  false        2024-07-29T16:34:43Z 2024-07-29T16:34:46Z 2024-07-29T16:34:46Z       16eafb1e-8b2e-48e6-bd7a-432ca3d8b9fc 1               1     abc                   ranges         2     2         []
+   82d7b2a4-146e-4a72-ba93-c66d5b4e9867 offstrategy compaction node    keyspace done  true         2024-07-29T16:34:13Z 2024-07-29T16:34:16Z 2024-07-29T16:34:16Z       none                                 954             0     abc                                  1     1         [{task_id: 9818277b-238d-4298-a56b-c0d2153bf140, node: 127.0.0.1 }, {task_id: c1eb0701-ad7a-45ff-956f-7b8d671fc5db, node: 127.0.0.1 }
+   9818277b-238d-4298-a56b-c0d2153bf140 offstrategy compaction node    shard    done  false        2024-07-29T16:34:13Z 2024-07-29T16:34:16Z 2024-07-29T16:34:16Z       82d7b2a4-146e-4a72-ba93-c66d5b4e9867 954             0     abc                                  1     1         []
+   c1eb0701-ad7a-45ff-956f-7b8d671fc5db offstrategy compaction node    shard    done  false        2024-07-29T16:34:13Z 2024-07-29T16:34:16Z 2024-07-29T16:34:16Z       82d7b2a4-146e-4a72-ba93-c66d5b4e9867 954             1     abc                                  1     1         []
 
 See also
 --------

idl/raft.idl.hh

@@ -129,6 +129,6 @@ struct direct_fd_ping_reply {
     std::variant<std::monostate, service::wrong_destination, service::group_liveness_info> result;
 };
 
-verb [[with_client_info, cancellable]] direct_fd_ping (raft::server_id dst_id) -> service::direct_fd_ping_reply;
+verb [[with_client_info, with_timeout, cancellable]] direct_fd_ping (raft::server_id dst_id) -> service::direct_fd_ping_reply;
 
 } // namespace service

main.cc

@@ -39,7 +39,6 @@
 #include "api/api_init.hh"
 #include "db/config.hh"
 #include "db/extensions.hh"
-#include "db/legacy_schema_migrator.hh"
 #include "service/storage_service.hh"
 #include "service/migration_manager.hh"
 #include "service/tablet_allocator.hh"
@@ -1641,7 +1640,7 @@ To start the scylla server proper, simply invoke as: scylla server (or just scyl
             fd.start(
                 std::ref(fd_pinger), std::ref(fd_clock),
                 service::direct_fd_clock::base::duration{std::chrono::milliseconds{100}}.count(),
-                service::direct_fd_clock::base::duration{std::chrono::milliseconds{cfg->direct_failure_detector_ping_timeout_in_ms()}}.count()).get();
+                service::direct_fd_clock::base::duration{std::chrono::milliseconds{cfg->direct_failure_detector_ping_timeout_in_ms()}}.count(), dbcfg.gossip_scheduling_group).get();
 
             auto stop_fd = defer_verbose_shutdown("direct_failure_detector", [] {
                 fd.stop().get();
@@ -1851,8 +1850,6 @@ To start the scylla server proper, simply invoke as: scylla server (or just scyl
             group0_client.init().get();
 
             checkpoint(stop_signal, "initializing system schema");
-            // schema migration, if needed, is also done on shard 0
-            db::legacy_schema_migrator::migrate(proxy, db, sys_ks, qp.local()).get();
             db::schema_tables::save_system_schema(qp.local()).get();
             db::schema_tables::recalculate_schema_version(sys_ks, proxy, feature_service.local()).get();
 

message/messaging_service.cc

@@ -686,6 +686,7 @@ static constexpr unsigned do_get_rpc_client_idx(messaging_verb verb) {
     case messaging_verb::RAFT_MODIFY_CONFIG:
     case messaging_verb::RAFT_PULL_SNAPSHOT:
     case messaging_verb::NOTIFY_BANNED:
+    case messaging_verb::DIRECT_FD_PING:
         // See comment above `TOPOLOGY_INDEPENDENT_IDX`.
         // DO NOT put any 'hot' (e.g. data path) verbs in this group,
         // only verbs which are 'rare' and 'cheap'.
@@ -747,7 +748,6 @@ static constexpr unsigned do_get_rpc_client_idx(messaging_verb verb) {
     case messaging_verb::PAXOS_ACCEPT:
     case messaging_verb::PAXOS_LEARN:
     case messaging_verb::PAXOS_PRUNE:
-    case messaging_verb::DIRECT_FD_PING:
         return 2;
     case messaging_verb::MUTATION_DONE:
     case messaging_verb::MUTATION_FAILED:

mutation/partition_version.cc

@@ -575,10 +575,15 @@ utils::coroutine partition_entry::apply_to_incomplete(const schema& s,
                         }
                         res.row.set_range_tombstone(cur.range_tombstone_for_row() + src_cur.range_tombstone());
 
+                        if (need_preempt()) {
+                            lb = position_in_partition(cur.position());
+                            ++tracker.get_stats().rows_covered_by_range_tombstones_from_memtable;
+                            return stop_iteration::no;
+                        }
+
                         // FIXME: Compact the row
                         ++tracker.get_stats().rows_covered_by_range_tombstones_from_memtable;
                         cur.next();
-                        // FIXME: preempt
                     }
                 }
                 {

node_ops/task_manager_module.cc

@@ -14,6 +14,7 @@
 #include "tasks/task_handler.hh"
 #include "tasks/virtual_task_hint.hh"
 #include "utils/error_injection.hh"
+#include "utils/UUID_gen.hh"
 #include <variant>
 #include "utils/overloaded_functor.hh"
 
@@ -90,6 +91,7 @@ future<std::optional<tasks::task_status>> node_ops_virtual_task::get_status_help
         .scope = "cluster",
         .state = get_state(entry),
         .is_abortable = co_await is_abortable(std::move(hint)),
+        .creation_time = db_clock::time_point(utils::UUID_gen::unix_timestamp(id.uuid())),
         .start_time = entry.start_time,
         .end_time = entry.end_time,
         .error = entry.error,
@@ -167,6 +169,7 @@ future<std::vector<tasks::task_stats>> node_ops_virtual_task::get_stats() {
             .table = "",
             .entity = "",
             .shard = 0,
+            .creation_time = db_clock::time_point(utils::UUID_gen::unix_timestamp(id)),
             .start_time = entry.start_time,
             .end_time = entry.end_time
         };

raft/tracker.cc

@@ -46,7 +46,7 @@ bool follower_progress::is_stray_reject(const append_reply::rejected& rejected)
         // any reject during snapshot transfer is stray one
         return true;
     default:
-        scylla_assert(false, "invalid follower_progress state: {}", static_cast<int>(state));
+        SCYLLA_ASSERT(false);
     }
     return false;
 }
@@ -87,7 +87,7 @@ bool follower_progress::can_send_to() {
         // before starting to sync the log.
         return false;
     }
-    scylla_assert(false, "invalid follower_progress state in can_send_to: {}", static_cast<int>(state));
+    SCYLLA_ASSERT(false);
     return false;
 }
 

service/direct_failure_detector/failure_detector.cc

@@ -6,6 +6,7 @@
  * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
  */
 
+#include "seastar/core/scheduling.hh"
 #include "utils/assert.hh"
 #include <unordered_set>
 
@@ -17,6 +18,7 @@
 #include <seastar/core/condition-variable.hh>
 #include <seastar/coroutine/parallel_for_each.hh>
 #include <seastar/util/defer.hh>
+#include <seastar/coroutine/switch_to.hh>
 
 #include "utils/log.hh"
 
@@ -118,7 +120,7 @@ struct failure_detector::impl {
 
     // Fetches endpoint updates from _endpoint_queue and performs the add/remove operation.
     // Runs on shard 0 only.
-    future<> update_endpoint_fiber();
+    future<> update_endpoint_fiber(seastar::scheduling_group sg);
     future<> _update_endpoint_fiber = make_ready_future<>();
 
     // Workers running on this shard.
@@ -140,7 +142,7 @@ struct failure_detector::impl {
     // The unregistering process requires cross-shard operations which we perform on this fiber.
     future<> _destroy_subscriptions = make_ready_future<>();
 
-    impl(failure_detector& parent, pinger&, clock&, clock::interval_t ping_period, clock::interval_t ping_timeout);
+    impl(failure_detector& parent, pinger&, clock&, clock::interval_t ping_period, clock::interval_t ping_timeout, seastar::scheduling_group sg);
     ~impl();
 
     // Inform update_endpoint_fiber() about an added/removed endpoint.
@@ -177,19 +179,19 @@ struct failure_detector::impl {
 };
 
 failure_detector::failure_detector(
-    pinger& pinger, clock& clock, clock::interval_t ping_period, clock::interval_t ping_timeout)
-        : _impl(std::make_unique<impl>(*this, pinger, clock, ping_period, ping_timeout))
+    pinger& pinger, clock& clock, clock::interval_t ping_period, clock::interval_t ping_timeout, seastar::scheduling_group sg)
+        : _impl(std::make_unique<impl>(*this, pinger, clock, ping_period, ping_timeout, sg))
 {}
 
 failure_detector::impl::impl(
-    failure_detector& parent, pinger& pinger, clock& clock, clock::interval_t ping_period, clock::interval_t ping_timeout)
+    failure_detector& parent, pinger& pinger, clock& clock, clock::interval_t ping_period, clock::interval_t ping_timeout, seastar::scheduling_group sg)
         : _parent(parent), _pinger(pinger), _clock(clock), _ping_period(ping_period), _ping_timeout(ping_timeout) {
     if (this_shard_id() != 0) {
         return;
     }
 
     _num_workers.resize(smp::count, 0);
-    _update_endpoint_fiber = update_endpoint_fiber();
+    _update_endpoint_fiber = update_endpoint_fiber(sg);
 }
 
 void failure_detector::impl::send_update_endpoint(pinger::endpoint_id ep, endpoint_update update) {
@@ -205,9 +207,9 @@ void failure_detector::impl::send_update_endpoint(pinger::endpoint_id ep, endpoi
     _endpoint_changed.signal();
 }
 
-future<> failure_detector::impl::update_endpoint_fiber() {
+future<> failure_detector::impl::update_endpoint_fiber(seastar::scheduling_group sg) {
     SCYLLA_ASSERT(this_shard_id() == 0);
-
+    co_await coroutine::switch_to(sg);
     while (true) {
         co_await _endpoint_changed.wait([this] { return !_endpoint_updates.empty(); });
 
@@ -480,7 +482,7 @@ static future<bool> ping_with_timeout(pinger::endpoint_id id, clock::timepoint_t
         }
     });
 
-    auto f = pinger.ping(id, timeout_as);
+    auto f = pinger.ping(id, timeout, timeout_as, c);
     auto sleep_and_abort = [] (clock::timepoint_t timeout, abort_source& timeout_as, clock& c) -> future<> {
         co_await c.sleep_until(timeout, timeout_as).then_wrapped([&timeout_as] (auto&& f) {
             // Avoid throwing if sleep was aborted.

service/direct_failure_detector/failure_detector.hh

@@ -19,26 +19,6 @@ class abort_source;
 
 namespace direct_failure_detector {
 
-class pinger {
-public:
-    // Opaque endpoint ID.
-    // A specific implementation of `pinger` maps those IDs to 'real' addresses.
-    using endpoint_id = utils::UUID;
-
-    // Send a message to `ep` and wait until it responds.
-    // The wait can be aborted using `as`.
-    // Abort should be signalized with `abort_requested_exception`.
-    //
-    // If the ping fails in an expected way (e.g. the endpoint is down and refuses to connect),
-    // returns `false`. If it succeeds, returns `true`.
-    virtual future<bool> ping(endpoint_id ep, abort_source& as) = 0;
-
-protected:
-    // The `pinger` object must not be destroyed through the `pinger` interface.
-    // `failure_detector` does not take ownership of `pinger`, only a non-owning reference.
-    ~pinger() = default;
-};
-
 // A clock that uses abstract units to measure time.
 // The implementation is responsible for periodically advancing the clock.
 //
@@ -60,12 +40,33 @@ public:
     // Aborts should be signalized using `seastar::sleep_aborted`.
     virtual future<> sleep_until(timepoint_t tp, abort_source& as) = 0;
 
+    virtual std::chrono::milliseconds to_milliseconds(timepoint_t tp) const = 0;
 protected:
     // The `clock` object must not be destroyed through the `clock` interface.
     // `failure_detector` does not take ownership of `clock`, only a non-owning reference.
     ~clock() = default;
 };
 
+class pinger {
+public:
+    // Opaque endpoint ID.
+    // A specific implementation of `pinger` maps those IDs to 'real' addresses.
+    using endpoint_id = utils::UUID;
+
+    // Send a message to `ep` and wait until it responds.
+    // The wait can be aborted using `as`.
+    // Abort should be signalized with `abort_requested_exception`.
+    //
+    // If the ping fails in an expected way (e.g. the endpoint is down and refuses to connect),
+    // returns `false`. If it succeeds, returns `true`.
+    virtual future<bool> ping(endpoint_id ep, clock::timepoint_t timeout, abort_source& as, clock& c) = 0;
+
+protected:
+    // The `pinger` object must not be destroyed through the `pinger` interface.
+    // `failure_detector` does not take ownership of `pinger`, only a non-owning reference.
+    ~pinger() = default;
+};
+
 class listener {
 public:
     // Called when an endpoint in the detected set (added by `failure_detector::add_endpoint`) responds to a ping
@@ -127,7 +128,10 @@ public:
 
         // Duration after which a ping is aborted, so that next ping can be started
         // (pings are sent sequentially).
-        clock::interval_t ping_timeout
+        clock::interval_t ping_timeout,
+
+        // Scheduling group used for fibers inside the failure detector.
+        seastar::scheduling_group sg
     );
 
     ~failure_detector();

service/raft/raft_group_registry.cc

@@ -18,6 +18,7 @@
 #include "utils/error_injection.hh"
 #include "seastar/core/shared_future.hh"
 
+#include <chrono>
 #include <seastar/core/coroutine.hh>
 #include <seastar/core/when_all.hh>
 #include <seastar/core/sleep.hh>
@@ -202,8 +203,11 @@ void raft_group_registry::init_rpc_verbs() {
     });
 
     ser::raft_rpc_verbs::register_direct_fd_ping(&_ms,
-            [this] (const rpc::client_info&, raft::server_id dst) -> future<direct_fd_ping_reply> {
-        // XXX: update address map here as well?
+            [this] (const rpc::client_info&, rpc::opt_time_point timeout, raft::server_id dst) -> future<direct_fd_ping_reply> {
+
+        if (timeout && *timeout <= netw::messaging_service::clock_type::now()) {
+            throw timed_out_error{};
+        }
 
         if (_my_id != dst) {
             return make_ready_future<direct_fd_ping_reply>(direct_fd_ping_reply {
@@ -213,19 +217,10 @@ void raft_group_registry::init_rpc_verbs() {
             });
         }
 
-        return container().invoke_on(0, [] (raft_group_registry& me) -> future<direct_fd_ping_reply> {
-            bool group0_alive = false;
-            if (me._group0_id) {
-                auto* group0_server = me.find_server(*me._group0_id);
-                if (group0_server && group0_server->is_alive()) {
-                    group0_alive = true;
-                }
+        return make_ready_future<direct_fd_ping_reply>(direct_fd_ping_reply {
+            .result = service::group_liveness_info{
+                .group0_alive = _group0_is_alive,
             }
-            co_return direct_fd_ping_reply {
-                .result = service::group_liveness_info{
-                    .group0_alive = group0_alive,
-                }
-            };
         });
     });
 }
@@ -380,6 +375,12 @@ future<> raft_group_registry::start_server_for_group(raft_server_for_group new_g
         co_await server.abort();
         std::rethrow_exception(ex);
     }
+
+    if (gid == _group0_id) {
+        co_await container().invoke_on_all([] (raft_group_registry& rg) {
+            rg._group0_is_alive = true;
+        });
+    }
 }
 
 future<> raft_group_registry::abort_server(raft::group_id gid, sstring reason) {
@@ -389,14 +390,18 @@ future<> raft_group_registry::abort_server(raft::group_id gid, sstring reason) {
     if (const auto it = _servers.find(gid); it != _servers.end()) {
         auto& [gid, s] = *it;
         if (!s.aborted) {
+            if (gid == _group0_id) {
+                co_await container().invoke_on_all([] (raft_group_registry& rg) {
+                    rg._group0_is_alive = false;
+                });
+            }
             s.aborted = s.server->abort(std::move(reason))
                 .handle_exception([gid] (std::exception_ptr ex) {
                     rslog.warn("Failed to abort raft group server {}: {}", gid, ex);
                 });
         }
-        return s.aborted->get_future();
+        co_await s.aborted->get_future();
     }
-    return make_ready_future<>();
 }
 
 unsigned raft_group_registry::shard_for_group(const raft::group_id& gid) const {
@@ -517,11 +522,13 @@ future<> raft_server_with_timeouts::read_barrier(seastar::abort_source* as, std:
     }, "read_barrier", as, timeout);
 }
 
-future<bool> direct_fd_pinger::ping(direct_failure_detector::pinger::endpoint_id id, abort_source& as) {
+future<bool> direct_fd_pinger::ping(direct_failure_detector::pinger::endpoint_id id, direct_failure_detector::clock::timepoint_t timeout, abort_source& as, direct_failure_detector::clock& c) {
     auto dst_id = raft::server_id{id};
 
     try {
-        auto reply = co_await ser::raft_rpc_verbs::send_direct_fd_ping(&_ms, locator::host_id{id}, as, dst_id);
+        std::chrono::milliseconds timeout_ms = c.to_milliseconds(timeout);
+        netw::messaging_service::clock_type::time_point deadline = netw::messaging_service::clock_type::now() + timeout_ms;
+        auto reply = co_await ser::raft_rpc_verbs::send_direct_fd_ping(&_ms, locator::host_id{id}, deadline, as, dst_id);
         if (auto* wrong_dst = std::get_if<wrong_destination>(&reply.result)) {
             // FIXME: after moving to host_id based verbs we will not get `wrong_destination`
             //        any more since the connection will fail
@@ -554,4 +561,11 @@ future<> direct_fd_clock::sleep_until(direct_failure_detector::clock::timepoint_
     return sleep_abortable(t - n, as);
 }
 
+std::chrono::milliseconds direct_fd_clock::to_milliseconds(direct_failure_detector::clock::timepoint_t tp) const {
+    auto t = base::time_point{base::duration{tp}};
+    auto n = base::now();
+    return std::chrono::duration_cast<std::chrono::milliseconds>(t - n);
+}
+
+
 } // end of namespace service

service/raft/raft_group_registry.hh

@@ -127,6 +127,7 @@ private:
     // My Raft ID. Shared between different Raft groups.
     raft::server_id _my_id;
 
+    bool _group0_is_alive = false;
 public:
     raft_group_registry(raft::server_id my_id, netw::messaging_service& ms,
             direct_failure_detector::failure_detector& fd);
@@ -181,6 +182,9 @@ public:
     unsigned shard_for_group(const raft::group_id& gid) const;
     shared_ptr<raft::failure_detector> failure_detector();
     direct_failure_detector::failure_detector& direct_fd() { return _direct_fd; }
+    bool is_group0_alive() const {
+        return _group0_is_alive;
+    }
 };
 
 // Implementation of `direct_failure_detector::pinger` which uses DIRECT_FD_PING verb for pinging.
@@ -198,7 +202,7 @@ public:
     direct_fd_pinger(const direct_fd_pinger&) = delete;
     direct_fd_pinger(direct_fd_pinger&&) = delete;
 
-    future<bool> ping(direct_failure_detector::pinger::endpoint_id id, abort_source& as) override;
+    future<bool> ping(direct_failure_detector::pinger::endpoint_id id, direct_failure_detector::clock::timepoint_t timeout, abort_source& as, direct_failure_detector::clock& c) override;
 };
 
 // XXX: find a better place to put this?
@@ -207,6 +211,7 @@ struct direct_fd_clock : public direct_failure_detector::clock {
 
     direct_failure_detector::clock::timepoint_t now() noexcept override;
     future<> sleep_until(direct_failure_detector::clock::timepoint_t tp, abort_source& as) override;
+    std::chrono::milliseconds to_milliseconds(direct_failure_detector::clock::timepoint_t tp) const override;
 };
 
 } // end of namespace service

service/storage_proxy.cc

@@ -1138,8 +1138,7 @@ private:
             topology_mutation_builder builder(guard.write_timestamp());
             topology_request_tracking_mutation_builder trbuilder(global_request_id, _sp._features.topology_requests_type_column);
             trbuilder.set_truncate_table_data(table_id)
-                     .set("done", false)
-                     .set("start_time", db_clock::now());
+                     .set("done", false);
 
             if (!_sp._features.topology_global_request_queue) {
                 builder.set_global_topology_request(global_topology_request::truncate_table)
@@ -6688,10 +6687,11 @@ storage_proxy::do_query_with_paxos(schema_ptr s,
         }
     };
 
-    auto request = seastar::make_shared<read_cas_request>();
+    auto request = std::make_unique<read_cas_request>();
+    auto* request_ptr = request.get();
 
-    return cas(std::move(s), std::move(cas_shard), request, cmd, std::move(partition_ranges), std::move(query_options),
-            cl, db::consistency_level::ANY, timeout, cas_timeout, false).then([request] (bool is_applied) mutable {
+    return cas(std::move(s), std::move(cas_shard), *request_ptr, cmd, std::move(partition_ranges), std::move(query_options),
+            cl, db::consistency_level::ANY, timeout, cas_timeout, false).then([request = std::move(request)] (bool is_applied) mutable {
         return make_ready_future<coordinator_query_result>(std::move(request->res));
     });
 }
@@ -6754,11 +6754,13 @@ static mutation_write_failure_exception read_failure_to_write(read_failure_excep
  * NOTE: `cmd` argument can be nullptr, in which case it's guaranteed that this function would not perform
  * any reads of committed values (in case user of the function is not interested in them).
  *
+ * NOTE: The `request` object must be guaranteed to be alive until the returned future is resolved.
+ *
  * WARNING: the function must be called on a shard that owns the key cas() operates on.
  * The cas_shard must be created *before* selecting the shard, to protect against
  * concurrent tablet migrations.
  */
-future<bool> storage_proxy::cas(schema_ptr schema, cas_shard cas_shard, shared_ptr<cas_request> request, lw_shared_ptr<query::read_command> cmd,
+future<bool> storage_proxy::cas(schema_ptr schema, cas_shard cas_shard, cas_request& request, lw_shared_ptr<query::read_command> cmd,
         dht::partition_range_vector partition_ranges, storage_proxy::coordinator_query_options query_options,
         db::consistency_level cl_for_paxos, db::consistency_level cl_for_learn,
         clock_type::time_point write_timeout, clock_type::time_point cas_timeout, bool write, cdc::per_request_options cdc_opts) {
@@ -6859,7 +6861,7 @@ future<bool> storage_proxy::cas(schema_ptr schema, cas_shard cas_shard, shared_p
                 qr = std::move(cqr.query_result);
             }
 
-            auto mutation = request->apply(std::move(qr), cmd->slice, utils::UUID_gen::micros_timestamp(ballot), cdc_opts);
+            auto mutation = request.apply(std::move(qr), cmd->slice, utils::UUID_gen::micros_timestamp(ballot), cdc_opts);
             condition_met = true;
             if (!mutation) {
                 if (write) {

service/storage_proxy.hh

@@ -829,7 +829,7 @@ public:
         clock_type::time_point timeout,
         tracing::trace_state_ptr trace_state = nullptr);
 
-    future<bool> cas(schema_ptr schema, cas_shard cas_shard, shared_ptr<cas_request> request, lw_shared_ptr<query::read_command> cmd,
+    future<bool> cas(schema_ptr schema, cas_shard cas_shard, cas_request& request, lw_shared_ptr<query::read_command> cmd,
             dht::partition_range_vector partition_ranges, coordinator_query_options query_options,
             db::consistency_level cl_for_paxos, db::consistency_level cl_for_learn,
             clock_type::time_point write_timeout, clock_type::time_point cas_timeout, bool write = true, cdc::per_request_options cdc_opts = {});

service/storage_service.cc

@@ -600,7 +600,7 @@ future<storage_service::nodes_to_notify_after_sync> storage_service::sync_raft_t
             co_await update_topology_change_info(tmptr, ::format("{} {}/{}", rs.state, id, ip));
             break;
         case node_state::replacing: {
-            scylla_assert(_topology_state_machine._topology.req_param.contains(id));
+            SCYLLA_ASSERT(_topology_state_machine._topology.req_param.contains(id));
             auto replaced_id = std::get<replace_param>(_topology_state_machine._topology.req_param[id]).replaced_id;
             auto existing_ip = _address_map.find(locator::host_id{replaced_id.uuid()});
             const auto replaced_host_id = locator::host_id(replaced_id.uuid());
@@ -688,7 +688,7 @@ future<> storage_service::notify_nodes_after_sync(nodes_to_notify_after_sync&& n
 future<> storage_service::topology_state_load(state_change_hint hint) {
 #ifdef SEASTAR_DEBUG
     static bool running = false;
-    scylla_assert(!running); // The function is not re-entrant
+    SCYLLA_ASSERT(!running); // The function is not re-entrant
     auto d = defer([] {
         running = false;
     });
@@ -854,7 +854,7 @@ future<> storage_service::topology_state_load(state_change_hint hint) {
 }
 
 future<> storage_service::topology_transition(state_change_hint hint) {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     co_await topology_state_load(std::move(hint)); // reload new state
 
     _topology_state_machine.event.broadcast();
@@ -898,7 +898,7 @@ future<> storage_service::view_building_state_load() {
 }
 
 future<> storage_service::view_building_transition() {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     co_await view_building_state_load();
 
     _view_building_state_machine.event.broadcast();
@@ -966,7 +966,7 @@ future<> storage_service::merge_topology_snapshot(raft_snapshot snp) {
 }
 
 future<> storage_service::update_service_levels_cache(qos::update_both_cache_levels update_only_effective_cache, qos::query_context ctx) {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     if (_sl_controller.local().is_v2()) {
         // Skip cache update unless the topology upgrade is done
         co_await _sl_controller.local().update_cache(update_only_effective_cache, ctx);
@@ -1520,7 +1520,7 @@ future<> storage_service::update_topology_with_local_metadata(raft::server& raft
 }
 
 future<> storage_service::start_upgrade_to_raft_topology() {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
 
     if (_topology_state_machine._topology.upgrade_state != topology::upgrade_state_type::not_upgraded) {
         co_return;
@@ -1572,7 +1572,7 @@ future<> storage_service::start_upgrade_to_raft_topology() {
 }
 
 topology::upgrade_state_type storage_service::get_topology_upgrade_state() const {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     return _topology_state_machine._topology.upgrade_state;
 }
 
@@ -1841,7 +1841,7 @@ future<> storage_service::join_topology(sharded<service::storage_proxy>& proxy,
         slogger.info("Nodes {} are alive", get_sync_nodes());
     }
 
-    scylla_assert(_group0);
+    SCYLLA_ASSERT(_group0);
 
     join_node_request_params join_params {
         .host_id = _group0->load_my_id(),
@@ -2083,7 +2083,7 @@ future<> storage_service::join_topology(sharded<service::storage_proxy>& proxy,
 
     if (!_sys_ks.local().bootstrap_complete()) {
         // If we're not bootstrapping then we shouldn't have chosen a CDC streams timestamp yet.
-        scylla_assert(should_bootstrap() || !cdc_gen_id);
+        SCYLLA_ASSERT(should_bootstrap() || !cdc_gen_id);
 
         // Don't try rewriting CDC stream description tables.
         // See cdc.md design notes, `Streams description table V1 and rewriting` section, for explanation.
@@ -2167,7 +2167,7 @@ future<> storage_service::join_topology(sharded<service::storage_proxy>& proxy,
         throw std::runtime_error(err);
     }
 
-    scylla_assert(_group0);
+    SCYLLA_ASSERT(_group0);
     co_await _group0->finish_setup_after_join(*this, _qp, _migration_manager.local(), false);
     co_await _cdc_gens.local().after_join(std::move(cdc_gen_id));
 
@@ -2192,7 +2192,7 @@ future<> storage_service::join_topology(sharded<service::storage_proxy>& proxy,
 }
 
 future<> storage_service::track_upgrade_progress_to_topology_coordinator(sharded<service::storage_proxy>& proxy) {
-    scylla_assert(_group0);
+    SCYLLA_ASSERT(_group0);
 
     while (true) {
         _group0_as.check();
@@ -2316,7 +2316,7 @@ future<> storage_service::bootstrap(std::unordered_set<token>& bootstrap_tokens,
 
             // After we pick a generation timestamp, we start gossiping it, and we stick with it.
             // We don't do any other generation switches (unless we crash before complecting bootstrap).
-            scylla_assert(!cdc_gen_id);
+            SCYLLA_ASSERT(!cdc_gen_id);
 
             cdc_gen_id = _cdc_gens.local().legacy_make_new_generation(bootstrap_tokens, !is_first_node()).get();
 
@@ -2349,9 +2349,9 @@ future<> storage_service::bootstrap(std::unordered_set<token>& bootstrap_tokens,
             slogger.debug("Removing replaced endpoint {} from system.peers", replace_addr);
             _sys_ks.local().remove_endpoint(replace_addr).get();
 
-            scylla_assert(replaced_host_id);
+            SCYLLA_ASSERT(replaced_host_id);
             auto raft_id = raft::server_id{replaced_host_id.uuid()};
-            scylla_assert(_group0);
+            SCYLLA_ASSERT(_group0);
             bool raft_available = _group0->wait_for_raft().get();
             if (raft_available) {
                 slogger.info("Replace: removing {}/{} from group 0...", replace_addr, raft_id);
@@ -3000,7 +3000,7 @@ future<> storage_service::stop_transport() {
 }
 
 future<> storage_service::drain_on_shutdown() {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     return (_operation_mode == mode::DRAINING || _operation_mode == mode::DRAINED) ?
         _drain_finished.get_future() : do_drain();
 }
@@ -3025,7 +3025,7 @@ bool storage_service::is_topology_coordinator_enabled() const {
 
 future<> storage_service::join_cluster(sharded<service::storage_proxy>& proxy,
         start_hint_manager start_hm, gms::generation_type new_generation) {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
 
     if (_sys_ks.local().was_decommissioned()) {
         auto msg = sstring("This node was decommissioned and will not rejoin the ring unless "
@@ -3225,7 +3225,7 @@ future<> storage_service::join_cluster(sharded<service::storage_proxy>& proxy,
 }
 
 future<token_metadata_change> storage_service::prepare_token_metadata_change(mutable_token_metadata_ptr tmptr, const schema_getter& schema_getter) {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     std::exception_ptr ex;
     token_metadata_change change;
 
@@ -3992,7 +3992,7 @@ future<> storage_service::decommission() {
                 slogger.info("DECOMMISSIONING: starts");
                 ctl.req.leaving_nodes = std::list<gms::inet_address>{endpoint};
 
-                scylla_assert(ss._group0);
+                SCYLLA_ASSERT(ss._group0);
                 bool raft_available = ss._group0->wait_for_raft().get();
 
                 try {
@@ -4044,7 +4044,7 @@ future<> storage_service::decommission() {
 
                     if (raft_available && left_token_ring) {
                         slogger.info("decommission[{}]: leaving Raft group 0", uuid);
-                        scylla_assert(ss._group0);
+                        SCYLLA_ASSERT(ss._group0);
                         ss._group0->leave_group0().get();
                         slogger.info("decommission[{}]: left Raft group 0", uuid);
                     }
@@ -4350,7 +4350,7 @@ future<> storage_service::removenode(locator::host_id host_id, locator::host_id_
             auto stop_ctl = deferred_stop(ctl);
             auto uuid = ctl.uuid();
             const auto& tmptr = ctl.tmptr;
-            scylla_assert(ss._group0);
+            SCYLLA_ASSERT(ss._group0);
             auto raft_id = raft::server_id{host_id.uuid()};
             bool raft_available = ss._group0->wait_for_raft().get();
             bool is_group0_member = raft_available && ss._group0->is_member(raft_id, false);
@@ -4470,7 +4470,7 @@ future<> storage_service::removenode(locator::host_id host_id, locator::host_id_
 }
 
 future<> storage_service::check_and_repair_cdc_streams() {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
 
     if (!_cdc_gens.local_is_initialized()) {
         return make_exception_future<>(std::runtime_error("CDC generation service not initialized yet"));
@@ -4940,7 +4940,6 @@ future<> storage_service::do_clusterwide_vnodes_cleanup() {
             builder.queue_global_topology_request_id(request_id);
             topology_request_tracking_mutation_builder rtbuilder(request_id, _feature_service.topology_requests_type_column);
             rtbuilder.set("done", false)
-                     .set("start_time", db_clock::now())
                      .set("request_type", global_topology_request::cleanup);
             muts.push_back(rtbuilder.build());
         } else {
@@ -5265,7 +5264,6 @@ future<> storage_service::raft_check_and_repair_cdc_streams() {
                 topology_request_tracking_mutation_builder rtbuilder(request_id, _feature_service.topology_requests_type_column);
                 builder.queue_global_topology_request_id(request_id);
                 rtbuilder.set("done", false)
-                         .set("start_time", db_clock::now())
                          .set("request_type", global_topology_request::new_cdc_generation);
                 muts.push_back(rtbuilder.build());
             } else {
@@ -5784,7 +5782,7 @@ future<> storage_service::mutate_token_metadata(std::function<future<> (mutable_
 }
 
 future<> storage_service::update_topology_change_info(mutable_token_metadata_ptr tmptr, sstring reason) {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
 
     try {
         locator::dc_rack_fn get_dc_rack_by_host_id([this, &tm = *tmptr] (locator::host_id host_id) -> std::optional<locator::endpoint_dc_rack> {
@@ -5831,7 +5829,7 @@ future<> storage_service::keyspace_changed(const sstring& ks_name) {
 }
 
 future<locator::mutable_token_metadata_ptr> storage_service::prepare_tablet_metadata(const locator::tablet_metadata_change_hint& hint, mutable_token_metadata_ptr pending_token_metadata) {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     if (hint) {
         co_await replica::update_tablet_metadata(_db.local(), _qp, pending_token_metadata->tablets(), hint);
     } else {
@@ -5955,7 +5953,7 @@ void storage_service::start_tablet_split_monitor() {
 }
 
 future<> storage_service::snitch_reconfigured() {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
     auto& snitch = _snitch.local();
     co_await mutate_token_metadata([&snitch] (mutable_token_metadata_ptr tmptr) -> future<> {
         // re-read local rack and DC info
@@ -6487,8 +6485,8 @@ future<> storage_service::stream_tablet(locator::global_tablet_id tablet) {
     co_await utils::get_local_injector().inject("block_tablet_streaming", [this, &tablet] (auto& handler) -> future<> {
         const auto keyspace = handler.get("keyspace");
         const auto table = handler.get("table");
-        scylla_assert(keyspace);
-        scylla_assert(table);
+        SCYLLA_ASSERT(keyspace);
+        SCYLLA_ASSERT(table);
         auto s = _db.local().find_column_family(tablet.table).schema();
         bool should_block = s->ks_name() == *keyspace && s->cf_name() == *table;
         while (should_block && !handler.poll_for_message() && !_async_gate.is_closed()) {
@@ -7509,7 +7507,7 @@ future<join_node_request_result> storage_service::join_node_request_handler(join
 }
 
 future<join_node_response_result> storage_service::join_node_response_handler(join_node_response_params params) {
-    scylla_assert(this_shard_id() == 0);
+    SCYLLA_ASSERT(this_shard_id() == 0);
 
     // Usually this handler will only run once, but there are some cases where we might get more than one RPC,
     // possibly happening at the same time, e.g.:

service/task_manager_module.cc

@@ -13,6 +13,7 @@
 #include "service/task_manager_module.hh"
 #include "tasks/task_handler.hh"
 #include "tasks/virtual_task_hint.hh"
+#include "utils/UUID_gen.hh"
 #include <seastar/coroutine/maybe_yield.hh>
 
 namespace service {
@@ -57,9 +58,14 @@ static std::optional<tasks::task_stats> maybe_make_task_stats(const locator::tab
         .kind = tasks::task_kind::cluster,
         .scope = get_scope(task_info.request_type),
         .state = tasks::task_manager::task_state::running,
+        .sequence_number = 0,
         .keyspace = schema->ks_name(),
         .table = schema->cf_name(),
-        .start_time = task_info.request_time
+        .entity = "",
+        .shard = 0,
+        .creation_time = task_info.request_time,
+        .start_time = task_info.sched_time,
+        .end_time = db_clock::time_point{}
     };
 }
 
@@ -225,7 +231,8 @@ static void update_status(const locator::tablet_task_info& task_info, tasks::tas
     sched_nr += task_info.sched_nr;
     status.type = locator::tablet_task_type_to_string(task_info.request_type);
     status.scope = get_scope(task_info.request_type);
-    status.start_time = task_info.request_time;
+    status.creation_time = task_info.request_time;
+    status.start_time = task_info.sched_time;
 }
 
 future<std::optional<status_helper>> tablet_virtual_task::get_status_helper(tasks::task_id id, tasks::virtual_task_hint hint) {

service/topology_coordinator.cc

@@ -360,7 +360,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
         auto& topo = _topo_sm._topology;
 
         auto it = topo.find(id);
-        scylla_assert(it);
+        SCYLLA_ASSERT(it);
 
         std::optional<topology_request> req;
         auto rit = topo.requests.find(id);
@@ -956,6 +956,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
             req_entry = co_await _sys_ks.get_topology_request_entry(req_id, true);
             req = std::get<global_topology_request>(req_entry.request_type);
         }
+        
         switch (req) {
         case global_topology_request::new_cdc_generation: {
             rtlogger.info("new CDC generation requested");
@@ -975,9 +976,14 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                    .set_global_topology_request(req)
                    .set_global_topology_request_id(req_id)
                    .drop_first_global_topology_request_id(_topo_sm._topology.global_requests_queue, req_id);
+            
+            // Set start_time when we begin executing the request
+            topology_request_tracking_mutation_builder rtbuilder(req_id);
+            rtbuilder.set("start_time", db_clock::now());
+            
             auto reason = ::format(
                 "insert CDC generation data (UUID: {})", gen_uuid);
-            co_await update_topology_state(std::move(guard), {std::move(mutation), builder.build()}, reason);
+            co_await update_topology_state(std::move(guard), {std::move(mutation), builder.build(), rtbuilder.build()}, reason);
         }
         break;
         case global_topology_request::cleanup:
@@ -1068,7 +1074,9 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                                                          .del_global_topology_request_id()
                                                          .drop_first_global_topology_request_id(_topo_sm._topology.global_requests_queue, req_id)
                                                          .build()));
+            // Set start_time when we begin executing the request and mark as done
             updates.push_back(canonical_mutation(topology_request_tracking_mutation_builder(req_id)
+                                                         .set("start_time", db_clock::now())
                                                          .done(error)
                                                          .build()));
 
@@ -1088,7 +1096,12 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                    .set_global_topology_request_id(req_id)
                    .drop_first_global_topology_request_id(_topo_sm._topology.global_requests_queue, req_id)
                    .set_session(session_id(req_id));
-            co_await update_topology_state(std::move(guard), {builder.build()}, "TRUNCATE TABLE requested");
+            
+            // Set start_time when we begin executing the request
+            topology_request_tracking_mutation_builder rtbuilder(req_id);
+            rtbuilder.set("start_time", db_clock::now());
+            
+            co_await update_topology_state(std::move(guard), {builder.build(), rtbuilder.build()}, "TRUNCATE TABLE requested");
         }
         break;
         }
@@ -2310,7 +2323,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
 
                 switch (node.rs->state) {
                     case node_state::bootstrapping: {
-                        scylla_assert(!node.rs->ring);
+                        SCYLLA_ASSERT(!node.rs->ring);
                         auto num_tokens = std::get<join_param>(node.req_param.value()).num_tokens;
                         auto tokens_string = std::get<join_param>(node.req_param.value()).tokens_string;
 
@@ -2359,11 +2372,11 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                     }
                         break;
                     case node_state::replacing: {
-                        scylla_assert(!node.rs->ring);
+                        SCYLLA_ASSERT(!node.rs->ring);
                         auto replaced_id = std::get<replace_param>(node.req_param.value()).replaced_id;
                         auto it = _topo_sm._topology.normal_nodes.find(replaced_id);
-                        scylla_assert(it != _topo_sm._topology.normal_nodes.end());
-                        scylla_assert(it->second.ring && it->second.state == node_state::normal);
+                        SCYLLA_ASSERT(it != _topo_sm._topology.normal_nodes.end());
+                        SCYLLA_ASSERT(it->second.ring && it->second.state == node_state::normal);
 
                         topology_mutation_builder builder(node.guard.write_timestamp());
 
@@ -3022,7 +3035,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                 rtbuilder.set("start_time", db_clock::now());
                 switch (node.request.value()) {
                     case topology_request::join: {
-                        scylla_assert(!node.rs->ring);
+                        SCYLLA_ASSERT(!node.rs->ring);
                         // Write chosen tokens through raft.
                         builder.set_transition_state(topology::transition_state::join_group0)
                                .with_node(node.id)
@@ -3033,7 +3046,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                         break;
                         }
                     case topology_request::leave:
-                        scylla_assert(node.rs->ring);
+                        SCYLLA_ASSERT(node.rs->ring);
                         // start decommission and put tokens of decommissioning nodes into write_both_read_old state
                         // meaning that reads will go to the replica being decommissioned
                         // but writes will go to new owner as well
@@ -3046,7 +3059,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                                                        "start decommission");
                         break;
                     case topology_request::remove: {
-                        scylla_assert(node.rs->ring);
+                        SCYLLA_ASSERT(node.rs->ring);
 
                         builder.set_transition_state(topology::transition_state::tablet_draining)
                                .set_version(_topo_sm._topology.version + 1)
@@ -3058,7 +3071,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
                         break;
                         }
                     case topology_request::replace: {
-                        scylla_assert(!node.rs->ring);
+                        SCYLLA_ASSERT(!node.rs->ring);
 
                         builder.set_transition_state(topology::transition_state::join_group0)
                                .with_node(node.id)
@@ -3163,7 +3176,7 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
 
         auto id = node.id;
 
-        scylla_assert(!_topo_sm._topology.transition_nodes.empty());
+        SCYLLA_ASSERT(!_topo_sm._topology.transition_nodes.empty());
 
         release_node(std::move(node));
 
@@ -3279,6 +3292,11 @@ class topology_coordinator : public endpoint_lifecycle_subscriber {
             topology_mutation_builder builder(guard.write_timestamp());
             builder.del_global_topology_request();
             if (_feature_service.topology_global_request_queue) {
+                // Set start_time when we begin executing the request
+                topology_request_tracking_mutation_builder start_rtbuilder(*global_request_id);
+                start_rtbuilder.set("start_time", db_clock::now());
+                muts.emplace_back(start_rtbuilder.build());
+                
                 topology_request_tracking_mutation_builder rtbuilder(*global_request_id);
                 builder.del_global_topology_request_id()
                        .drop_first_global_topology_request_id(_topo_sm._topology.global_requests_queue, *global_request_id);
@@ -4013,7 +4031,7 @@ future<> topology_coordinator::stop() {
         // but let's check all of them because we never reset these holders
         // once they are added as barriers
         for (auto& [stage, barrier]: tablet_state.barriers) {
-            scylla_assert(barrier.has_value());
+            SCYLLA_ASSERT(barrier.has_value());
             co_await stop_background_action(barrier, gid, [stage] { return format("at stage {}", tablet_transition_stage_to_string(stage)); });
         }
 

sstables/compress.cc

@@ -251,7 +251,7 @@ void compression::discard_hidden_options() {
 }
 
 compressor& compression::get_compressor() const {
-    scylla_assert(_compressor);
+    SCYLLA_ASSERT(_compressor);
     return *_compressor.get();
 }
 

sstables/compress.hh

@@ -170,7 +170,7 @@ struct compression {
             const_iterator(const const_iterator& other) = default;
 
             const_iterator& operator=(const const_iterator& other) {
-                scylla_assert(&_offsets == &other._offsets);
+                SCYLLA_ASSERT(&_offsets == &other._offsets);
                 _index = other._index;
                 return *this;
             }

sstables/compressor.cc

@@ -24,7 +24,6 @@
 #include "sstables/sstable_compressor_factory.hh"
 #include "compressor.hh"
 #include "exceptions/exceptions.hh"
-#include "utils/assert.hh"
 #include "utils/config_file_impl.hh"
 #include "utils/class_registrator.hh"
 #include "gms/feature_service.hh"
@@ -296,7 +295,7 @@ size_t zstd_processor::uncompress(const char* input, size_t input_len, char* out
         if (_ddict) {
             return ZSTD_decompress_usingDDict(dctx, output, output_len, input, input_len, _ddict->dict());
         } else {
-            scylla_assert(!_cdict && "Write-only compressor used for reading");
+            SCYLLA_ASSERT(!_cdict && "Write-only compressor used for reading");
             return ZSTD_decompressDCtx(dctx, output, output_len, input, input_len);
         }
     });
@@ -311,7 +310,7 @@ size_t zstd_processor::compress(const char* input, size_t input_len, char* outpu
         if (_cdict) {
             return ZSTD_compress_usingCDict(cctx, output, output_len, input, input_len, _cdict->dict());
         } else {
-            scylla_assert(!_ddict && "Read-only compressor used for writing");
+            SCYLLA_ASSERT(!_ddict && "Read-only compressor used for writing");
             return ZSTD_compressCCtx(cctx, output, output_len, input, input_len, _compression_level);
         }
     });
@@ -628,7 +627,7 @@ size_t lz4_processor::uncompress(const char* input, size_t input_len,
     if (_ddict) {
         ret = LZ4_decompress_safe_usingDict(input, output, input_len, output_len, reinterpret_cast<const char*>(_ddict->raw().data()), _ddict->raw().size());
     } else {
-        scylla_assert(!_cdict && "Write-only compressor used for reading");
+        SCYLLA_ASSERT(!_cdict && "Write-only compressor used for reading");
         ret = LZ4_decompress_safe(input, output, input_len, output_len);
     }
     if (ret < 0) {
@@ -658,7 +657,7 @@ size_t lz4_processor::compress(const char* input, size_t input_len,
             LZ4_resetStream_fast(ctx);
         }
     } else {
-        scylla_assert(!_ddict && "Read-only compressor used for writing");
+        SCYLLA_ASSERT(!_ddict && "Read-only compressor used for writing");
         ret = LZ4_compress_default(input, output + 4, input_len, LZ4_compressBound(input_len));
     }
     if (ret == 0) {
@@ -1269,7 +1268,7 @@ lz4_cdict::~lz4_cdict() {
 }
 
 std::unique_ptr<sstable_compressor_factory> make_sstable_compressor_factory_for_tests_in_thread() {
-    scylla_assert(thread::running_in_thread());
+    SCYLLA_ASSERT(thread::running_in_thread());
     struct wrapper : sstable_compressor_factory {
         using impl = default_sstable_compressor_factory;
         sharded<impl> _impl;

sstables/downsampling.hh

@@ -44,14 +44,14 @@ public:
      * @return A list of `sampling_level` unique indices between 0 and `sampling_level`
      */
     static const std::vector<int>& get_sampling_pattern(int sampling_level) {
-        scylla_assert(sampling_level > 0 && sampling_level <= BASE_SAMPLING_LEVEL);
+        SCYLLA_ASSERT(sampling_level > 0 && sampling_level <= BASE_SAMPLING_LEVEL);
         auto& entry = _sample_pattern_cache[sampling_level-1];
         if (!entry.empty()) {
             return entry;
         }
 
         if (sampling_level <= 1) {
-            scylla_assert(_sample_pattern_cache[0].empty());
+            SCYLLA_ASSERT(_sample_pattern_cache[0].empty());
             _sample_pattern_cache[0].push_back(0);
             return _sample_pattern_cache[0];
         }
@@ -96,7 +96,7 @@ public:
      * @return a list of original indexes for current summary entries
      */
     static const std::vector<int>& get_original_indexes(int sampling_level) {
-        scylla_assert(sampling_level > 0 && sampling_level <= BASE_SAMPLING_LEVEL);
+        SCYLLA_ASSERT(sampling_level > 0 && sampling_level <= BASE_SAMPLING_LEVEL);
         auto& entry = _original_index_cache[sampling_level-1];
         if (!entry.empty()) {
             return entry;
@@ -128,7 +128,7 @@ public:
      * @return the number of partitions before the next index summary entry, inclusive on one end
      */
     static int get_effective_index_interval_after_index(int index, int sampling_level, int min_index_interval) {
-        scylla_assert(index >= -1);
+        SCYLLA_ASSERT(index >= -1);
         const std::vector<int>& original_indexes = get_original_indexes(sampling_level);
         if (index == -1) {
             return original_indexes[0] * min_index_interval;

sstables/exceptions.hh

@@ -31,7 +31,7 @@ public:
 [[noreturn]] void on_parse_error(sstring message, std::optional<component_name> filename);
 [[noreturn, gnu::noinline]] void on_bti_parse_error(uint64_t pos);
 
-// Use this instead of scylla_assert() or assert() in code that is used while parsing SSTables.
+// Use this instead of SCYLLA_ASSERT() or assert() in code that is used while parsing SSTables.
 // SSTables can be corrupted either by ScyllaDB itself or by a freak accident like cosmic background
 // radiation hitting the disk the wrong way. Either way a corrupt SSTable should not bring down the
 // whole server. This method will call on_internal_error() if the condition is false.

sstables/generation_type.hh

@@ -129,7 +129,7 @@ public:
     /// way to determine that is overlapping its partition-ranges with the shard's
     /// owned ranges.
     static bool maybe_owned_by_this_shard(const sstables::generation_type& gen) {
-        scylla_assert(bool(gen));
+        SCYLLA_ASSERT(bool(gen));
         int64_t hint = 0;
         if (gen.is_uuid_based()) {
             hint = std::hash<utils::UUID>{}(gen.as_uuid());

sstables/index_reader.hh

@@ -57,7 +57,10 @@ public:
     index_list indexes;
 
     index_consumer(logalloc::region& r, schema_ptr s)
-        : _s(std::move(s))
+        : _s(s)
+        , _alloc_section(abstract_formatter([s] (fmt::format_context& ctx) {
+            fmt::format_to(ctx.out(), "index_consumer {}.{}", s->ks_name(), s->cf_name());
+        }))
         , _region(r)
     { }
 
@@ -785,6 +788,9 @@ public:
                                                       _sstable->manager().get_cache_tracker().region(),
                                                       _sstable->manager().get_cache_tracker().get_partition_index_cache_stats()))
         , _index_cache(caching ? *_sstable->_index_cache : *_local_index_cache)
+        , _alloc_section(abstract_formatter([sst = _sstable] (fmt::format_context& ctx) {
+            fmt::format_to(ctx.out(), "index_reader {}", sst->get_filename());
+        }))
         , _region(_sstable->manager().get_cache_tracker().region())
         , _use_caching(caching)
         , _single_page_read(single_partition_read) // all entries for a given partition are within a single page

sstables/mx/bsearch_clustered_cursor.hh

@@ -284,6 +284,9 @@ public:
         , _clustering_parser(s, permit, _ctr.clustering_column_value_fix_legths(), true)
         , _block_parser(s, permit, _ctr.clustering_column_value_fix_legths())
         , _permit(std::move(permit))
+        , _as(abstract_formatter([s] (fmt::format_context& ctx) {
+            fmt::format_to(ctx.out(), "cached_promoted_index {}.{}", s.ks_name(), s.cf_name());
+        }))
     { }
 
     ~cached_promoted_index() {

sstables/mx/writer.cc

@@ -91,7 +91,7 @@ public:
         {}
 
         void increment() {
-            scylla_assert(_range);
+            SCYLLA_ASSERT(_range);
             if (!_range->next()) {
                 _range = nullptr;
             }
@@ -102,7 +102,7 @@ public:
         }
 
         const ValueType dereference() const {
-            scylla_assert(_range);
+            SCYLLA_ASSERT(_range);
             return _range->get_value();
         }
 
@@ -153,7 +153,7 @@ public:
         auto limit = std::min(_serialization_limit_size, _offset + clustering_block::max_block_size);
 
         _current_block = {};
-        scylla_assert (_offset % clustering_block::max_block_size == 0);
+        SCYLLA_ASSERT (_offset % clustering_block::max_block_size == 0);
         while (_offset < limit) {
             auto shift = _offset % clustering_block::max_block_size;
             if (_offset < _prefix.size(_schema)) {
@@ -280,7 +280,7 @@ public:
                     ++_current_index;
                 }
             } else {
-                scylla_assert(_mode == encoding_mode::large_encode_missing);
+                SCYLLA_ASSERT(_mode == encoding_mode::large_encode_missing);
                 while (_current_index < total_size) {
                     auto cell = _row.find_cell(_columns[_current_index].get().id);
                     if (!cell) {
@@ -1180,7 +1180,7 @@ void writer::write_cell(bytes_ostream& writer, const clustering_key_prefix* clus
 
     if (cdef.is_counter()) {
         if (!is_deleted) {
-            scylla_assert(!cell.is_counter_update());
+            SCYLLA_ASSERT(!cell.is_counter_update());
             auto ccv = counter_cell_view(cell);
             write_counter_value(ccv, writer, _sst.get_version(), [] (bytes_ostream& out, uint32_t value) {
                 return write_vint(out, value);
@@ -1489,7 +1489,7 @@ template <typename W>
 requires Writer<W>
 static void write_clustering_prefix(sstable_version_types v, W& writer, bound_kind_m kind,
     const schema& s, const clustering_key_prefix& clustering) {
-    scylla_assert(kind != bound_kind_m::static_clustering);
+    SCYLLA_ASSERT(kind != bound_kind_m::static_clustering);
     write(v, writer, kind);
     auto is_ephemerally_full = ephemerally_full_prefix{s.is_compact_table()};
     if (kind != bound_kind_m::clustering) {

sstables/partition_index_cache.hh

@@ -59,7 +59,7 @@ private:
                 // Live entry_ptr should keep the entry alive, except when the entry failed on loading.
                 // In that case, entry_ptr holders are not supposed to use the pointer, so it's safe
                 // to nullify those entry_ptrs.
-                scylla_assert(!ready());
+                SCYLLA_ASSERT(!ready());
             }
         }
 

sstables/sstable_directory.cc

@@ -496,7 +496,7 @@ sstable_directory::move_foreign_sstables(sharded<sstable_directory>& source_dire
             return make_ready_future<>();
         }
         // Should be empty, since an SSTable that belongs to this shard is not remote.
-        scylla_assert(shard_id != this_shard_id());
+        SCYLLA_ASSERT(shard_id != this_shard_id());
         dirlog.debug("Moving {} unshared SSTables of {}.{} to shard {} ", info_vec.size(), _schema->ks_name(), _schema->cf_name(), shard_id);
         return source_directory.invoke_on(shard_id, &sstables::sstable_directory::load_foreign_sstables, std::move(info_vec));
     });
@@ -540,7 +540,7 @@ sstable_directory::collect_output_unshared_sstables(std::vector<sstables::shared
     dirlog.debug("Collecting {} output SSTables (remote={})", resharded_sstables.size(), remote_ok);
     return parallel_for_each(std::move(resharded_sstables), [this, remote_ok] (sstables::shared_sstable sst) {
         auto shards = sst->get_shards_for_this_sstable();
-        scylla_assert(shards.size() == 1);
+        SCYLLA_ASSERT(shards.size() == 1);
         auto shard = shards[0];
 
         if (shard == this_shard_id()) {

sstables/sstable_set.cc

@@ -283,7 +283,7 @@ bool partitioned_sstable_set::store_as_unleveled(const shared_sstable& sst) cons
         }
         sstlog.info("SSTable {}, as_unleveled={}, expect_unleveled={}, sst_tr={}, overlap_ratio={}",
             sst->generation(), as_unleveled, expect_unleveled, sst_tr, dht::overlap_ratio(_token_range, sst_tr));
-        scylla_assert(as_unleveled == expect_unleveled);
+        SCYLLA_ASSERT(as_unleveled == expect_unleveled);
     });
 
     return as_unleveled;
@@ -712,8 +712,8 @@ public:
 
         // by !empty(bound) and `_it` invariant:
         //      _it != _end, _it->first <= bound, and filter(*_it->second) == true
-        scylla_assert(_cmp(_it->first, bound) <= 0);
-        // we don't scylla_assert(filter(*_it->second)) due to the requirement that `filter` is called at most once for each sstable
+        SCYLLA_ASSERT(_cmp(_it->first, bound) <= 0);
+        // we don't SCYLLA_ASSERT(filter(*_it->second)) due to the requirement that `filter` is called at most once for each sstable
 
         // Find all sstables with the same position as `_it` (they form a contiguous range in the container).
         auto next = std::find_if(std::next(_it), _end, [this] (const value_t& v) { return _cmp(v.first, _it->first) != 0; });
@@ -1301,7 +1301,7 @@ sstable_set::create_single_key_sstable_reader(
         mutation_reader::forwarding fwd_mr,
         const sstable_predicate& predicate,
         sstables::integrity_check integrity) const {
-    scylla_assert(pr.is_singular() && pr.start()->value().has_key());
+    SCYLLA_ASSERT(pr.is_singular() && pr.start()->value().has_key());
     return _impl->create_single_key_sstable_reader(cf, std::move(schema),
             std::move(permit), sstable_histogram, pr, slice, std::move(trace_state), fwd, fwd_mr, predicate, integrity);
 }
@@ -1408,7 +1408,7 @@ sstable_set::make_local_shard_sstable_reader(
 {
     auto reader_factory_fn = [s, permit, &slice, trace_state, fwd, fwd_mr, &monitor_generator, &predicate, integrity]
             (shared_sstable& sst, const dht::partition_range& pr) mutable {
-        scylla_assert(!sst->is_shared());
+        SCYLLA_ASSERT(!sst->is_shared());
         if (!predicate(*sst)) {
             return make_empty_mutation_reader(s, permit);
         }

sstables/sstables.cc

@@ -36,7 +36,6 @@
 
 #include "utils/error_injection.hh"
 #include "utils/to_string.hh"
-#include "utils/assert.hh"
 #include "data_dictionary/storage_options.hh"
 #include "dht/sharder.hh"
 #include "writer.hh"
@@ -4162,7 +4161,7 @@ future<data_sink> file_io_extension::wrap_sink(const sstable& sst, component_typ
 }
 
 future<data_source> file_io_extension::wrap_source(const sstable& sst, component_type c, data_source) {
-    scylla_assert(0 && "You are not supposed to get here, file_io_extension::wrap_source() is not implemented");
+    SCYLLA_ASSERT(0 && "You are not supposed to get here, file_io_extension::wrap_source() is not implemented");
 }
 
 namespace trie {

sstables/sstables_manager.cc

@@ -55,9 +55,9 @@ sstables_manager::sstables_manager(
 }
 
 sstables_manager::~sstables_manager() {
-    scylla_assert(_closing);
-    scylla_assert(_active.empty());
-    scylla_assert(_undergoing_close.empty());
+    SCYLLA_ASSERT(_closing);
+    SCYLLA_ASSERT(_active.empty());
+    SCYLLA_ASSERT(_undergoing_close.empty());
 }
 
 void sstables_manager::subscribe(sstables_manager_event_handler& handler) {

sstables/sstables_manager.hh

@@ -185,12 +185,12 @@ public:
             size_t buffer_size = default_sstable_buffer_size);
 
     shared_ptr<object_storage_client> get_endpoint_client(sstring endpoint) const {
-        scylla_assert(_storage != nullptr);
+        SCYLLA_ASSERT(_storage != nullptr);
         return _storage->get_endpoint_client(std::move(endpoint));
     }
 
     bool is_known_endpoint(sstring endpoint) const {
-        scylla_assert(_storage != nullptr);
+        SCYLLA_ASSERT(_storage != nullptr);
         return _storage->is_known_endpoint(std::move(endpoint));
     }
 
@@ -241,7 +241,7 @@ public:
 
     // Only for sstable::storage usage
     sstables::sstables_registry& sstables_registry() const noexcept {
-        scylla_assert(_sstables_registry && "sstables_registry is not plugged");
+        SCYLLA_ASSERT(_sstables_registry && "sstables_registry is not plugged");
         return *_sstables_registry;
     }
 

sstables/storage.cc

@@ -109,7 +109,7 @@ future<data_sink> filesystem_storage::make_data_or_index_sink(sstable& sst, comp
     options.buffer_size = sst.sstable_buffer_size;
     options.write_behind = 10;
 
-    scylla_assert(
+    SCYLLA_ASSERT(
         type == component_type::Data
         || type == component_type::Index
         || type == component_type::Rows
@@ -129,7 +129,7 @@ future<data_sink> filesystem_storage::make_data_or_index_sink(sstable& sst, comp
 }
 
 future<data_source> filesystem_storage::make_data_or_index_source(sstable&, component_type type, file f, uint64_t offset, uint64_t len, file_input_stream_options opt) const {
-    scylla_assert(type == component_type::Data || type == component_type::Index);
+    SCYLLA_ASSERT(type == component_type::Data || type == component_type::Index);
     co_return make_file_data_source(std::move(f), offset, len, std::move(opt));
 }
 
@@ -717,7 +717,7 @@ static future<data_source> maybe_wrap_source(const sstable& sst, component_type
 }
 
 future<data_sink> object_storage_base::make_data_or_index_sink(sstable& sst, component_type type) {
-    scylla_assert(
+    SCYLLA_ASSERT(
         type == component_type::Data
         || type == component_type::Index
         || type == component_type::Rows

sstables/storage.hh

@@ -83,13 +83,13 @@ class storage {
 
     // Internal, but can also be used by tests
     virtual future<> change_dir_for_test(sstring nd) {
-        scylla_assert(false && "Changing directory not implemented");
+        SCYLLA_ASSERT(false && "Changing directory not implemented");
     }
     virtual future<> create_links(const sstable& sst, const std::filesystem::path& dir) const {
-        scylla_assert(false && "Direct links creation not implemented");
+        SCYLLA_ASSERT(false && "Direct links creation not implemented");
     }
     virtual future<> move(const sstable& sst, sstring new_dir, generation_type generation, delayed_commit_changes* delay) {
-        scylla_assert(false && "Direct move not implemented");
+        SCYLLA_ASSERT(false && "Direct move not implemented");
     }
 
 public:

sstables/trie/bti_key_translation.cc

@@ -8,7 +8,6 @@
 
 #include "bti_key_translation.hh"
 #include "sstables/mx/types.hh"
-#include "utils/assert.hh"
 
 namespace sstables::trie {
 
@@ -57,7 +56,7 @@ void lazy_comparable_bytes_from_ring_position::init_first_fragment(dht::token dh
 }
 
 void lazy_comparable_bytes_from_ring_position::trim(const size_t n) {
-    scylla_assert(n <= _size);
+    SCYLLA_ASSERT(n <= _size);
     _size = n;
 }
 
@@ -128,7 +127,7 @@ lazy_comparable_bytes_from_clustering_position::lazy_comparable_bytes_from_clust
 {}
 
 void lazy_comparable_bytes_from_clustering_position::trim(unsigned n) {
-    scylla_assert(n <= _size);
+    SCYLLA_ASSERT(n <= _size);
     _size = n;
 }
 

sstables/trie/bti_node_reader.cc

@@ -8,7 +8,6 @@
 
 #include "bti_node_reader.hh"
 #include "bti_node_type.hh"
-#include "utils/assert.hh"
 
 namespace sstables::trie {
 
@@ -449,37 +448,37 @@ seastar::future<> bti_node_reader::load(int64_t pos, const reader_permit& permit
 }
 
 trie::load_final_node_result bti_node_reader::read_node(int64_t pos) {
-    scylla_assert(cached(pos));
+    SCYLLA_ASSERT(cached(pos));
     auto sp = _cached_page->get_view().subspan(pos % cached_file::page_size);
     return bti_read_node(pos, sp);
 }
 
 trie::node_traverse_result bti_node_reader::walk_down_along_key(int64_t pos, const_bytes key) {
-    scylla_assert(cached(pos));
+    SCYLLA_ASSERT(cached(pos));
     auto sp = _cached_page->get_view().subspan(pos % cached_file::page_size);
     return bti_walk_down_along_key(pos, sp, key);
 }
 
 trie::node_traverse_sidemost_result bti_node_reader::walk_down_leftmost_path(int64_t pos) {
-    scylla_assert(cached(pos));
+    SCYLLA_ASSERT(cached(pos));
     auto sp = _cached_page->get_view().subspan(pos % cached_file::page_size);
     return bti_walk_down_leftmost_path(pos, sp);
 }
 
 trie::node_traverse_sidemost_result bti_node_reader::walk_down_rightmost_path(int64_t pos) {
-    scylla_assert(cached(pos));
+    SCYLLA_ASSERT(cached(pos));
     auto sp = _cached_page->get_view().subspan(pos % cached_file::page_size);
     return bti_walk_down_rightmost_path(pos, sp);
 }
 
 trie::get_child_result bti_node_reader::get_child(int64_t pos, int child_idx, bool forward) const {
-    scylla_assert(cached(pos));
+    SCYLLA_ASSERT(cached(pos));
     auto sp = _cached_page->get_view().subspan(pos % cached_file::page_size);
     return bti_get_child(pos, sp, child_idx, forward);
 }
 
 const_bytes bti_node_reader::get_payload(int64_t pos) const {
-    scylla_assert(cached(pos));
+    SCYLLA_ASSERT(cached(pos));
     auto sp = _cached_page->get_view().subspan(pos % cached_file::page_size);
     return bti_get_payload(pos, sp);
 }

sstables/trie/trie_traversal.hh

@@ -204,7 +204,7 @@ inline void descend_leftmost_single_page(
             next_pos = -1;
             trail.back().child_idx = -1;
         } else {
-            scylla_assert(traverse_one.n_children >= 1);
+            SCYLLA_ASSERT(traverse_one.n_children >= 1);
             next_pos = traverse_one.body_pos - traverse_one.child_offset;
         }
     }

sstables/trie/trie_writer.cc

@@ -9,7 +9,6 @@
 #include <seastar/util/log.hh>
 #include "writer_node.hh"
 #include "common.hh"
-#include "utils/assert.hh"
 
 seastar::logger trie_logger("trie");
 
@@ -28,7 +27,7 @@ auto writer_node::create(const_bytes b, bump_allocator& alctr) -> ptr<writer_nod
 }
 
 auto writer_node::add_child(const_bytes b, bump_allocator& alctr) -> ptr<writer_node> {
-    scylla_assert(get_children().empty() || b[0] > get_children().back()->_transition[0]);
+    SCYLLA_ASSERT(get_children().empty() || b[0] > get_children().back()->_transition[0]);
     reserve_children(get_children().size() + 1, alctr);
     auto new_child = create(b, alctr);
     push_child(new_child, alctr);

sstables/trie/trie_writer.hh

@@ -406,7 +406,7 @@ inline void trie_writer<Output>::complete_until_depth(size_t depth) {
 
 template <trie_writer_sink Output>
 inline void trie_writer<Output>::add(size_t depth, const_bytes key_tail, const trie_payload& p) {
-    scylla_assert(p._payload_bits);
+    SCYLLA_ASSERT(p._payload_bits);
     add_partial(depth, key_tail);
     _stack.back()->set_payload(p);
 }
@@ -416,10 +416,10 @@ template <trie_writer_sink Output>
 inline void trie_writer<Output>::add_partial(size_t depth, const_bytes key_frag) {
     expensive_log("writer_node::add_partial: end, stack={}, depth={}, _current_depth={} tail={}", _stack.size(), depth, _current_depth, fmt_hex(key_frag));
     expensive_assert(_stack.size() >= 1);
-    scylla_assert(_current_depth >= depth);
+    SCYLLA_ASSERT(_current_depth >= depth);
     // There is only one case where a zero-length tail is legal:
     // when inserting the empty key.
-    scylla_assert(!key_frag.empty() || depth == 0);
+    SCYLLA_ASSERT(!key_frag.empty() || depth == 0);
 
     complete_until_depth(depth);
     if (key_frag.size()) {
@@ -444,7 +444,7 @@ inline sink_pos trie_writer<Output>::finish() {
     if (!try_write(_stack[0])) {
         _out.pad_to_page_boundary();
         bool ok = try_write(_stack[0]);
-        scylla_assert(ok);
+        SCYLLA_ASSERT(ok);
     }
     auto root_pos = _stack[0]->_pos;
 

sstables/trie/writer_node.hh

@@ -203,7 +203,7 @@ private:
     [[nodiscard]] ptr<T> alloc_impl(size_t n) {
         using value_type = ptr<T>::value_type;
         expensive_assert(n < _segment_size / sizeof(value_type));
-        scylla_assert(n > 0);
+        SCYLLA_ASSERT(n > 0);
         auto sz = n * sizeof(value_type);
         _remaining -= _remaining % alignof(value_type);
         if (sz > _remaining) [[unlikely]] {
@@ -230,7 +230,7 @@ private:
 
 public:
     bump_allocator(size_t segment_size) : _segment_size(segment_size) {
-        scylla_assert(_segment_size % alignof(max_align_t) == 0);
+        SCYLLA_ASSERT(_segment_size % alignof(max_align_t) == 0);
     }
 
     // Total memory usage by this allocator.

sstables/trie/writer_node.impl.hh

@@ -9,7 +9,6 @@
 #pragma once
 
 #include "writer_node.hh"
-#include "utils/assert.hh"
 #include "utils/small_vector.hh"
 
 namespace sstables::trie {
@@ -112,9 +111,9 @@ void writer_node::write(ptr<writer_node> self, Output& out, bool guaranteed_fit)
             fmt::ptr(node.get()), out.pos().value, node->get_children().size(), node->_node_size.value, node->_transition_length);
 
         if (guaranteed_fit) {
-            scylla_assert(out.pos() - startpos == node->_branch_size);
+            SCYLLA_ASSERT(out.pos() - startpos == node->_branch_size);
             node->_pos = sink_pos(out.write(*node, sink_pos(out.pos())));
-            scylla_assert(out.pos() - startpos == node->_branch_size + node->_node_size);
+            SCYLLA_ASSERT(out.pos() - startpos == node->_branch_size + node->_node_size);
         } else {
             if (uint64_t(out.serialized_size(*node, sink_pos(out.pos())).value) > out.bytes_left_in_page()) {
                 out.pad_to_page_boundary();

tasks/task_handler.cc

@@ -10,6 +10,7 @@
 #include "tasks/task_handler.hh"
 #include "tasks/virtual_task_hint.hh"
 #include "utils/overloaded_functor.hh"
+#include "utils/UUID_gen.hh"
 
 #include <seastar/core/with_timeout.hh>
 
@@ -19,6 +20,11 @@ namespace tasks {
 
 using task_status_variant = std::variant<tasks::task_manager::foreign_task_ptr, tasks::task_manager::task::task_essentials>;
 
+static db_clock::time_point get_creation_time_from_task_id(task_id id) {
+    // Task IDs are timeuuids (version 1 UUIDs), so we can extract the timestamp from them
+    return db_clock::time_point(utils::UUID_gen::unix_timestamp(id.uuid()));
+}
+
 static future<task_status> get_task_status(task_manager::task_ptr task) {
     auto host_id = task->get_module()->get_task_manager().get_host_id();
     auto local_task_status = task->get_status();
@@ -29,6 +35,7 @@ static future<task_status> get_task_status(task_manager::task_ptr task) {
         .scope = local_task_status.scope,
         .state = local_task_status.state,
         .is_abortable = task->is_abortable(),
+        .creation_time = get_creation_time_from_task_id(local_task_status.id),
         .start_time = local_task_status.start_time,
         .end_time = local_task_status.end_time,
         .error = local_task_status.error,
@@ -173,6 +180,7 @@ future<utils::chunked_vector<task_status>> task_handler::get_status_recursively(
                     .scope = task.task_status.scope,
                     .state = task.task_status.state,
                     .is_abortable = task.abortable,
+                    .creation_time = get_creation_time_from_task_id(task.task_status.id),
                     .start_time = task.task_status.start_time,
                     .end_time = task.task_status.end_time,
                     .error = task.task_status.error,

tasks/task_handler.hh

@@ -26,6 +26,7 @@ struct task_status {
     std::string scope;
     task_manager::task_state state;
     is_abortable is_abortable;
+    db_clock::time_point creation_time;
     db_clock::time_point start_time;
     db_clock::time_point end_time;
     std::string error;
@@ -51,6 +52,7 @@ struct task_stats {
     std::string table;
     std::string entity;
     unsigned shard;
+    db_clock::time_point creation_time;
     db_clock::time_point start_time;
     db_clock::time_point end_time;
 };

tasks/task_manager.cc

@@ -21,6 +21,7 @@
 #include "utils/assert.hh"
 #include "utils/chunked_vector.hh"
 #include "utils/overloaded_functor.hh"
+#include "utils/UUID_gen.hh"
 #include "service/storage_service.hh"
 #include "tasks/task_handler.hh"
 #include "task_manager.hh"
@@ -559,6 +560,7 @@ future<utils::chunked_vector<task_stats>> task_manager::module::get_stats(is_int
                 .table = task->get_status().table,
                 .entity = task->get_status().entity,
                 .shard = task->get_status().shard,
+                .creation_time = db_clock::time_point(utils::UUID_gen::unix_timestamp(task->id().uuid())),
                 .start_time = task->get_status().start_time,
                 .end_time = task->get_status().end_time,
             });

test/alternator/test_batch.py

@@ -205,7 +205,7 @@ def test_batch_write_invalid_operation(test_table_s):
 
 # In test_item.py we have a bunch of test_empty_* tests on different ways to
 # create an empty item (which in Scylla requires the special CQL row marker
-# to be supported correctly). BatchWriteItems provides yet another way of
+# to be supported correctly). BatchWriteItem provides yet another way of
 # creating items, so check the empty case here too:
 def test_empty_batch_write(test_table):
     p = random_string()
@@ -214,7 +214,7 @@ def test_empty_batch_write(test_table):
         batch.put_item({'p': p, 'c': c})
     assert test_table.get_item(Key={'p': p, 'c': c}, ConsistentRead=True)['Item'] == {'p': p, 'c': c}
 
-# Test that BatchWriteItems allows writing to multiple tables in one operation
+# Test that BatchWriteItem allows writing to multiple tables in one operation
 def test_batch_write_multiple_tables(test_table_s, test_table):
     p1 = random_string()
     c1 = random_string()

test/cluster/conftest.py

@@ -13,7 +13,8 @@ import ssl
 import tempfile
 import platform
 import urllib.parse
-from multiprocessing import Event, Process
+from concurrent.futures.thread import ThreadPoolExecutor
+from multiprocessing import Event
 from pathlib import Path
 from typing import TYPE_CHECKING
 from test.pylib.runner import testpy_test_fixture_scope
@@ -186,15 +187,14 @@ async def manager_api_sock_path(request: pytest.FixtureRequest, testpy_test: Tes
                 await asyncio.get_running_loop().run_in_executor(None, stop_event.wait)
             finally:
                 await mgr.stop()
+        with ThreadPoolExecutor(max_workers=1) as executor:
+            future = executor.submit(asyncio.run, run_manager())
+            start_event.wait()
 
-        manager_process = Process(target=lambda: asyncio.run(run_manager()))
-        manager_process.start()
-        start_event.wait()
+            yield sock_path
 
-        yield sock_path
-
-        stop_event.set()
-        manager_process.join()
+            stop_event.set()
+            future.result()
 
 
 @pytest.fixture(scope=testpy_test_fixture_scope)

test/cluster/tasks/task_manager_types.py

@@ -38,6 +38,7 @@ class TaskStats(NamedTuple):
     entity: str
     sequence_number: SequenceNum
     shard: int
+    creation_time: str
     start_time: str
     end_time: str
 
@@ -54,6 +55,7 @@ class TaskStatus(NamedTuple):
     entity: str
     sequence_number: SequenceNum
     is_abortable: bool
+    creation_time: str
     start_time: str
     end_time: str
     error: str

test/lib/cql_test_env.cc

@@ -881,7 +881,7 @@ private:
             _fd.start(
                 std::ref(_fd_pinger), std::ref(fd_clock),
                 service::direct_fd_clock::base::duration{std::chrono::milliseconds{100}}.count(),
-                service::direct_fd_clock::base::duration{std::chrono::milliseconds{600}}.count()).get();
+                service::direct_fd_clock::base::duration{std::chrono::milliseconds{600}}.count(), gcfg.gossip_scheduling_group).get();
 
             auto stop_fd = defer_verbose_shutdown("direct failure detector", [this] {
                 _fd.stop().get();

test/manual/test_scylla_assert.cc

@@ -1,49 +0,0 @@
-/*
- * Copyright (C) 2024-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-// Simple manual test for scylla_assert macro
-
-#include "utils/assert.hh"
-#include <iostream>
-#include <exception>
-
-void test_passing_assertion() {
-    scylla_assert(true);
-    scylla_assert(1 + 1 == 2);
-    scylla_assert(1 + 1 == 2, "basic math should work");
-    std::cout << "✓ All passing assertions succeeded\n";
-}
-
-void test_failing_assertion_without_message() {
-    try {
-        scylla_assert(false);
-        std::cout << "✗ Expected exception was not thrown\n";
-    } catch (const std::exception& e) {
-        std::cout << "✓ Caught expected exception: " << e.what() << "\n";
-    }
-}
-
-void test_failing_assertion_with_message() {
-    try {
-        scylla_assert(1 + 1 == 3, "this should fail");
-        std::cout << "✗ Expected exception was not thrown\n";
-    } catch (const std::exception& e) {
-        std::cout << "✓ Caught expected exception with message: " << e.what() << "\n";
-    }
-}
-
-int main() {
-    std::cout << "Testing scylla_assert macro...\n\n";
-    
-    test_passing_assertion();
-    test_failing_assertion_without_message();
-    test_failing_assertion_with_message();
-    
-    std::cout << "\n✓ All tests completed successfully\n";
-    return 0;
-}

test/perf/perf_row_cache_update.cc

@@ -30,7 +30,7 @@ static const int cell_size = 128;
 static bool cancelled = false;
 
 template<typename MutationGenerator>
-void run_test(const sstring& name, schema_ptr s, MutationGenerator&& gen) {
+void run_test(const sstring& name, schema_ptr s, MutationGenerator&& gen, std::function<mutation()> before_flush = {}) {
     tests::reader_concurrency_semaphore_wrapper semaphore;
     cache_tracker tracker;
     row_cache cache(s, make_empty_snapshot_source(), tracker, is_continuous::yes);
@@ -58,6 +58,10 @@ void run_test(const sstring& name, schema_ptr s, MutationGenerator&& gen) {
                     return;
                 }
             }
+            if (before_flush) {
+                mutation m = before_flush();
+                mt->apply(m);
+            }
         });
         memtable_slm.stop();
         std::cout << format("Memtable fill took {:.6f} [ms], {}", fill_d.count() * 1000, memtable_slm) << std::endl;
@@ -181,6 +185,43 @@ static void test_partition_with_lots_of_small_rows() {
     });
 }
 
+static void test_partition_with_lots_of_small_rows_covered_by_tombstone() {
+    auto s = schema_builder("ks", "cf")
+        .with_column("pk", uuid_type, column_kind::partition_key)
+        .with_column("ck", int32_type, column_kind::clustering_key)
+        .with_column("v1", bytes_type, column_kind::regular_column)
+        .with_column("v2", bytes_type, column_kind::regular_column)
+        .with_column("v3", bytes_type, column_kind::regular_column)
+        .build();
+
+    auto pk = dht::decorate_key(*s, partition_key::from_single_value(*s,
+        serialized(utils::UUID_gen::get_time_UUID())));
+    int ck_idx = 0;
+    int flush_ck_idx = 0;
+
+    run_test("Large partition, lots of small rows covered by single tombstone", s, [&] {
+        mutation m(s, pk);
+        auto val = data_value(bytes(bytes::initialized_later(), cell_size));
+        auto ck = clustering_key::from_single_value(*s, serialized(ck_idx++));
+        auto ts = api::new_timestamp();
+        m.set_clustered_cell(ck, "v1", val, ts);
+        m.set_clustered_cell(ck, "v2", val, ts);
+        m.set_clustered_cell(ck, "v3", val, ts);
+        return m;
+    }, [&] { // before_flush
+        // Delete key range [-inf, flush_ck_idx)
+        std::cout << "Generated " << (ck_idx - flush_ck_idx) << " rows\n";
+        auto m = mutation(s, pk);
+        auto ck = clustering_key::from_single_value(*s, serialized(flush_ck_idx));
+        m.partition().apply_row_tombstone(*s, range_tombstone(
+                position_in_partition_view::before_all_clustered_rows(),
+                position_in_partition_view::before_key(ck),
+                tombstone(api::new_timestamp(), gc_clock::now())));
+        flush_ck_idx = ck_idx;
+        return m;
+    });
+}
+
 static void test_partition_with_few_small_rows() {
     auto s = schema_builder("ks", "cf")
         .with_column("pk", uuid_type, column_kind::partition_key)
@@ -275,6 +316,7 @@ int scylla_row_cache_update_main(int argc, char** argv) {
                 cancelled = true;
             });
             logalloc::prime_segment_pool(memory::stats().total_memory(), memory::min_free_memory()).get();
+            test_partition_with_lots_of_small_rows_covered_by_tombstone();
             test_small_partitions();
             test_partition_with_few_small_rows();
             test_partition_with_lots_of_small_rows();

test/raft/failure_detector_test.cc

@@ -31,7 +31,7 @@ struct test_pinger: public direct_failure_detector::pinger {
     std::unordered_map<endpoint_id, size_t> _pings;
     bool _block = false;
 
-    virtual future<bool> ping(endpoint_id ep, abort_source& as) override {
+    virtual future<bool> ping(endpoint_id ep, direct_failure_detector::clock::timepoint_t timeout, abort_source& as, direct_failure_detector::clock& c) override {
         bool ret = false;
         co_await invoke_abortable_on(0, [this, ep, &ret] (abort_source& as) -> future<> {
             ++_pings[ep];
@@ -91,6 +91,9 @@ struct test_clock : public direct_failure_detector::clock {
             throw sleep_aborted{};
         }
     }
+    virtual std::chrono::milliseconds to_milliseconds(timepoint_t tp) const override {
+        throw std::logic_error("to_milliseconds is not implemented");
+    }
 };
 
 struct test_listener : public direct_failure_detector::listener {
@@ -129,7 +132,7 @@ SEASTAR_TEST_CASE(failure_detector_test) {
     test_pinger pinger;
     test_clock clock;
     sharded<direct_failure_detector::failure_detector> fd;
-    co_await fd.start(std::ref(pinger), std::ref(clock), 10, 30);
+    co_await fd.start(std::ref(pinger), std::ref(clock), 10, 30, seastar::current_scheduling_group());
 
     test_listener l1, l2;
     auto sub1 = co_await fd.local().register_listener(l1, 95);

test/raft/randomized_nemesis_test.cc

@@ -1065,7 +1065,7 @@ public:
     }
 
     // Can be called on any shard.
-    future<bool> ping(direct_failure_detector::pinger::endpoint_id id, abort_source& as) override {
+    future<bool> ping(direct_failure_detector::pinger::endpoint_id id, direct_failure_detector::clock::timepoint_t timeout, abort_source& as, direct_failure_detector::clock& c) override {
         try {
             co_await invoke_abortable_on(0, [this, id] (abort_source& as) {
                 return _rpc.ping(raft::server_id{id}, as);
@@ -1127,6 +1127,10 @@ public:
             throw sleep_aborted{};
         }
     }
+
+    virtual std::chrono::milliseconds to_milliseconds(timepoint_t tp) const override {
+        throw std::logic_error("to_milliseconds is not implemented");
+    }
 };
 
 class direct_fd_listener : public raft::failure_detector, public direct_failure_detector::listener {
@@ -1436,7 +1440,7 @@ public:
         // _fd_service must be started before raft server,
         // because as soon as raft server is started, it may start adding endpoints to the service.
         // _fd_service is using _server's RPC, but not until the first endpoint is added.
-        co_await _fd_service->start(std::ref(*_fd_pinger), std::ref(*_fd_clock), fd_ping_period.count(), fd_ping_timeout.count());
+        co_await _fd_service->start(std::ref(*_fd_pinger), std::ref(*_fd_clock), fd_ping_period.count(), fd_ping_timeout.count(), seastar::current_scheduling_group());
         _fd_subscription.emplace(co_await _fd_service->local().register_listener(*_fd_listener, _fd_convict_threshold.count()));
         co_await _server->start();
     }

tools/scylla-nodetool.cc

@@ -3174,7 +3174,7 @@ void tasks_print_status(const rjson::value& res) {
     auto status = res.GetObject();
     for (const auto& x: status) {
         if (x.value.IsString()) {
-            if (strcmp(x.name.GetString(), "start_time") == 0 || strcmp(x.name.GetString(), "end_time") == 0) {
+            if (strcmp(x.name.GetString(), "creation_time") == 0 || strcmp(x.name.GetString(), "start_time") == 0 || strcmp(x.name.GetString(), "end_time") == 0) {
                 fmt::print("{}: {}\n", x.name.GetString(), get_time(x.value.GetString()));
             } else {
                 fmt::print("{}: {}\n", x.name.GetString(), x.value.GetString());
@@ -3226,6 +3226,7 @@ void tasks_add_tree_to_statuses_lists(Tabulate& table, const rjson::value& res)
                 rjson::to_string_view(status["scope"]),
                 rjson::to_string_view(status["state"]),
                 status["is_abortable"].GetBool(),
+                get_time(rjson::to_string_view(status["creation_time"])),
                 get_time(rjson::to_string_view(status["start_time"])),
                 get_time(rjson::to_string_view(status["end_time"])),
                 rjson::to_string_view(status["error"]),
@@ -3245,7 +3246,7 @@ void tasks_add_tree_to_statuses_lists(Tabulate& table, const rjson::value& res)
 void tasks_print_trees(const std::vector<rjson::value>& res) {
     Tabulate table;
     table.add("id", "type", "kind", "scope", "state",
-        "is_abortable", "start_time", "end_time", "error", "parent_id",
+        "is_abortable", "creation_time", "start_time", "end_time", "error", "parent_id",
         "sequence_number", "shard", "keyspace", "table", "entity",
         "progress_units", "total", "completed", "children_ids");
 
@@ -3259,7 +3260,7 @@ void tasks_print_trees(const std::vector<rjson::value>& res) {
 void tasks_print_stats_list(const rjson::value& res) {
     auto stats = res.GetArray();
     Tabulate table;
-    table.add("task_id", "type", "kind", "scope", "state", "sequence_number", "keyspace", "table", "entity", "shard", "start_time", "end_time");
+    table.add("task_id", "type", "kind", "scope", "state", "sequence_number", "keyspace", "table", "entity", "shard", "creation_time", "start_time", "end_time");
     for (auto& element : stats) {
         const auto& s = element.GetObject();
 
@@ -3273,6 +3274,7 @@ void tasks_print_stats_list(const rjson::value& res) {
                 rjson::to_string_view(s["table"]),
                 rjson::to_string_view(s["entity"]),
                 s["shard"].GetUint(),
+                get_time(rjson::to_string_view(s["creation_time"])),
                 get_time(rjson::to_string_view(s["start_time"])),
                 get_time(rjson::to_string_view(s["end_time"])));
     }

utils/abstract_formatter.hh

@@ -0,0 +1,41 @@
+/*
+ * Copyright (C) 2025-present ScyllaDB
+ */
+
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#pragma once
+
+#include <fmt/format.h>
+#include <functional>
+
+/// Type-erased formatter.
+/// Allows passing formattable objects without exposing their types.
+class abstract_formatter {
+    std::function<void(fmt::format_context&)> _formatter;
+public:
+    abstract_formatter() = default;
+
+    template<typename Func>
+    requires std::is_invocable_v<Func, fmt::format_context&>
+    explicit abstract_formatter(Func&& f) : _formatter(std::forward<Func>(f)) {}
+
+    fmt::format_context::iterator format_to(fmt::format_context& ctx) const {
+        if (_formatter) {
+            _formatter(ctx);
+        }
+        return ctx.out();
+    }
+
+    explicit operator bool() const noexcept { return bool(_formatter); }
+};
+
+template <> struct fmt::formatter<abstract_formatter> {
+    constexpr auto parse(format_parse_context& ctx) { return ctx.begin(); }
+
+    auto format(const abstract_formatter& formatter, fmt::format_context& ctx) const {
+        return formatter.format_to(ctx);
+    }
+};

utils/assert.hh

@@ -4,27 +4,6 @@
 #pragma once
 
 #include <cassert>
-#include <seastar/core/format.hh>
-#include "utils/on_internal_error.hh"
 
 /// Like assert(), but independent of NDEBUG. Active in all build modes.
 #define SCYLLA_ASSERT(x) do { if (!(x)) { __assert_fail(#x, __FILE__, __LINE__, __PRETTY_FUNCTION__); } } while (0)
-
-/// Exception-throwing assertion based on on_internal_error()
-///
-/// Unlike SCYLLA_ASSERT which crashes the process, scylla_assert() throws an
-/// exception (or aborts depending on configuration). This prevents cluster-wide
-/// crashes and loss of availability by allowing graceful error handling.
-///
-/// Use this instead of SCYLLA_ASSERT in contexts where throwing exceptions is safe.
-/// DO NOT use in:
-/// - noexcept functions
-/// - destructors
-/// - contexts with special exception-safety requirements
-#define scylla_assert(condition, ...) \
-    do { \
-        if (!(condition)) [[unlikely]] { \
-            ::utils::on_internal_error(::seastar::format("Assertion failed: {} at {}:{} in {}" __VA_OPT__(": {}"), \
-                #condition, __FILE__, __LINE__, __PRETTY_FUNCTION__ __VA_OPT__(,) __VA_ARGS__)); \
-        } \
-    } while (0)

utils/cached_file.hh

@@ -461,6 +461,9 @@ public:
         , _metrics(m)
         , _lru(l)
         , _region(reg)
+        , _as(abstract_formatter([this] (fmt::format_context& ctx) {
+            fmt::format_to(ctx.out(), "cached_file {}", _file_name);
+        }))
         , _cache(page_idx_less_comparator())
         , _size(size)
     {

utils/logalloc.cc

@@ -2948,10 +2948,10 @@ void allocating_section::on_alloc_failure(logalloc::region& r) {
     r.allocator().invalidate_references();
     if (r.get_tracker().get_impl().segment_pool().allocation_failure_flag()) {
         _lsa_reserve *= 2;
-        llogger.info("LSA allocation failure, increasing reserve in section {} to {} segments; trace: {}", fmt::ptr(this), _lsa_reserve, current_backtrace());
+        llogger.info("LSA allocation failure, increasing reserve in section {} ({}) to {} segments; trace: {}", fmt::ptr(this), _name, _lsa_reserve, current_backtrace());
     } else {
         _std_reserve *= 2;
-        llogger.info("Standard allocator failure, increasing head-room in section {} to {} [B]; trace: {}", fmt::ptr(this), _std_reserve, current_backtrace());
+        llogger.info("Standard allocator failure, increasing head-room in section {} ({}) to {} [B]; trace: {}", fmt::ptr(this), _name, _std_reserve, current_backtrace());
     }
     reserve(r.get_tracker().get_impl());
 }

utils/logalloc.hh

@@ -17,6 +17,7 @@
 #include "utils/assert.hh"
 #include "utils/entangled.hh"
 #include "utils/memory_limit_reached.hh"
+#include "utils/abstract_formatter.hh"
 
 namespace logalloc {
 
@@ -442,6 +443,7 @@ class allocating_section {
     size_t _minimum_lsa_emergency_reserve = 0;
     int64_t _remaining_std_bytes_until_decay = s_bytes_per_decay;
     int _remaining_lsa_segments_until_decay = s_segments_per_decay;
+    abstract_formatter _name;
 private:
     struct guard {
         tracker::impl& _tracker;
@@ -453,6 +455,8 @@ private:
     void maybe_decay_reserve() noexcept;
     void on_alloc_failure(logalloc::region&);
 public:
+    allocating_section() = default;
+    explicit allocating_section(abstract_formatter name) : _name(std::move(name)) {}
 
     void set_lsa_reserve(size_t) noexcept;
     void set_std_reserve(size_t) noexcept;