test: test_remove_garbage_group0_members: wait for token ring and group0 consistency before removenode

The removenove initiator could have an outdated token ring (still considering
the node removed by the previous removenode a token owner) and unexpectedly
reject the operation.

Fix that by waiting for token ring and group0 consistency before removenode.
Note that the test already checks that consistency, but only for one node,
which is different from the removenode initiator.

This test has been removed in master together with the code being tested
(the gossip-based topology). Hence, the fix is submitted directly to 2026.1.

Fixes https://scylladb.atlassian.net/browse/SCYLLADB-1103

Backport to all supported branches (other than 2026.1), as the test can fail
there.

Closes scylladb/scylladb#29108

(cherry picked from commit 1398a55d16)

Closes scylladb/scylladb#29205

.github/CODEOWNERS

@@ -1,5 +1,5 @@
 # AUTH
-auth/* @nuivall
+auth/* @nuivall @ptrsmrn
 
 # CACHE
 row_cache* @tgrabiec
@@ -25,11 +25,11 @@ compaction/* @raphaelsc
 transport/*
 
 # CQL QUERY LANGUAGE
-cql3/* @tgrabiec @nuivall
+cql3/* @tgrabiec @nuivall @ptrsmrn
 
 # COUNTERS
-counters* @nuivall
-tests/counter_test* @nuivall
+counters* @nuivall @ptrsmrn
+tests/counter_test* @nuivall @ptrsmrn
 
 # DOCS
 docs/* @annastuchlik @tzach
@@ -57,6 +57,7 @@ repair/* @tgrabiec @asias
 
 # SCHEMA MANAGEMENT
 db/schema_tables* @tgrabiec
+db/legacy_schema_migrator* @tgrabiec
 service/migration* @tgrabiec
 schema* @tgrabiec
 

.github/copilot-instructions.md

@@ -1,97 +0,0 @@
-# ScyllaDB Development Instructions
-
-## Project Context
-High-performance distributed NoSQL database. Core values: performance, correctness, readability.
-
-## Build System
-
-### Modern Build (configure.py + ninja)
-```bash
-# Configure (run once per mode, or when switching modes)
-./configure.py --mode=<mode>  # mode: dev, debug, release, sanitize
-
-# Build everything
-ninja <mode>-build  # e.g., ninja dev-build
-
-# Build Scylla binary only (sufficient for Python integration tests)
-ninja build/<mode>/scylla
-
-# Build specific test
-ninja build/<mode>/test/boost/<test_name>
-```
-
-## Running Tests
-
-### C++ Unit Tests
-```bash
-# Run all tests in a file
-./test.py --mode=<mode> test/<suite>/<test_name>.cc
-
-# Run a single test case from a file
-./test.py --mode=<mode> test/<suite>/<test_name>.cc::<test_case_name>
-
-# Examples
-./test.py --mode=dev test/boost/memtable_test.cc
-./test.py --mode=dev test/raft/raft_server_test.cc::test_check_abort_on_client_api
-```
-
-**Important:** 
-- Use full path with `.cc` extension (e.g., `test/boost/test_name.cc`, not `boost/test_name`)
-- To run a single test case, append `::<test_case_name>` to the file path
-- If you encounter permission issues with cgroup metric gathering, add `--no-gather-metrics` flag
-
-**Rebuilding Tests:**
-- test.py does NOT automatically rebuild when test source files are modified
-- Many tests are part of composite binaries (e.g., `combined_tests` in test/boost contains multiple test files)
-- To find which binary contains a test, check `configure.py` in the repository root (primary source) or `test/<suite>/CMakeLists.txt`
-- To rebuild a specific test binary: `ninja build/<mode>/test/<suite>/<binary_name>`
-- Examples: 
-  - `ninja build/dev/test/boost/combined_tests` (contains group0_voter_calculator_test.cc and others)
-  - `ninja build/dev/test/raft/replication_test` (standalone Raft test)
-
-### Python Integration Tests
-```bash
-# Only requires Scylla binary (full build usually not needed)
-ninja build/<mode>/scylla
-
-# Run all tests in a file
-./test.py --mode=<mode> <test_path>
-
-# Run a single test case from a file
-./test.py --mode=<mode> <test_path>::<test_function_name>
-
-# Examples
-./test.py --mode=dev alternator/
-./test.py --mode=dev cluster/test_raft_voters::test_raft_limited_voters_retain_coordinator
-
-# Optional flags
-./test.py --mode=dev cluster/test_raft_no_quorum -v  # Verbose output
-./test.py --mode=dev cluster/test_raft_no_quorum --repeat 5  # Repeat test 5 times
-```
-
-**Important:**
-- Use path without `.py` extension (e.g., `cluster/test_raft_no_quorum`, not `cluster/test_raft_no_quorum.py`)
-- To run a single test case, append `::<test_function_name>` to the file path
-- Add `-v` for verbose output
-- Add `--repeat <num>` to repeat a test multiple times
-- After modifying C++ source files, only rebuild the Scylla binary for Python tests - building the entire repository is unnecessary
-
-## Code Philosophy
-- Performance matters in hot paths (data read/write, inner loops)
-- Self-documenting code through clear naming
-- Comments explain "why", not "what"
-- Prefer standard library over custom implementations
-- Strive for simplicity and clarity, add complexity only when clearly justified
-- Question requests: don't blindly implement requests - evaluate trade-offs, identify issues, and suggest better alternatives when appropriate
-- Consider different approaches, weigh pros and cons, and recommend the best fit for the specific context
-
-## Test Philosophy
-- Performance matters. Tests should run as quickly as possible. Sleeps in the code are highly discouraged and should be avoided, to reduce run time and flakiness.
-- Stability matters. Tests should be stable. New tests should be executed 100 times at least to ensure they pass 100 out of 100 times. (use --repeat 100 --max-failures 1 when running it)
-- Unit tests should ideally test one thing and one thing only.
-- Tests for bug fixes should run before the fix - and show the failure and after the fix - and show they now pass.
-- Tests for bug fixes should have in their comments which bug fixes (GitHub or JIRA issue) they test.
-- Tests in debug are always slower, so if needed, reduce number of iterations, rows, data used, cycles, etc. in debug mode.
-- Tests should strive to be repeatable, and not use random input that will make their results unpredictable.
-- Tests should consume as little resources as possible. Prefer running tests on a single node if it is sufficient, for example.
-

.github/instructions/cpp.instructions.md

@@ -1,115 +0,0 @@
----
-applyTo: "**/*.{cc,hh}"
----
-
-# C++ Guidelines
-
-**Important:** Always match the style and conventions of existing code in the file and directory.
-
-## Memory Management
-- Prefer stack allocation whenever possible
-- Use `std::unique_ptr` by default for dynamic allocations
-- `new`/`delete` are forbidden (use RAII)
-- Use `seastar::lw_shared_ptr` or `seastar::shared_ptr` for shared ownership within same shard
-- Use `seastar::foreign_ptr` for cross-shard sharing
-- Avoid `std::shared_ptr` except when interfacing with external C++ APIs
-- Avoid raw pointers except for non-owning references or C API interop
-
-## Seastar Asynchronous Programming
-- Use `seastar::future<T>` for all async operations
-- Prefer coroutines (`co_await`, `co_return`) over `.then()` chains for readability
-- Coroutines are preferred over `seastar::do_with()` for managing temporary state
-- In hot paths where futures are ready, continuations may be more efficient than coroutines
-- Chain futures with `.then()`, don't block with `.get()` (unless in `seastar::thread` context)
-- All I/O must be asynchronous (no blocking calls)
-- Use `seastar::gate` for shutdown coordination
-- Use `seastar::semaphore` for resource limiting (not `std::mutex`)
-- Break long loops with `maybe_yield()` to avoid reactor stalls
-
-## Coroutines
-```cpp
-seastar::future<T> func() {
-    auto result = co_await async_operation();
-    co_return result;
-}
-```
-
-## Error Handling
-- Throw exceptions for errors (futures propagate them automatically)
-- In data path: avoid exceptions, use `std::expected` (or `boost::outcome`) instead
-- Use standard exceptions (`std::runtime_error`, `std::invalid_argument`)
-- Database-specific: throw appropriate schema/query exceptions
-
-## Performance
-- Pass large objects by `const&` or `&&` (move semantics)
-- Use `std::string_view` for non-owning string references
-- Avoid copies: prefer move semantics
-- Use `utils::chunked_vector` instead of `std::vector` for large allocations (>128KB)
-- Minimize dynamic allocations in hot paths
-
-## Database-Specific Types
-- Use `schema_ptr` for schema references
-- Use `mutation` and `mutation_partition` for data modifications
-- Use `partition_key` and `clustering_key` for keys
-- Use `api::timestamp_type` for database timestamps
-- Use `gc_clock` for garbage collection timing
-
-## Style
-- C++23 standard (prefer modern features, especially coroutines)
-- Use `auto` when type is obvious from RHS
-- Avoid `auto` when it obscures the type
-- Use range-based for loops: `for (const auto& item : container)`
-- Use standard algorithms when they clearly simplify code (e.g., replacing 10-line loops)
-- Avoid chaining multiple algorithms if a straightforward loop is clearer
-- Mark functions and variables `const` whenever possible
-- Use scoped enums: `enum class` (not unscoped `enum`)
-
-## Headers
-- Use `#pragma once`
-- Include order: own header, C++ std, Seastar, Boost, project headers
-- Forward declare when possible
-- Never `using namespace` in headers (exception: `using namespace seastar` is globally available via `seastarx.hh`)
-
-## Documentation
-- Public APIs require clear documentation
-- Implementation details should be self-evident from code
-- Use `///` or Doxygen `/** */` for public documentation, `//` for implementation notes - follow the existing style
-
-## Naming
-- `snake_case` for most identifiers (classes, functions, variables, namespaces)
-- Template parameters: `CamelCase` (e.g., `template<typename ValueType>`)
-- Member variables: prefix with `_` (e.g., `int _count;`)
-- Structs (value-only): no `_` prefix on members
-- Constants and `constexpr`: `snake_case` (e.g., `static constexpr int max_size = 100;`)
-- Files: `.hh` for headers, `.cc` for source
-
-## Formatting
-- 4 spaces indentation, never tabs
-- Opening braces on same line as control structure (except namespaces)
-- Space after keywords: `if (`, `while (`, `return `
-- Whitespace around operators matches precedence: `*a + *b` not `* a+* b`
-- Line length: keep reasonable (<160 chars), use continuation lines with double indent if needed
-- Brace all nested scopes, even single statements
-- Minimal patches: only format code you modify, never reformat entire files
-
-## Logging
-- Use structured logging with appropriate levels: DEBUG, INFO, WARN, ERROR
-- Include context in log messages (e.g., request IDs)
-- Never log sensitive data (credentials, PII)
-
-## Forbidden
-- `malloc`/`free`
-- `printf` family (use logging or fmt)
-- Raw pointers for ownership
-- `using namespace` in headers
-- Blocking operations: `std::sleep`, `std::read`, `std::mutex` (use Seastar equivalents)
-- `std::atomic` (reserved for very special circumstances only)
-- Macros (use `inline`, `constexpr`, or templates instead)
-
-## Testing
-When modifying existing code, follow TDD: create/update test first, then implement.
-- Examine existing tests for style and structure
-- Use Boost.Test framework
-- Use `SEASTAR_THREAD_TEST_CASE` for Seastar asynchronous tests
-- Aim for high code coverage, especially for new features and bug fixes
-- Maintain bisectability: all tests must pass in every commit. Mark failing tests with `BOOST_FAIL()` or similar, then fix in subsequent commit

.github/instructions/python.instructions.md

@@ -1,51 +0,0 @@
----
-applyTo: "**/*.py"
----
-
-# Python Guidelines
-
-**Important:** Match existing code style. Some directories (like `test/cqlpy` and `test/alternator`) prefer simplicity over type hints and docstrings.
-
-## Style
-- Follow PEP 8
-- Use type hints for function signatures (unless directory style omits them)
-- Use f-strings for formatting
-- Line length: 160 characters max
-- 4 spaces for indentation
-
-## Imports
-Order: standard library, third-party, local imports
-```python
-import os
-import sys
-
-import pytest
-from cassandra.cluster import Cluster
-
-from test.utils import setup_keyspace
-```
-
-Never use `from module import *`
-
-## Documentation
-All public functions/classes need docstrings (unless the current directory conventions omit them):
-```python
-def my_function(arg1: str, arg2: int) -> bool:
-    """
-    Brief summary of function purpose.
-
-    Args:
-        arg1: Description of first argument.
-        arg2: Description of second argument.
-
-    Returns:
-        Description of return value.
-    """
-    pass
-```
-
-## Testing Best Practices
-- Maintain bisectability: all tests must pass in every commit
-- Mark currently-failing tests with `@pytest.mark.xfail`, unmark when fixed
-- Use descriptive names that convey intent
-- Docstrings/comments should explain what the test verifies and why, and if it reproduces a specific issue or how it fits into the larger test suite

.github/workflows/backport-pr-fixes-validation.yaml

@@ -18,7 +18,7 @@ jobs:
             
             // Regular expression pattern to check for "Fixes" prefix
             // Adjusted to dynamically insert the repository full name
-            const pattern = `Fixes:? ((?:#|${repo.replace('/', '\\/')}#|https://github\\.com/${repo.replace('/', '\\/')}/issues/)(\\d+)|([A-Z]+-\\d+))`;
+            const pattern = `Fixes:? ((?:#|${repo.replace('/', '\\/')}#|https://github\\.com/${repo.replace('/', '\\/')}/issues/)(\\d+)|(?:https://scylladb\\.atlassian\\.net/browse/)?([A-Z]+-\\d+))`;
             const regex = new RegExp(pattern);
             
             if (!regex.test(body)) {

.github/workflows/call_backport_with_jira.yaml

@@ -0,0 +1,53 @@
+name: Backport with Jira Integration
+
+on:
+  push:
+    branches:
+      - master
+      - next-*.*
+      - branch-*.*
+  pull_request_target:
+    types: [labeled, closed]
+    branches: 
+      - master
+      - next
+      - next-*.*
+      - branch-*.*
+
+jobs:
+  backport-on-push:
+    if: github.event_name == 'push'
+    uses: scylladb/github-automation/.github/workflows/backport-with-jira.yaml@main
+    with:
+      event_type: 'push'
+      base_branch: ${{ github.ref }}
+      commits: ${{ github.event.before }}..${{ github.sha }}
+    secrets:
+      gh_token: ${{ secrets.AUTO_BACKPORT_TOKEN }}
+      jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
+
+  backport-on-label:
+    if: github.event_name == 'pull_request_target' && github.event.action == 'labeled'
+    uses: scylladb/github-automation/.github/workflows/backport-with-jira.yaml@main
+    with:
+      event_type: 'labeled'
+      base_branch: refs/heads/${{ github.event.pull_request.base.ref }}
+      pull_request_number: ${{ github.event.pull_request.number }}
+      head_commit: ${{ github.event.pull_request.base.sha }}
+      label_name: ${{ github.event.label.name }}
+      pr_state: ${{ github.event.pull_request.state }}
+    secrets:
+      gh_token: ${{ secrets.AUTO_BACKPORT_TOKEN }}
+      jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
+
+  backport-chain:
+    if: github.event_name == 'pull_request_target' && github.event.action == 'closed' && github.event.pull_request.merged == true
+    uses: scylladb/github-automation/.github/workflows/backport-with-jira.yaml@main
+    with:
+      event_type: 'chain'
+      base_branch: refs/heads/${{ github.event.pull_request.base.ref }}
+      pull_request_number: ${{ github.event.pull_request.number }}
+      pr_body: ${{ github.event.pull_request.body }}
+    secrets:
+      gh_token: ${{ secrets.AUTO_BACKPORT_TOKEN }}
+      jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}

.github/workflows/call_jira_status_in_progress.yml

@@ -0,0 +1,12 @@
+name: Call Jira Status In Progress
+
+on:
+  pull_request_target:
+    types: [opened]
+
+jobs:
+  call-jira-status-in-progress:
+    uses: scylladb/github-automation/.github/workflows/main_update_jira_status_to_in_progress.yml@main
+    secrets:
+      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
+

.github/workflows/call_jira_status_in_review.yml

@@ -0,0 +1,12 @@
+name: Call Jira Status In Review
+
+on:
+  pull_request_target:
+    types: [ready_for_review, review_requested]
+
+jobs:
+  call-jira-status-in-review:
+    uses: scylladb/github-automation/.github/workflows/main_update_jira_status_to_in_review.yml@main
+    secrets:
+      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
+

.github/workflows/call_jira_status_ready_for_merge.yml

@@ -0,0 +1,12 @@
+name: Call Jira Status Ready For Merge
+
+on:
+  pull_request_target:
+    types: [labeled]
+
+jobs:
+  call-jira-status-update:
+    uses: scylladb/github-automation/.github/workflows/main_update_jira_status_to_ready_for_merge.yml@main
+    secrets:
+      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
+

.github/workflows/call_jira_sync.yml

@@ -1,41 +0,0 @@
-name: Sync Jira Based on PR Events
-
-on:
-  pull_request_target:
-    types: [opened, ready_for_review, review_requested, labeled, unlabeled, closed]
-
-permissions:
-  contents: read
-  pull-requests: write
-  issues: write
-
-jobs:
-  jira-sync-pr-opened:
-    if: github.event.action == 'opened'
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_pr_opened.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
-
-  jira-sync-in-review:
-    if: github.event.action == 'ready_for_review' || github.event.action == 'review_requested'
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_in_review.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
-
-  jira-sync-add-label:
-    if: github.event.action == 'labeled'
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_add_label.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
-
-  jira-status-remove-label:
-    if: github.event.action == 'unlabeled'
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_remove_label.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}
-
-  jira-status-pr-closed:
-    if: github.event.action == 'closed' 
-    uses: scylladb/github-automation/.github/workflows/main_jira_sync_pr_closed.yml@main
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}

.github/workflows/call_sync_milestone_to_jira.yml

@@ -1,14 +0,0 @@
-name: Call Jira release creation for new milestone
-
-on:
-  milestone:
-    types: [created]
-
-jobs:
-  sync-milestone-to-jira:
-    uses: scylladb/github-automation/.github/workflows/main_sync_milestone_to_jira_release.yml@main
-    with:
-      # Comma-separated list of Jira project keys
-      jira_project_keys: "SCYLLADB,CUSTOMER"
-    secrets:
-      caller_jira_auth: ${{ secrets.USER_AND_KEY_FOR_JIRA_AUTOMATION }}

.github/workflows/call_validate_pr_author_email.yml

@@ -1,13 +0,0 @@
-name: validate_pr_author_email
-
-on:
-  pull_request_target:
-    types:
-      - opened
-      - synchronize
-      - reopened
-
-jobs:
-  validate_pr_author_email:
-    uses: scylladb/github-automation/.github/workflows/validate_pr_author_email.yml@main
-

.github/workflows/codespell.yaml

@@ -13,5 +13,5 @@ jobs:
       - uses: codespell-project/actions-codespell@master
         with:
           only_warn: 1
-          ignore_words_list: "ans,datas,fo,ser,ue,crate,nd,reenable,strat,stap,te,raison,iif,tread"
+          ignore_words_list: "ans,datas,fo,ser,ue,crate,nd,reenable,strat,stap,te,raison"
           skip: "./.git,./build,./tools,*.js,*.lock,./test,./licenses,./redis/lolwut.cc,*.svg"

.github/workflows/docs-pages.yaml

@@ -18,8 +18,6 @@ on:
 
 jobs:
   release:
-    permissions:
-      contents: write
     runs-on: ubuntu-latest
     steps:
       - name: Checkout

.github/workflows/docs-pr.yaml

@@ -2,9 +2,6 @@ name: "Docs / Build PR"
 # For more information,
 # see https://sphinx-theme.scylladb.com/stable/deployment/production.html#available-workflows
 
-permissions:
-  contents: read
-
 env:
   FLAG: ${{ github.repository == 'scylladb/scylla-enterprise' && 'enterprise' || 'opensource' }}
 

.github/workflows/docs-validate-metrics.yml

@@ -1,37 +0,0 @@
-name: Docs / Validate metrics
-
-permissions:
-  contents: read
-
-on:
-  pull_request:
-    branches:
-      - master
-      - enterprise
-    paths:
-      - '**/*.cc'
-      - 'scripts/metrics-config.yml'
-      - 'scripts/get_description.py'
-      - 'docs/_ext/scylladb_metrics.py'
-
-jobs:
-  validate-metrics:
-    runs-on: ubuntu-latest
-    name: Check metrics documentation coverage
-
-    steps:
-    - name: Checkout code
-      uses: actions/checkout@v4
-      with:
-        submodules: true
-
-    - name: Set up Python
-      uses: actions/setup-python@v6
-      with:
-        python-version: '3.10'
-
-    - name: Install dependencies
-      run: pip install PyYAML
-
-    - name: Validate metrics
-      run: python3 scripts/get_description.py --validate -c scripts/metrics-config.yml

.github/workflows/read-toolchain.yaml

@@ -10,8 +10,6 @@ on:
 jobs:
   read-toolchain:
     runs-on: ubuntu-latest
-    permissions:
-      contents: read
     outputs:
       image: ${{ steps.read.outputs.image }}
     steps:

.github/workflows/trigger-scylla-ci.yaml

@@ -9,16 +9,57 @@ on:
 
 jobs:
   trigger-jenkins:
-    if: (github.event.comment.user.login != 'scylladbbot' && contains(github.event.comment.body, '@scylladbbot') && contains(github.event.comment.body, 'trigger-ci')) || github.event.label.name == 'conflicts'
+    if: (github.event_name == 'issue_comment' && github.event.comment.user.login != 'scylladbbot') || github.event.label.name == 'conflicts'
     runs-on: ubuntu-latest
     steps:
+      - name: Verify Org Membership
+        id: verify_author
+        env:
+          EVENT_NAME: ${{ github.event_name }}
+          PR_AUTHOR: ${{ github.event.pull_request.user.login }}
+          PR_ASSOCIATION: ${{ github.event.pull_request.author_association }}
+          COMMENT_AUTHOR: ${{ github.event.comment.user.login }}
+          COMMENT_ASSOCIATION: ${{ github.event.comment.author_association }}
+        shell: bash
+        run: |
+          if [[ "$EVENT_NAME" == "pull_request_target" ]]; then
+            AUTHOR="$PR_AUTHOR"
+            ASSOCIATION="$PR_ASSOCIATION"
+          else
+            AUTHOR="$COMMENT_AUTHOR"
+            ASSOCIATION="$COMMENT_ASSOCIATION"
+          fi
+          ORG="scylladb"
+          if gh api "/orgs/${ORG}/members/${AUTHOR}" --silent 2>/dev/null; then
+            echo "member=true" >> $GITHUB_OUTPUT
+          else
+            echo "::warning::${AUTHOR} is not a member of ${ORG}; skipping CI trigger."
+            echo "member=false" >> $GITHUB_OUTPUT
+          fi
+
+      - name: Validate Comment Trigger
+        if: github.event_name == 'issue_comment'
+        id: verify_comment
+        env:
+          COMMENT_BODY: ${{ github.event.comment.body }}
+        shell: bash
+        run: |
+          CLEAN_BODY=$(echo "$COMMENT_BODY" | grep -v '^[[:space:]]*>')
+
+          if echo "$CLEAN_BODY" | grep -qi '@scylladbbot' && echo "$CLEAN_BODY" | grep -qi 'trigger-ci'; then
+            echo "trigger=true" >> $GITHUB_OUTPUT
+          else
+            echo "trigger=false" >> $GITHUB_OUTPUT
+          fi
+
       - name: Trigger Scylla-CI-Route Jenkins Job
+        if: steps.verify_author.outputs.member == 'true' && (github.event_name == 'pull_request_target' || steps.verify_comment.outputs.trigger == 'true')
         env:
           JENKINS_USER: ${{ secrets.JENKINS_USERNAME }}
           JENKINS_API_TOKEN: ${{ secrets.JENKINS_TOKEN }}
           JENKINS_URL: "https://jenkins.scylladb.com"
+          PR_NUMBER: "${{ github.event.issue.number || github.event.pull_request.number }}"
+          PR_REPO_NAME: "${{ github.event.repository.full_name }}"
         run: |
-          PR_NUMBER=${{ github.event.issue.number }}
-          PR_REPO_NAME=${{ github.event.repository.full_name }}
           curl -X POST "$JENKINS_URL/job/releng/job/Scylla-CI-Route/buildWithParameters?PR_NUMBER=$PR_NUMBER&PR_REPO_NAME=$PR_REPO_NAME" \
-          --user "$JENKINS_USER:$JENKINS_API_TOKEN" --fail -i -v
+            --user "$JENKINS_USER:$JENKINS_API_TOKEN" --fail

.github/workflows/trigger_ci.yaml

@@ -1,242 +0,0 @@
-name: Trigger next gating
-
-on:
-  pull_request_target:
-    types: [opened, reopened, synchronize]
-  issue_comment:
-    types: [created]
-    
-jobs:
-  trigger-ci:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Dump GitHub context
-        env:
-          GITHUB_CONTEXT: ${{ toJson(github) }}
-        run: echo "$GITHUB_CONTEXT"
-      - name: Checkout PR code
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: 0  # Needed to access full history
-          ref: ${{ github.event.pull_request.head.ref }}
-
-      - name: Fetch before commit if needed
-        run: |
-          if ! git cat-file -e ${{ github.event.before }} 2>/dev/null; then
-            echo "Fetching before commit ${{ github.event.before }}"
-            git fetch --depth=1 origin ${{ github.event.before }}
-          fi
-
-      - name: Compare commits for file changes
-        if: github.action == 'synchronize'
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        run: |
-          echo "Base: ${{ github.event.before }}"
-          echo "Head: ${{ github.event.after }}"
-
-          TREE_BEFORE=$(git show -s --format=%T ${{ github.event.before }})
-          TREE_AFTER=$(git show -s --format=%T ${{ github.event.after }})
-          
-          echo "TREE_BEFORE=$TREE_BEFORE" >> $GITHUB_ENV
-          echo "TREE_AFTER=$TREE_AFTER" >> $GITHUB_ENV
-
-      - name: Check if last push has file changes
-        run: |
-          if [[ "${{ env.TREE_BEFORE }}" == "${{ env.TREE_AFTER }}" ]]; then
-            echo "No file changes detected in the last push, only commit message edit."
-            echo "has_file_changes=false" >> $GITHUB_ENV
-          else
-            echo "File changes detected in the last push."
-            echo "has_file_changes=true" >> $GITHUB_ENV
-          fi
-
-      - name: Rule 1 - Check PR draft or conflict status
-        run: |
-          # Check if PR is in draft mode
-          IS_DRAFT="${{ github.event.pull_request.draft }}"
-          
-          # Check if PR has 'conflict' label
-          HAS_CONFLICT_LABEL="false"
-          LABELS='${{ toJson(github.event.pull_request.labels) }}'
-          if echo "$LABELS" | jq -r '.[].name' | grep -q "^conflict$"; then
-            HAS_CONFLICT_LABEL="true"
-          fi
-          
-          # Set draft_or_conflict variable
-          if [[ "$IS_DRAFT" == "true" || "$HAS_CONFLICT_LABEL" == "true" ]]; then
-            echo "draft_or_conflict=true" >> $GITHUB_ENV
-            echo "✅ Rule 1: PR is in draft mode or has conflict label - setting draft_or_conflict=true"
-          else
-            echo "draft_or_conflict=false" >> $GITHUB_ENV
-            echo "✅ Rule 1: PR is ready and has no conflict label - setting draft_or_conflict=false"
-          fi
-          
-          echo "Draft status: $IS_DRAFT"
-          echo "Has conflict label: $HAS_CONFLICT_LABEL"
-          echo "Result: draft_or_conflict = $draft_or_conflict"
-
-      - name: Rule 2 - Check labels
-        run: |
-          # Check if PR has P0 or P1 labels
-          HAS_P0_P1_LABEL="false"
-          LABELS='${{ toJson(github.event.pull_request.labels) }}'
-          if echo "$LABELS" | jq -r '.[].name' | grep -E "^(P0|P1)$" > /dev/null; then
-            HAS_P0_P1_LABEL="true"
-          fi
-          
-          # Check if PR already has force_on_cloud label
-          echo "HAS_FORCE_ON_CLOUD_LABEL=false" >> $GITHUB_ENV
-          if echo "$LABELS" | jq -r '.[].name' | grep -q "^force_on_cloud$"; then
-            HAS_FORCE_ON_CLOUD_LABEL="true"
-            echo "HAS_FORCE_ON_CLOUD_LABEL=true" >> $GITHUB_ENV
-          fi
-          
-          echo "Has P0/P1 label: $HAS_P0_P1_LABEL"
-          echo "Has force_on_cloud label: $HAS_FORCE_ON_CLOUD_LABEL"
-          
-          # Add force_on_cloud label if PR has P0/P1 and doesn't already have force_on_cloud
-          if [[ "$HAS_P0_P1_LABEL" == "true" && "$HAS_FORCE_ON_CLOUD_LABEL" == "false" ]]; then
-            echo "✅ Rule 2: PR has P0 or P1 label - adding force_on_cloud label"
-            curl -X POST \
-              -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
-              -H "Accept: application/vnd.github.v3+json" \
-              "https://api.github.com/repos/${{ github.repository }}/issues/${{ github.event.pull_request.number }}/labels" \
-              -d '{"labels":["force_on_cloud"]}'
-          elif [[ "$HAS_P0_P1_LABEL" == "true" && "$HAS_FORCE_ON_CLOUD_LABEL" == "true" ]]; then
-            echo "✅ Rule 2: PR has P0 or P1 label and already has force_on_cloud label - no action needed"
-          else
-            echo "✅ Rule 2: PR does not have P0 or P1 label - no force_on_cloud label needed"
-          fi
-
-          SKIP_UNIT_TEST_CUSTOM="false"
-          if echo "$LABELS" | jq -r '.[].name' | grep -q "^ci/skip_unit-tests_custom$"; then
-            SKIP_UNIT_TEST_CUSTOM="true"
-          fi
-          echo "SKIP_UNIT_TEST_CUSTOM=$SKIP_UNIT_TEST_CUSTOM" >> $GITHUB_ENV
-
-      - name: Rule 3 - Analyze changed files and set build requirements
-        run: |
-          # Get list of changed files
-          CHANGED_FILES=$(git diff --name-only ${{ github.event.pull_request.base.sha }} ${{ github.event.pull_request.head.sha }})
-          echo "Changed files:"
-          echo "$CHANGED_FILES"
-          echo ""
-          
-          # Initialize all requirements to false
-          REQUIRE_BUILD="false"
-          REQUIRE_DTEST="false"
-          REQUIRE_UNITTEST="false"
-          REQUIRE_ARTIFACTS="false"
-          REQUIRE_SCYLLA_GDB="false"
-          
-          # Check each file against patterns
-          while IFS= read -r file; do
-            if [[ -n "$file" ]]; then
-              echo "Checking file: $file"
-              
-              # Build pattern: ^(?!scripts\/pull_github_pr.sh).*$
-              # Everything except scripts/pull_github_pr.sh
-              if [[ "$file" != "scripts/pull_github_pr.sh" ]]; then
-                REQUIRE_BUILD="true"
-                echo "  ✓ Matches build pattern"
-              fi
-              
-              # Dtest pattern: ^(?!test(.py|\/)|dist\/docker\/|dist\/common\/scripts\/).*$
-              # Everything except test files, dist/docker/, dist/common/scripts/
-              if [[ ! "$file" =~ ^test\.(py|/).*$ ]] && [[ ! "$file" =~ ^dist/docker/.*$ ]] && [[ ! "$file" =~ ^dist/common/scripts/.*$ ]]; then
-                REQUIRE_DTEST="true"
-                echo "  ✓ Matches dtest pattern"
-              fi
-              
-              # Unittest pattern: ^(?!dist\/docker\/|dist\/common\/scripts).*$
-              # Everything except dist/docker/, dist/common/scripts/
-              if [[ ! "$file" =~ ^dist/docker/.*$ ]] && [[ ! "$file" =~ ^dist/common/scripts.*$ ]]; then
-                REQUIRE_UNITTEST="true"
-                echo "  ✓ Matches unittest pattern"
-              fi
-              
-              # Artifacts pattern: ^(?:dist|tools\/toolchain).*$
-              # Files starting with dist or tools/toolchain
-              if [[ "$file" =~ ^dist.*$ ]] || [[ "$file" =~ ^tools/toolchain.*$ ]]; then
-                REQUIRE_ARTIFACTS="true"
-                echo "  ✓ Matches artifacts pattern"
-              fi
-              
-              # Scylla GDB pattern: ^(scylla-gdb.py).*$
-              # Files starting with scylla-gdb.py
-              if [[ "$file" =~ ^scylla-gdb\.py.*$ ]]; then
-                REQUIRE_SCYLLA_GDB="true"
-                echo "  ✓ Matches scylla_gdb pattern"
-              fi
-            fi
-          done <<< "$CHANGED_FILES"
-          
-          # Set environment variables
-          echo "requireBuild=$REQUIRE_BUILD" >> $GITHUB_ENV
-          echo "requireDtest=$REQUIRE_DTEST" >> $GITHUB_ENV
-          echo "requireUnittest=$REQUIRE_UNITTEST" >> $GITHUB_ENV
-          echo "requireArtifacts=$REQUIRE_ARTIFACTS" >> $GITHUB_ENV
-          echo "requireScyllaGdb=$REQUIRE_SCYLLA_GDB" >> $GITHUB_ENV
-          
-          echo ""
-          echo "✅ Rule 3: File analysis complete"
-          echo "Build required: $REQUIRE_BUILD"
-          echo "Dtest required: $REQUIRE_DTEST"
-          echo "Unittest required: $REQUIRE_UNITTEST"
-          echo "Artifacts required: $REQUIRE_ARTIFACTS"
-          echo "Scylla GDB required: $REQUIRE_SCYLLA_GDB"
-
-      - name: Determine Jenkins Job Name
-        run: |
-          if [[ "${{ github.ref_name }}" == "next" ]]; then
-            FOLDER_NAME="scylla-master"
-          elif [[ "${{ github.ref_name }}" == "next-enterprise" ]]; then
-            FOLDER_NAME="scylla-enterprise"
-          else
-            VERSION=$(echo "${{ github.ref_name }}" | awk -F'-' '{print $2}')
-            if [[ "$VERSION" =~ ^202[0-4]\.[0-9]+$ ]]; then
-              FOLDER_NAME="enterprise-$VERSION"
-            elif [[ "$VERSION" =~ ^[0-9]+\.[0-9]+$ ]]; then
-              FOLDER_NAME="scylla-$VERSION"
-            fi
-          fi
-          echo "JOB_NAME=${FOLDER_NAME}/job/scylla-ci" >> $GITHUB_ENV
-
-      - name: Trigger Jenkins Job
-        if: env.draft_or_conflict == 'false' && env.has_file_changes == 'true' && github.action == 'opened' || github.action == 'reopened'
-        env:
-          JENKINS_USER: ${{ secrets.JENKINS_USERNAME }}
-          JENKINS_API_TOKEN: ${{ secrets.JENKINS_TOKEN }}
-          JENKINS_URL: "https://jenkins.scylladb.com"
-          SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
-        run: |
-          PR_NUMBER=${{ github.event.issue.number }}
-          PR_REPO_NAME=${{ github.event.repository.full_name }}
-          echo "Triggering Jenkins Job: $JOB_NAME"
-          curl -X POST \
-            "$JENKINS_URL/job/$JOB_NAME/buildWithParameters? \
-            PR_NUMBER=$PR_NUMBER& \
-            RUN_DTEST=$REQUIRE_DTEST& \
-            RUN_ONLY_SCYLLA_GDB=$REQUIRE_SCYLLA_GDB& \
-            RUN_UNIT_TEST=$REQUIRE_UNITTEST& \
-            FORCE_ON_CLOUD=$HAS_FORCE_ON_CLOUD_LABEL& \
-            SKIP_UNIT_TEST_CUSTOM=$SKIP_UNIT_TEST_CUSTOM& \
-            RUN_ARTIFACT_TESTS=$REQUIRE_ARTIFACTS" \
-            --fail \
-            --user "$JENKINS_USER:$JENKINS_API_TOKEN" \
-            -i -v
-  trigger-ci-via-comment:
-    if: github.event.comment.user.login != 'scylladbbot' && contains(github.event.comment.body, '@scylladbbot') && contains(github.event.comment.body, 'trigger-ci')
-    runs-on: ubuntu-latest
-    steps:
-      - name: Trigger Scylla-CI Jenkins Job
-        env:
-          JENKINS_USER: ${{ secrets.JENKINS_USERNAME }}
-          JENKINS_API_TOKEN: ${{ secrets.JENKINS_TOKEN }}
-          JENKINS_URL: "https://jenkins.scylladb.com"
-        run: |
-          PR_NUMBER=${{ github.event.issue.number }}
-          PR_REPO_NAME=${{ github.event.repository.full_name }}
-          curl -X POST "$JENKINS_URL/job/$JOB_NAME/buildWithParameters?PR_NUMBER=$PR_NUMBER" \
-          --user "$JENKINS_USER:$JENKINS_API_TOKEN" --fail -i -v

.gitmodules

@@ -1,6 +1,6 @@
 [submodule "seastar"]
 	path = seastar
-	url = ../seastar
+	url = ../scylla-seastar
 	ignore = dirty
 [submodule "swagger-ui"]
 	path = swagger-ui

CMakeLists.txt

@@ -49,7 +49,7 @@ include(limit_jobs)
 set(CMAKE_CXX_STANDARD "23" CACHE INTERNAL "")
 set(CMAKE_CXX_EXTENSIONS ON CACHE INTERNAL "")
 set(CMAKE_CXX_SCAN_FOR_MODULES OFF CACHE INTERNAL "")
-set(CMAKE_VISIBILITY_INLINES_HIDDEN ON)
+set(CMAKE_CXX_VISIBILITY_PRESET hidden)
 
 if(is_multi_config)
     find_package(Seastar)
@@ -90,13 +90,13 @@ if(is_multi_config)
     add_dependencies(Seastar::seastar_testing Seastar)
 else()
     set(Seastar_TESTING ON CACHE BOOL "" FORCE)
-    set(Seastar_API_LEVEL 9 CACHE STRING "" FORCE)
+    set(Seastar_API_LEVEL 8 CACHE STRING "" FORCE)
     set(Seastar_DEPRECATED_OSTREAM_FORMATTERS OFF CACHE BOOL "" FORCE)
     set(Seastar_APPS ON CACHE BOOL "" FORCE)
     set(Seastar_EXCLUDE_APPS_FROM_ALL ON CACHE BOOL "" FORCE)
     set(Seastar_EXCLUDE_TESTS_FROM_ALL ON CACHE BOOL "" FORCE)
     set(Seastar_IO_URING ON CACHE BOOL "" FORCE)
-    set(Seastar_SCHEDULING_GROUPS_COUNT 21 CACHE STRING "" FORCE)
+    set(Seastar_SCHEDULING_GROUPS_COUNT 20 CACHE STRING "" FORCE)
     set(Seastar_UNUSED_RESULT_ERROR ON CACHE BOOL "" FORCE)
     add_subdirectory(seastar)
     target_compile_definitions (seastar
@@ -116,7 +116,6 @@ list(APPEND absl_cxx_flags
 if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
     list(APPEND ABSL_GCC_FLAGS ${absl_cxx_flags})
 elseif(CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
-    list(APPEND absl_cxx_flags "-Wno-deprecated-builtins")
     list(APPEND ABSL_LLVM_FLAGS ${absl_cxx_flags})
 endif()
 set(ABSL_DEFAULT_LINKOPTS
@@ -164,45 +163,7 @@ file(MAKE_DIRECTORY "${scylla_gen_build_dir}")
 include(add_version_library)
 generate_scylla_version()
 
-option(Scylla_USE_PRECOMPILED_HEADER "Use precompiled header for Scylla" ON)
-add_library(scylla-precompiled-header STATIC exported_templates.cc)
-target_link_libraries(scylla-precompiled-header PRIVATE
-    absl::headers
-    absl::btree
-    absl::hash
-    absl::raw_hash_set
-    Seastar::seastar
-    Snappy::snappy
-    systemd
-    ZLIB::ZLIB
-    lz4::lz4_static
-    zstd::zstd_static)
-if (Scylla_USE_PRECOMPILED_HEADER)
-  set(Scylla_USE_PRECOMPILED_HEADER_USE ON)
-  find_program(DISTCC_EXEC NAMES distcc OPTIONAL)
-  if (DISTCC_EXEC)
-    if(DEFINED ENV{DISTCC_HOSTS})
-      set(Scylla_USE_PRECOMPILED_HEADER_USE OFF)
-      message(STATUS "Disabling precompiled header usage because distcc exists and DISTCC_HOSTS is set, assuming you're using distributed compilation.")
-    else()
-      file(REAL_PATH "~/.distcc/hosts" DIST_CC_HOSTS_PATH EXPAND_TILDE)
-      if (EXISTS ${DIST_CC_HOSTS_PATH})
-        set(Scylla_USE_PRECOMPILED_HEADER_USE OFF)
-        message(STATUS "Disabling precompiled header usage because distcc and ~/.distcc/hosts exists, assuming you're using distributed compilation.")
-      endif()
-    endif()
-  endif()
-  if (Scylla_USE_PRECOMPILED_HEADER_USE)
-    message(STATUS "Using precompiled header for Scylla - remember to add `sloppiness = pch_defines,time_macros` to ccache.conf, if you're using ccache.")
-    target_precompile_headers(scylla-precompiled-header PRIVATE "stdafx.hh")
-    target_compile_definitions(scylla-precompiled-header PRIVATE SCYLLA_USE_PRECOMPILED_HEADER)
-  endif()
-else()
-  set(Scylla_USE_PRECOMPILED_HEADER_USE OFF)
-endif()
-
 add_library(scylla-main STATIC)
-
 target_sources(scylla-main
   PRIVATE
     absl-flat_hash_map.cc
@@ -217,6 +178,7 @@ target_sources(scylla-main
     mutation_query.cc
     node_ops/task_manager_module.cc
     partition_slice_builder.cc
+    querier.cc
     query/query.cc
     query_ranges_to_vnodes.cc
     query/query-result-set.cc
@@ -247,7 +209,6 @@ target_link_libraries(scylla-main
     ZLIB::ZLIB
     lz4::lz4_static
     zstd::zstd_static
-    scylla-precompiled-header
 )
 
 option(Scylla_CHECK_HEADERS

CONTRIBUTING.md

@@ -12,7 +12,7 @@ Please use the [issue tracker](https://github.com/scylladb/scylla/issues/) to re
 
 ## Contributing code to Scylla
 
-Before you can contribute code to Scylla for the first time, you should sign the [Contributor License Agreement](https://www.scylladb.com/open-source/contributor-agreement/) and send the signed form to cla@scylladb.com. You can then submit your changes as patches to the [scylladb-dev mailing list](https://groups.google.com/forum/#!forum/scylladb-dev) or as a pull request to the [Scylla project on github](https://github.com/scylladb/scylla).
+Before you can contribute code to Scylla for the first time, you should sign the [Contributor License Agreement](https://www.scylladb.com/open-source/contributor-agreement/) and send the signed form cla@scylladb.com. You can then submit your changes as patches to the [scylladb-dev mailing list](https://groups.google.com/forum/#!forum/scylladb-dev) or as a pull request to the [Scylla project on github](https://github.com/scylladb/scylla).
 If you need help formatting or sending patches, [check out these instructions](https://github.com/scylladb/scylla/wiki/Formatting-and-sending-patches).
 
 The Scylla C++ source code uses the [Seastar coding style](https://github.com/scylladb/seastar/blob/master/coding-style.md) so please adhere to that in your patches. Note that Scylla code is written with `using namespace seastar`, so should not explicitly add the `seastar::` prefix to Seastar symbols. You will usually not need to add `using namespace seastar` to new source files, because most Scylla header files have `#include "seastarx.hh"`, which does this.

HACKING.md

@@ -43,7 +43,7 @@ $ ./tools/toolchain/dbuild ninja build/release/scylla
 $ ./tools/toolchain/dbuild ./build/release/scylla --developer-mode 1
 ```
 
-Note: do not mix environments - either perform all your work with dbuild, or natively on the host.
+Note: do not mix environemtns - either perform all your work with dbuild, or natively on the host.
 Note2: you can get to an interactive shell within dbuild by running it without any parameters:
 ```bash
 $ ./tools/toolchain/dbuild
@@ -91,7 +91,7 @@ You can also specify a single mode. For example
 $ ninja-build release
 ```
 
-Will build everything in release mode. The valid modes are
+Will build everytihng in release mode. The valid modes are
 
 * Debug: Enables [AddressSanitizer](https://github.com/google/sanitizers/wiki/AddressSanitizer)
   and other sanity checks. It has no optimizations, which allows for debugging with tools like
@@ -361,7 +361,7 @@ avoid that the gold linker can be told to create an index with
 
 More info at https://gcc.gnu.org/wiki/DebugFission.
 
-Both options can be enabled by passing `--split-dwarf` to configure.py.
+Both options can be enable by passing `--split-dwarf` to configure.py.
 
 Note that distcc is *not* compatible with it, but icecream
 (https://github.com/icecc/icecream) is.
@@ -370,7 +370,7 @@ Note that distcc is *not* compatible with it, but icecream
 
 Sometimes Scylla development is closely tied with a feature being developed in Seastar. It can be useful to compile Scylla with a particular check-out of Seastar.
 
-One way to do this is to create a local remote for the Seastar submodule in the Scylla repository:
+One way to do this it to create a local remote for the Seastar submodule in the Scylla repository:
 
 ```bash
 $ cd $HOME/src/scylla

README.md

@@ -18,7 +18,7 @@ Scylla is fairly fussy about its build environment, requiring very recent
 versions of the C++23 compiler and of many libraries to build. The document
 [HACKING.md](HACKING.md) includes detailed information on building and
 developing Scylla, but to get Scylla building quickly on (almost) any build
-machine, Scylla offers a [frozen toolchain](tools/toolchain/README.md).
+machine, Scylla offers a [frozen toolchain](tools/toolchain/README.md),
 This is a pre-configured Docker image which includes recent versions of all
 the required compilers, libraries and build tools. Using the frozen toolchain
 allows you to avoid changing anything in your build machine to meet Scylla's

SCYLLA-VERSION-GEN

@@ -78,7 +78,7 @@ fi
 
 # Default scylla product/version tags
 PRODUCT=scylla
-VERSION=2026.1.0-dev
+VERSION=2025.4.6
 
 if test -f version
 then

alternator/CMakeLists.txt

@@ -18,7 +18,6 @@ target_sources(alternator
     consumed_capacity.cc
     ttl.cc
     parsed_expression_cache.cc
-    http_compression.cc
     ${cql_grammar_srcs})
 target_include_directories(alternator
   PUBLIC
@@ -35,8 +34,5 @@ target_link_libraries(alternator
     idl
     absl::headers)
 
-if (Scylla_USE_PRECOMPILED_HEADER_USE)
-  target_precompile_headers(alternator REUSE_FROM scylla-precompiled-header)
-endif()
 check_headers(check-headers alternator
   GLOB_RECURSE ${CMAKE_CURRENT_SOURCE_DIR}/*.hh)

alternator/auth.cc

@@ -11,6 +11,7 @@
 #include "utils/log.hh"
 #include <string>
 #include <string_view>
+#include "bytes.hh"
 #include "alternator/auth.hh"
 #include <fmt/format.h>
 #include "auth/password_authenticator.hh"

alternator/conditions.cc

@@ -42,7 +42,7 @@ comparison_operator_type get_comparison_operator(const rjson::value& comparison_
     if (!comparison_operator.IsString()) {
         throw api_error::validation(fmt::format("Invalid comparison operator definition {}", rjson::print(comparison_operator)));
     }
-    std::string op = rjson::to_string(comparison_operator);
+    std::string op = comparison_operator.GetString();
     auto it = ops.find(op);
     if (it == ops.end()) {
         throw api_error::validation(fmt::format("Unsupported comparison operator {}", op));
@@ -377,8 +377,8 @@ bool check_compare(const rjson::value* v1, const rjson::value& v2, const Compara
         return cmp(unwrap_number(*v1, cmp.diagnostic), unwrap_number(v2, cmp.diagnostic));
     }
     if (kv1.name == "S") {
-        return cmp(rjson::to_string_view(kv1.value),
-                   rjson::to_string_view(kv2.value));
+        return cmp(std::string_view(kv1.value.GetString(), kv1.value.GetStringLength()),
+                   std::string_view(kv2.value.GetString(), kv2.value.GetStringLength()));
     }
     if (kv1.name == "B") {
         auto d_kv1 = unwrap_bytes(kv1.value, v1_from_query);
@@ -470,9 +470,9 @@ static bool check_BETWEEN(const rjson::value* v, const rjson::value& lb, const r
         return check_BETWEEN(unwrap_number(*v, diag), unwrap_number(lb, diag), unwrap_number(ub, diag), bounds_from_query);
     }
     if (kv_v.name == "S") {
-        return check_BETWEEN(rjson::to_string_view(kv_v.value),
-                             rjson::to_string_view(kv_lb.value),
-                             rjson::to_string_view(kv_ub.value),
+        return check_BETWEEN(std::string_view(kv_v.value.GetString(), kv_v.value.GetStringLength()),
+                             std::string_view(kv_lb.value.GetString(), kv_lb.value.GetStringLength()),
+                             std::string_view(kv_ub.value.GetString(), kv_ub.value.GetStringLength()),
                              bounds_from_query);
     }
     if (kv_v.name == "B") {

alternator/consumed_capacity.cc

@@ -8,8 +8,6 @@
 
 #include "consumed_capacity.hh"
 #include "error.hh"
-#include "utils/rjson.hh"
-#include <fmt/format.h>
 
 namespace alternator {
 
@@ -34,12 +32,12 @@ bool consumed_capacity_counter::should_add_capacity(const rjson::value& request)
     if (!return_consumed->IsString()) {
         throw api_error::validation("Non-string ReturnConsumedCapacity field in request");
     }
-    std::string_view consumed = rjson::to_string_view(*return_consumed);
+    std::string consumed = return_consumed->GetString();
     if (consumed == "INDEXES") {
         throw api_error::validation("INDEXES consumed capacity is not supported");
     }
     if (consumed != "TOTAL") {
-        throw api_error::validation(fmt::format("Unknown consumed capacity {}", consumed));
+        throw api_error::validation("Unknown consumed capacity "+ consumed);
     }
     return true;
 }

alternator/controller.cc

@@ -28,7 +28,6 @@ static logging::logger logger("alternator_controller");
 controller::controller(
         sharded<gms::gossiper>& gossiper,
         sharded<service::storage_proxy>& proxy,
-        sharded<service::storage_service>& ss,
         sharded<service::migration_manager>& mm,
         sharded<db::system_distributed_keyspace>& sys_dist_ks,
         sharded<cdc::generation_service>& cdc_gen_svc,
@@ -40,7 +39,6 @@ controller::controller(
     : protocol_server(sg)
     , _gossiper(gossiper)
     , _proxy(proxy)
-    , _ss(ss)
     , _mm(mm)
     , _sys_dist_ks(sys_dist_ks)
     , _cdc_gen_svc(cdc_gen_svc)
@@ -91,7 +89,7 @@ future<> controller::start_server() {
         auto get_timeout_in_ms = [] (const db::config& cfg) -> utils::updateable_value<uint32_t> {
             return cfg.alternator_timeout_in_ms;
         };
-        _executor.start(std::ref(_gossiper), std::ref(_proxy), std::ref(_ss), std::ref(_mm), std::ref(_sys_dist_ks),
+        _executor.start(std::ref(_gossiper), std::ref(_proxy), std::ref(_mm), std::ref(_sys_dist_ks),
                         sharded_parameter(get_cdc_metadata, std::ref(_cdc_gen_svc)), _ssg.value(),
                         sharded_parameter(get_timeout_in_ms, std::ref(_config))).get();
         _server.start(std::ref(_executor), std::ref(_proxy), std::ref(_gossiper), std::ref(_auth_service), std::ref(_sl_controller)).get();
@@ -105,23 +103,11 @@ future<> controller::start_server() {
             alternator_port = _config.alternator_port();
             _listen_addresses.push_back({addr, *alternator_port});
         }
-        std::optional<uint16_t> alternator_port_proxy_protocol;
-        if (_config.alternator_port_proxy_protocol()) {
-            alternator_port_proxy_protocol = _config.alternator_port_proxy_protocol();
-            _listen_addresses.push_back({addr, *alternator_port_proxy_protocol});
-        }
         std::optional<uint16_t> alternator_https_port;
-        std::optional<uint16_t> alternator_https_port_proxy_protocol;
         std::optional<tls::credentials_builder> creds;
-        if (_config.alternator_https_port() || _config.alternator_https_port_proxy_protocol()) {
-            if (_config.alternator_https_port()) {
-                alternator_https_port = _config.alternator_https_port();
-                _listen_addresses.push_back({addr, *alternator_https_port});
-            }
-            if (_config.alternator_https_port_proxy_protocol()) {
-                alternator_https_port_proxy_protocol = _config.alternator_https_port_proxy_protocol();
-                _listen_addresses.push_back({addr, *alternator_https_port_proxy_protocol});
-            }
+        if (_config.alternator_https_port()) {
+            alternator_https_port = _config.alternator_https_port();
+            _listen_addresses.push_back({addr, *alternator_https_port});
             creds.emplace();
             auto opts = _config.alternator_encryption_options();
             if (opts.empty()) {
@@ -147,29 +133,19 @@ future<> controller::start_server() {
             }
         }
         _server.invoke_on_all(
-                [this, addr, alternator_port, alternator_https_port, alternator_port_proxy_protocol, alternator_https_port_proxy_protocol, creds = std::move(creds)] (server& server) mutable {
-            return server.init(addr, alternator_port, alternator_https_port, alternator_port_proxy_protocol, alternator_https_port_proxy_protocol, creds,
+                [this, addr, alternator_port, alternator_https_port, creds = std::move(creds)] (server& server) mutable {
+            return server.init(addr, alternator_port, alternator_https_port, creds,
                     _config.alternator_enforce_authorization,
                     _config.alternator_warn_authorization,
-                    _config.alternator_max_users_query_size_in_trace_output,
                     &_memory_limiter.local().get_semaphore(),
                     _config.max_concurrent_requests_per_shard);
-        }).handle_exception([this, addr, alternator_port, alternator_https_port, alternator_port_proxy_protocol, alternator_https_port_proxy_protocol] (std::exception_ptr ep) {
-            logger.error("Failed to set up Alternator HTTP server on {} port {}, TLS port {}, proxy-protocol port {}, TLS proxy-protocol port {}: {}",
-                    addr,
-                    alternator_port ? std::to_string(*alternator_port) : "OFF",
-                    alternator_https_port ? std::to_string(*alternator_https_port) : "OFF",
-                    alternator_port_proxy_protocol ? std::to_string(*alternator_port_proxy_protocol) : "OFF",
-                    alternator_https_port_proxy_protocol ? std::to_string(*alternator_https_port_proxy_protocol) : "OFF",
-                    ep);
+        }).handle_exception([this, addr, alternator_port, alternator_https_port] (std::exception_ptr ep) {
+            logger.error("Failed to set up Alternator HTTP server on {} port {}, TLS port {}: {}",
+                    addr, alternator_port ? std::to_string(*alternator_port) : "OFF", alternator_https_port ? std::to_string(*alternator_https_port) : "OFF", ep);
             return stop_server().then([ep = std::move(ep)] { return make_exception_future<>(ep); });
-        }).then([addr, alternator_port, alternator_https_port, alternator_port_proxy_protocol, alternator_https_port_proxy_protocol] {
-            logger.info("Alternator server listening on {}, HTTP port {}, HTTPS port {}, proxy-protocol port {}, TLS proxy-protocol port {}",
-                    addr,
-                    alternator_port ? std::to_string(*alternator_port) : "OFF",
-                    alternator_https_port ? std::to_string(*alternator_https_port) : "OFF",
-                    alternator_port_proxy_protocol ? std::to_string(*alternator_port_proxy_protocol) : "OFF",
-                    alternator_https_port_proxy_protocol ? std::to_string(*alternator_https_port_proxy_protocol) : "OFF");
+        }).then([addr, alternator_port, alternator_https_port] {
+            logger.info("Alternator server listening on {}, HTTP port {}, HTTPS port {}",
+                    addr, alternator_port ? std::to_string(*alternator_port) : "OFF", alternator_https_port ? std::to_string(*alternator_https_port) : "OFF");
         }).get();
     });
 }
@@ -192,7 +168,7 @@ future<> controller::request_stop_server() {
     });
 }
 
-future<utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>>> controller::get_client_data() {
+future<utils::chunked_vector<client_data>> controller::get_client_data() {
     return _server.local().get_client_data();
 }
 

alternator/controller.hh

@@ -15,7 +15,6 @@
 
 namespace service {
 class storage_proxy;
-class storage_service;
 class migration_manager;
 class memory_limiter;
 }
@@ -58,7 +57,6 @@ class server;
 class controller : public protocol_server {
     sharded<gms::gossiper>& _gossiper;
     sharded<service::storage_proxy>& _proxy;
-    sharded<service::storage_service>& _ss;
     sharded<service::migration_manager>& _mm;
     sharded<db::system_distributed_keyspace>& _sys_dist_ks;
     sharded<cdc::generation_service>& _cdc_gen_svc;
@@ -76,7 +74,6 @@ public:
     controller(
         sharded<gms::gossiper>& gossiper,
         sharded<service::storage_proxy>& proxy,
-        sharded<service::storage_service>& ss,
         sharded<service::migration_manager>& mm,
         sharded<db::system_distributed_keyspace>& sys_dist_ks,
         sharded<cdc::generation_service>& cdc_gen_svc,
@@ -96,7 +93,7 @@ public:
     // This virtual function is called (on each shard separately) when the
     // virtual table "system.clients" is read. It is expected to generate a
     // list of clients connected to this server (on this shard).
-    virtual future<utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>>> get_client_data() override;
+    virtual future<utils::chunked_vector<client_data>> get_client_data() override;
 };
 
 }

alternator/error.hh

@@ -94,9 +94,6 @@ public:
     static api_error internal(std::string msg) {
         return api_error("InternalServerError", std::move(msg), http::reply::status_type::internal_server_error);
     }
-    static api_error payload_too_large(std::string msg) {
-        return api_error("PayloadTooLarge", std::move(msg), status_type::payload_too_large);
-    }
 
     // Provide the "std::exception" interface, to make it easier to print this
     // exception in log messages. Note that this function is *not* used to

alternator/executor.hh

@@ -17,13 +17,11 @@
 #include "service/client_state.hh"
 #include "service_permit.hh"
 #include "db/timeout_clock.hh"
-#include "db/config.hh"
 
 #include "alternator/error.hh"
 #include "stats.hh"
 #include "utils/rjson.hh"
 #include "utils/updateable_value.hh"
-#include "utils/simple_value_with_expiry.hh"
 
 #include "tracing/trace_state.hh"
 
@@ -43,7 +41,6 @@ namespace cql3::selection {
 namespace service {
     class storage_proxy;
     class cas_shard;
-    class storage_service;
 }
 
 namespace cdc {
@@ -60,7 +57,6 @@ class schema_builder;
 
 namespace alternator {
 
-enum class table_status;
 class rmw_operation;
 class put_or_delete_item;
 
@@ -140,7 +136,6 @@ class expression_cache;
 
 class executor : public peering_sharded_service<executor> {
     gms::gossiper& _gossiper;
-    service::storage_service& _ss;
     service::storage_proxy& _proxy;
     service::migration_manager& _mm;
     db::system_distributed_keyspace& _sdks;
@@ -153,11 +148,6 @@ class executor : public peering_sharded_service<executor> {
 
     std::unique_ptr<parsed::expression_cache> _parsed_expression_cache;
 
-    struct describe_table_info_manager;
-    std::unique_ptr<describe_table_info_manager> _describe_table_info_manager;
-
-    future<> cache_newly_calculated_size_on_all_shards(schema_ptr schema, std::uint64_t size_in_bytes, std::chrono::nanoseconds ttl);
-    future<> fill_table_size(rjson::value &table_description, schema_ptr schema, bool deleting);
 public:
     using client_state = service::client_state;
     // request_return_type is the return type of the executor methods, which
@@ -183,7 +173,6 @@ public:
 
     executor(gms::gossiper& gossiper,
              service::storage_proxy& proxy,
-             service::storage_service& ss,
              service::migration_manager& mm,
              db::system_distributed_keyspace& sdks,
              cdc::metadata& cdc_metadata,
@@ -231,8 +220,6 @@ private:
     friend class rmw_operation;
 
     static void describe_key_schema(rjson::value& parent, const schema&, std::unordered_map<std::string,std::string> * = nullptr, const std::map<sstring, sstring> *tags = nullptr);
-    future<rjson::value> fill_table_description(schema_ptr schema, table_status tbl_status, service::client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit);
-    future<executor::request_return_type> create_table_on_shard0(service::client_state&& client_state, tracing::trace_state_ptr trace_state, rjson::value request, bool enforce_authorization, bool warn_authorization, const db::tablets_mode_t::mode tablets_mode);
 
     future<> do_batch_write(
         std::vector<std::pair<schema_ptr, put_or_delete_item>> mutation_builders,

alternator/http_compression.cc

@@ -1,301 +0,0 @@
-/*
- * Copyright 2025-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#include "alternator/http_compression.hh"
-#include "alternator/server.hh"
-#include <seastar/coroutine/maybe_yield.hh>
-#include <zlib.h>
-
-static logging::logger slogger("alternator-http-compression");
-
-namespace alternator {
-
-
-static constexpr size_t compressed_buffer_size = 1024;
-class zlib_compressor {
-    z_stream _zs;
-    temporary_buffer<char> _output_buf;
-    noncopyable_function<future<>(temporary_buffer<char>&&)> _write_func;
-public:
-    zlib_compressor(bool gzip, int compression_level, noncopyable_function<future<>(temporary_buffer<char>&&)> write_func)
-     : _write_func(std::move(write_func)) {
-        memset(&_zs, 0, sizeof(_zs));
-        if (deflateInit2(&_zs, std::clamp(compression_level, Z_NO_COMPRESSION, Z_BEST_COMPRESSION), Z_DEFLATED,
-                (gzip ? 16 : 0) + MAX_WBITS, 8, Z_DEFAULT_STRATEGY) != Z_OK) {
-            // Should only happen if memory allocation fails
-            throw std::bad_alloc();
-        }
-    }
-    ~zlib_compressor() {
-        deflateEnd(&_zs);
-    }
-    future<> close() {
-        return compress(nullptr, 0, true);
-    }
-
-    future<> compress(const char* buf, size_t len, bool is_last_chunk = false) {
-        _zs.next_in = reinterpret_cast<unsigned char*>(const_cast<char*>(buf));
-        _zs.avail_in = (uInt) len;
-        int mode = is_last_chunk ? Z_FINISH : Z_NO_FLUSH;
-        while(_zs.avail_in > 0 || is_last_chunk) {
-            co_await coroutine::maybe_yield();
-            if (_output_buf.empty()) {
-                if (is_last_chunk) {
-                    uint32_t max_buffer_size = 0;
-                    deflatePending(&_zs, &max_buffer_size, nullptr);
-                    max_buffer_size += deflateBound(&_zs, _zs.avail_in) + 1;
-                    _output_buf = temporary_buffer<char>(std::min(compressed_buffer_size, (size_t) max_buffer_size));
-                } else {
-                    _output_buf = temporary_buffer<char>(compressed_buffer_size);
-                }
-                _zs.next_out = reinterpret_cast<unsigned char*>(_output_buf.get_write());
-                _zs.avail_out = compressed_buffer_size;
-            }
-            int e = deflate(&_zs, mode);
-            if (e < Z_OK) {
-                throw api_error::internal("Error during compression of response body");
-            }
-            if (e == Z_STREAM_END || _zs.avail_out < compressed_buffer_size / 4) {
-                _output_buf.trim(compressed_buffer_size - _zs.avail_out);
-                co_await _write_func(std::move(_output_buf));
-                if (e == Z_STREAM_END) {
-                    break;
-                }
-            }
-        }
-    }
-};
-
-// Helper string_view functions for parsing Accept-Encoding header
-struct case_insensitive_cmp_sv {
-    bool operator()(std::string_view s1, std::string_view s2) const {
-        return std::equal(s1.begin(), s1.end(), s2.begin(), s2.end(),
-            [](char a, char b) { return ::tolower(a) == ::tolower(b); });
-    }
-};
-static inline std::string_view trim_left(std::string_view sv) {
-    while (!sv.empty() && std::isspace(static_cast<unsigned char>(sv.front())))
-        sv.remove_prefix(1);
-    return sv;
-}
-static inline std::string_view trim_right(std::string_view sv) {
-    while (!sv.empty() && std::isspace(static_cast<unsigned char>(sv.back())))
-        sv.remove_suffix(1);
-    return sv;
-}
-static inline std::string_view trim(std::string_view sv) {
-    return trim_left(trim_right(sv));
-}
-
-inline std::vector<std::string_view> split(std::string_view text, char separator) {
-    std::vector<std::string_view> tokens;
-    if (text == "") {
-        return tokens;
-    }
-
-    while (true) {
-        auto pos = text.find_first_of(separator);
-        if (pos != std::string_view::npos) {
-            tokens.emplace_back(text.data(), pos);
-            text.remove_prefix(pos + 1);
-        } else {
-            tokens.emplace_back(text);
-            break;
-        }
-    }
-    return tokens;
-}
-
-constexpr response_compressor::compression_type response_compressor::get_compression_type(std::string_view encoding) {
-    for (size_t i = 0; i < static_cast<size_t>(compression_type::count); ++i) {
-        if (case_insensitive_cmp_sv{}(encoding, compression_names[i])) {
-            return static_cast<compression_type>(i);
-        }
-    }
-    return compression_type::unknown;
-}
-
-response_compressor::compression_type response_compressor::find_compression(std::string_view accept_encoding, size_t response_size) {
-    std::optional<float> ct_q[static_cast<size_t>(compression_type::count)];
-    ct_q[static_cast<size_t>(compression_type::none)] = std::numeric_limits<float>::min(); // enabled, but lowest priority
-    compression_type selected_ct = compression_type::none;
-
-    std::vector<std::string_view> entries = split(accept_encoding, ',');
-    for (auto& e : entries) {
-        std::vector<std::string_view> params = split(e, ';');
-        if (params.size() == 0) {
-            continue;
-        }
-        compression_type ct = get_compression_type(trim(params[0]));
-        if (ct == compression_type::unknown) {
-            continue; // ignore unknown encoding types
-        }
-        if (ct_q[static_cast<size_t>(ct)].has_value() && ct_q[static_cast<size_t>(ct)] != 0.0f) {
-            continue; // already processed this encoding
-        }
-        if (response_size < _threshold[static_cast<size_t>(ct)]) {
-            continue; // below threshold treat as unknown
-        }
-        for (size_t i = 1; i < params.size(); ++i) { // find "q=" parameter
-            auto pos = params[i].find("q=");
-            if (pos == std::string_view::npos) {
-                continue;
-            }
-            std::string_view param = params[i].substr(pos + 2);
-            param = trim(param);
-            // parse quality value
-            float q_value = 1.0f;
-            auto [ptr, ec] = std::from_chars(param.data(), param.data() + param.size(), q_value);
-            if (ec != std::errc() || ptr != param.data() + param.size()) {
-                continue;
-            }
-            if (q_value < 0.0) {
-                q_value = 0.0;
-            } else if (q_value > 1.0) {
-                q_value = 1.0;
-            }
-            ct_q[static_cast<size_t>(ct)] = q_value;
-            break; // we parsed quality value
-        }
-        if (!ct_q[static_cast<size_t>(ct)].has_value()) {
-            ct_q[static_cast<size_t>(ct)] = 1.0f; // default quality value
-        }
-        // keep the highest encoding (in the order, unless 'any')
-        if (selected_ct == compression_type::any) {
-            if (ct_q[static_cast<size_t>(ct)] >= ct_q[static_cast<size_t>(selected_ct)]) {
-                selected_ct = ct;
-            }
-        } else {
-            if (ct_q[static_cast<size_t>(ct)] > ct_q[static_cast<size_t>(selected_ct)]) {
-                selected_ct = ct;
-            }
-        }
-    }
-    if (selected_ct == compression_type::any) {
-        // select any not mentioned or highest quality
-        selected_ct = compression_type::none;
-        for (size_t i = 0; i < static_cast<size_t>(compression_type::compressions_count); ++i) {
-            if (!ct_q[i].has_value()) {
-                return static_cast<compression_type>(i);
-            }
-            if (ct_q[i] > ct_q[static_cast<size_t>(selected_ct)]) {
-                selected_ct = static_cast<compression_type>(i);
-            }
-        }
-    }
-    return selected_ct;
-}
-
-static future<chunked_content> compress(response_compressor::compression_type ct, const db::config& cfg, std::string str) {
-    chunked_content compressed;
-    auto write = [&compressed](temporary_buffer<char>&& buf) -> future<> {
-        compressed.push_back(std::move(buf));
-        return make_ready_future<>();
-    };
-    zlib_compressor compressor(ct != response_compressor::compression_type::deflate,
-        cfg.alternator_response_gzip_compression_level(), std::move(write));
-    co_await compressor.compress(str.data(), str.size(), true);
-    co_return compressed;
-}
-
-static sstring flatten(chunked_content&& cc) {
-    size_t total_size = 0;
-    for (const auto& chunk : cc) {
-        total_size += chunk.size();
-    }
-    sstring result = sstring{ sstring::initialized_later{}, total_size };
-    size_t offset = 0;
-    for (const auto& chunk : cc) {
-        std::copy(chunk.begin(), chunk.end(), result.begin() + offset);
-        offset += chunk.size();
-    }
-    return result;
-}
-
-future<std::unique_ptr<http::reply>> response_compressor::generate_reply(std::unique_ptr<http::reply> rep, sstring accept_encoding, const char* content_type, std::string&& response_body) {
-    response_compressor::compression_type ct = find_compression(accept_encoding, response_body.size());
-    if (ct != response_compressor::compression_type::none) {
-        rep->add_header("Content-Encoding", get_encoding_name(ct));
-        rep->set_content_type(content_type);
-        return compress(ct, cfg, std::move(response_body)).then([rep = std::move(rep)] (chunked_content compressed) mutable {
-            rep->_content = flatten(std::move(compressed));
-            return make_ready_future<std::unique_ptr<http::reply>>(std::move(rep));
-        });
-    } else {
-        // Note that despite the move, there is a copy here -
-        // as str is std::string and rep->_content is sstring.
-        rep->_content = std::move(response_body);
-        rep->set_content_type(content_type);
-    }
-    return make_ready_future<std::unique_ptr<http::reply>>(std::move(rep));
-}
-
-template<typename Compressor>
-class compressed_data_sink_impl : public data_sink_impl {
-    output_stream<char> _out;
-    Compressor _compressor;
-public:
-    template<typename... Args>
-    compressed_data_sink_impl(output_stream<char>&& out, Args&&... args)
-     : _out(std::move(out)), _compressor(std::forward<Args>(args)..., [this](temporary_buffer<char>&& buf) {
-        return _out.write(std::move(buf));
-    }) { }
-
-    future<> put(std::span<temporary_buffer<char>> data) override {
-        return data_sink_impl::fallback_put(data, [this] (temporary_buffer<char>&& buf) {
-            return do_put(std::move(buf));
-        });
-    }
-
-private:
-    future<> do_put(temporary_buffer<char> buf) {
-        co_return co_await _compressor.compress(buf.get(), buf.size());
-
-    }
-    future<> close() override {
-        return _compressor.close().then([this] {
-            return _out.close();
-        });
-    }
-};
-
-executor::body_writer compress(response_compressor::compression_type ct, const db::config& cfg, executor::body_writer&& bw) {
-    return [bw = std::move(bw), ct, level = cfg.alternator_response_gzip_compression_level()](output_stream<char>&& out) mutable -> future<> {
-        output_stream_options opts;
-        opts.trim_to_size = true;
-        std::unique_ptr<data_sink_impl> data_sink_impl;
-        switch (ct) {
-            case response_compressor::compression_type::gzip:
-                data_sink_impl = std::make_unique<compressed_data_sink_impl<zlib_compressor>>(std::move(out), true, level);
-                break;
-            case response_compressor::compression_type::deflate:
-                data_sink_impl = std::make_unique<compressed_data_sink_impl<zlib_compressor>>(std::move(out), false, level);
-                break;
-            case response_compressor::compression_type::none:
-            case response_compressor::compression_type::any:
-            case response_compressor::compression_type::unknown:
-                on_internal_error(slogger,"Compression not selected");
-            default:
-                on_internal_error(slogger, "Unsupported compression type for data sink");
-        }
-        return bw(output_stream<char>(data_sink(std::move(data_sink_impl)), compressed_buffer_size, opts));
-    };
-}
-
-future<std::unique_ptr<http::reply>> response_compressor::generate_reply(std::unique_ptr<http::reply> rep, sstring accept_encoding, const char* content_type, executor::body_writer&& body_writer) {
-    response_compressor::compression_type ct = find_compression(accept_encoding, std::numeric_limits<size_t>::max());
-    if (ct != response_compressor::compression_type::none) {
-        rep->add_header("Content-Encoding", get_encoding_name(ct));
-        rep->write_body(content_type, compress(ct, cfg, std::move(body_writer)));
-    } else {
-        rep->write_body(content_type, std::move(body_writer));
-    }
-    return make_ready_future<std::unique_ptr<http::reply>>(std::move(rep));
-}
-
-} // namespace alternator

alternator/http_compression.hh

@@ -1,91 +0,0 @@
-/*
- * Copyright 2025-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#pragma once
-
-#include "alternator/executor.hh"
-#include <seastar/http/httpd.hh>
-#include "db/config.hh"
-
-namespace alternator {
-
-class response_compressor {
-public:
-    enum class compression_type {
-        gzip,
-        deflate,
-        compressions_count,
-        any = compressions_count,
-        none,
-        count,
-        unknown = count
-    };
-    static constexpr std::string_view compression_names[] = {
-        "gzip",
-        "deflate",
-        "*",
-        "identity"
-    };
-
-    static sstring get_encoding_name(compression_type ct) {
-        return sstring(compression_names[static_cast<size_t>(ct)]);
-    }
-    static constexpr compression_type get_compression_type(std::string_view encoding);
-
-    sstring get_accepted_encoding(const http::request& req) {
-        if (get_threshold() == 0) {
-            return "";
-        }
-        return req.get_header("Accept-Encoding");
-    }
-    compression_type find_compression(std::string_view accept_encoding, size_t response_size);
-
-    response_compressor(const db::config& cfg)
-        : cfg(cfg)
-        ,_gzip_level_observer(
-            cfg.alternator_response_gzip_compression_level.observe([this](int v) {
-                    update_threshold();
-                }))
-        ,_gzip_threshold_observer(
-            cfg.alternator_response_compression_threshold_in_bytes.observe([this](uint32_t v) {
-                    update_threshold();
-                }))
-    {
-        update_threshold();
-    }
-    response_compressor(const response_compressor& rhs) : response_compressor(rhs.cfg) {}
-
-private:
-    const db::config& cfg;
-    utils::observable<int>::observer _gzip_level_observer;
-    utils::observable<uint32_t>::observer _gzip_threshold_observer;
-    uint32_t _threshold[static_cast<size_t>(compression_type::count)];
-
-    size_t get_threshold() { return _threshold[static_cast<size_t>(compression_type::any)]; }
-    void update_threshold() {
-        _threshold[static_cast<size_t>(compression_type::none)] = std::numeric_limits<uint32_t>::max();
-        _threshold[static_cast<size_t>(compression_type::any)] = std::numeric_limits<uint32_t>::max();
-        uint32_t gzip = cfg.alternator_response_gzip_compression_level() <= 0 ? std::numeric_limits<uint32_t>::max()
-            : cfg.alternator_response_compression_threshold_in_bytes();
-        _threshold[static_cast<size_t>(compression_type::gzip)] = gzip;
-        _threshold[static_cast<size_t>(compression_type::deflate)] = gzip;
-        for (size_t i = 0; i < static_cast<size_t>(compression_type::compressions_count); ++i) {
-            if (_threshold[i] < _threshold[static_cast<size_t>(compression_type::any)]) {
-                _threshold[static_cast<size_t>(compression_type::any)] = _threshold[i];
-            }
-        }
-    }
-
-public:
-    future<std::unique_ptr<http::reply>> generate_reply(std::unique_ptr<http::reply> rep,
-         sstring accept_encoding, const char* content_type, std::string&& response_body);
-    future<std::unique_ptr<http::reply>> generate_reply(std::unique_ptr<http::reply> rep,
-         sstring accept_encoding, const char* content_type, executor::body_writer&& body_writer);
-};
-
-}

alternator/rmw_operation.hh

@@ -8,8 +8,6 @@
 
 #pragma once
 
-#include "cdc/cdc_options.hh"
-#include "cdc/log.hh"
 #include "seastarx.hh"
 #include "service/paxos/cas_request.hh"
 #include "service/cas_shard.hh"
@@ -58,7 +56,7 @@ public:
     static write_isolation get_write_isolation_for_schema(schema_ptr schema);
 
     static write_isolation default_write_isolation;
-
+public:
     static void set_default_write_isolation(std::string_view mode);
 
 protected:
@@ -109,11 +107,10 @@ public:
     // violating this). We mark apply() "const" to let the compiler validate
     // this for us. The output-only field _return_attributes is marked
     // "mutable" above so that apply() can still write to it.
-    virtual std::optional<mutation> apply(std::unique_ptr<rjson::value> previous_item, api::timestamp_type ts, cdc::per_request_options& cdc_opts) const = 0;
+    virtual std::optional<mutation> apply(std::unique_ptr<rjson::value> previous_item, api::timestamp_type ts) const = 0;
     // Convert the above apply() into the signature needed by cas_request:
-    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;
+    virtual std::optional<mutation> apply(foreign_ptr<lw_shared_ptr<query::result>> qr, const query::partition_slice& slice, api::timestamp_type ts) override;
     virtual ~rmw_operation() = default;
-    const wcu_consumed_capacity_counter& consumed_capacity() const noexcept { return _consumed_capacity; }
     schema_ptr schema() const { return _schema; }
     const rjson::value& request() const { return _request; }
     rjson::value&& move_request() && { return std::move(_request); }
@@ -127,9 +124,6 @@ public:
             stats& per_table_stats,
             uint64_t& wcu_total);
     std::optional<service::cas_shard> shard_for_execute(bool needs_read_before_write);
-
-private:
-    inline bool should_fill_preimage() const { return _schema->cdc_options().enabled(); }
 };
 
 } // namespace alternator

alternator/serialization.cc

@@ -12,7 +12,7 @@
 #include "serialization.hh"
 #include "error.hh"
 #include "types/concrete_types.hh"
-#include "types/json_utils.hh"
+#include "cql3/type_json.hh"
 #include "mutation/position_in_partition.hh"
 
 static logging::logger slogger("alternator-serialization");
@@ -496,7 +496,7 @@ const std::pair<std::string, const rjson::value*> unwrap_set(const rjson::value&
         return {"", nullptr};
     }
     auto it = v.MemberBegin();
-    const std::string it_key = rjson::to_string(it->name);
+    const std::string it_key = it->name.GetString();
     if (it_key != "SS" && it_key != "BS" && it_key != "NS") {
         return {std::move(it_key), nullptr};
     }

alternator/server.cc

@@ -13,7 +13,6 @@
 #include <seastar/http/function_handlers.hh>
 #include <seastar/http/short_streams.hh>
 #include <seastar/core/coroutine.hh>
-#include <seastar/coroutine/maybe_yield.hh>
 #include <seastar/util/defer.hh>
 #include <seastar/util/short_streams.hh>
 #include "seastarx.hh"
@@ -33,8 +32,6 @@
 #include "utils/aws_sigv4.hh"
 #include "client_data.hh"
 #include "utils/updateable_value.hh"
-#include <zlib.h>
-#include "alternator/http_compression.hh"
 
 static logging::logger slogger("alternator-server");
 
@@ -112,12 +109,9 @@ class api_handler : public handler_base {
     // type applies to all replies, both success and error.
     static constexpr const char* REPLY_CONTENT_TYPE = "application/x-amz-json-1.0";
 public:
-    api_handler(const std::function<future<executor::request_return_type>(std::unique_ptr<request> req)>& _handle,
-                const db::config& config) : _response_compressor(config), _f_handle(
+    api_handler(const std::function<future<executor::request_return_type>(std::unique_ptr<request> req)>& _handle) : _f_handle(
          [this, _handle](std::unique_ptr<request> req, std::unique_ptr<reply> rep) {
-         sstring accept_encoding = _response_compressor.get_accepted_encoding(*req);
-         return seastar::futurize_invoke(_handle, std::move(req)).then_wrapped(
-            [this, rep = std::move(rep), accept_encoding=std::move(accept_encoding)](future<executor::request_return_type> resf) mutable {
+         return seastar::futurize_invoke(_handle, std::move(req)).then_wrapped([this, rep = std::move(rep)](future<executor::request_return_type> resf) mutable {
              if (resf.failed()) {
                  // Exceptions of type api_error are wrapped as JSON and
                  // returned to the client as expected. Other types of
@@ -137,20 +131,22 @@ public:
                  return make_ready_future<std::unique_ptr<reply>>(std::move(rep));
              }
              auto res = resf.get();
-             return std::visit(overloaded_functor {
+             std::visit(overloaded_functor {
                 [&] (std::string&& str) {
-                    return _response_compressor.generate_reply(std::move(rep), std::move(accept_encoding),
-                                                               REPLY_CONTENT_TYPE, std::move(str));
+                    // Note that despite the move, there is a copy here -
+                    // as str is std::string and rep->_content is sstring.
+                    rep->_content = std::move(str);
+                    rep->set_content_type(REPLY_CONTENT_TYPE);
                 },
                 [&] (executor::body_writer&& body_writer) {
-                    return _response_compressor.generate_reply(std::move(rep), std::move(accept_encoding),
-                                                               REPLY_CONTENT_TYPE, std::move(body_writer));
+                    rep->write_body(REPLY_CONTENT_TYPE, std::move(body_writer));
                 },
                 [&] (const api_error& err) {
                     generate_error_reply(*rep, err);
-                    return make_ready_future<std::unique_ptr<reply>>(std::move(rep));
                 }
              }, std::move(res));
+
+             return make_ready_future<std::unique_ptr<reply>>(std::move(rep));
          });
     }) { }
 
@@ -179,7 +175,6 @@ protected:
         slogger.trace("api_handler error case: {}", rep._content);
     }
 
-    response_compressor _response_compressor;
     future_handler_function _f_handle;
 };
 
@@ -374,40 +369,13 @@ future<std::string> server::verify_signature(const request& req, const chunked_c
     for (const auto& header : signed_headers) {
         signed_headers_map.emplace(header, std::string_view());
     }
-    std::vector<std::string> modified_values;
     for (auto& header : req._headers) {
         std::string header_str;
         header_str.resize(header.first.size());
         std::transform(header.first.begin(), header.first.end(), header_str.begin(), ::tolower);
         auto it = signed_headers_map.find(header_str);
         if (it != signed_headers_map.end()) {
-            // replace multiple spaces in the header value header.second with
-            // a single space, as required by AWS SigV4 header canonization.
-            // If we modify the value, we need to save it in modified_values
-            // to keep it alive.
-            std::string value;
-            value.reserve(header.second.size());
-            bool prev_space = false;
-            bool modified = false;
-            for (char ch : header.second) {
-                if (ch == ' ') {
-                    if (!prev_space) {
-                        value += ch;
-                        prev_space = true;
-                    } else {
-                        modified = true; // skip a space
-                    }
-                } else {
-                    value += ch;
-                    prev_space = false;
-                }
-            }
-            if (modified) {
-                modified_values.emplace_back(std::move(value));
-                it->second = std::string_view(modified_values.back());
-            } else {
-                it->second = std::string_view(header.second);
-            }
+            it->second = std::string_view(header.second);
         }
     }
 
@@ -420,7 +388,6 @@ future<std::string> server::verify_signature(const request& req, const chunked_c
                                                     datestamp = std::move(datestamp),
                                                     signed_headers_str = std::move(signed_headers_str),
                                                     signed_headers_map = std::move(signed_headers_map),
-                                                    modified_values = std::move(modified_values),
                                                     region = std::move(region),
                                                     service = std::move(service),
                                                     user_signature = std::move(user_signature)] (future<key_cache::value_ptr> key_ptr_fut) {
@@ -461,82 +428,35 @@ static tracing::trace_state_ptr create_tracing_session(tracing::tracing& tracing
     return tracing_instance.create_session(tracing::trace_type::QUERY, props);
 }
 
-// A helper class to represent a potentially truncated view of a chunked_content.
-// If the content is short enough and single chunked, it just holds a view into the content.
-// Otherwise it will be copied into an internal buffer, possibly truncated (depending on maximum allowed size passed in),
-// and the view will point into that buffer.
-// `as_view()` method will return the view.
-// `take_as_sstring()` will either move out the internal buffer (if any), or create a new sstring from the view.
-// You should consider `as_view()` valid as long both the original chunked_content and the truncated_content object are alive.
-class truncated_content {
-    std::string_view _view;
-    sstring _content_maybe;
-
-    void copy_from_content(const chunked_content& content) {
-        size_t offset = 0;
-        for(auto &tmp : content) {
-            size_t to_copy = std::min(tmp.size(), _content_maybe.size() - offset);
-            std::copy(tmp.get(), tmp.get() + to_copy, _content_maybe.data() + offset);
-            offset += to_copy;
-            if (offset >= _content_maybe.size()) {
-                break;
-            }
-        }
+// truncated_content_view() prints a potentially long chunked_content for
+// debugging purposes. In the common case when the content is not excessively
+// long, it just returns a view into the given content, without any copying.
+// But when the content is very long, it is truncated after some arbitrary
+// max_len (or one chunk, whichever comes first), with "<truncated>" added at
+// the end. To do this modification to the string, we need to create a new
+// std::string, so the caller must pass us a reference to one, "buf", where
+// we can store the content. The returned view is only alive for as long this
+// buf is kept alive.
+static std::string_view truncated_content_view(const chunked_content& content, std::string& buf) {
+    constexpr size_t max_len = 1024;
+    if (content.empty()) {
+        return std::string_view();
+    } else if (content.size() == 1 && content.begin()->size() <= max_len) {
+        return std::string_view(content.begin()->get(), content.begin()->size());
+    } else {
+        buf = std::string(content.begin()->get(), std::min(content.begin()->size(), max_len)) + "<truncated>";
+        return std::string_view(buf);
     }
-public:
-    truncated_content(const chunked_content& content, size_t max_len = std::numeric_limits<size_t>::max()) {
-        if (content.empty()) return;
-        if (content.size() == 1 && content.begin()->size() <= max_len) {
-            _view = std::string_view(content.begin()->get(), content.begin()->size());
-            return;
-        }
-
-        constexpr std::string_view truncated_text = "<truncated>";
-        size_t content_size = 0;
-        for(auto &tmp : content) {
-            content_size += tmp.size();
-        }
-        if (content_size <= max_len) {
-            _content_maybe = sstring{ sstring::initialized_later{}, content_size };
-            copy_from_content(content);
-        }
-        else {
-            _content_maybe = sstring{ sstring::initialized_later{}, max_len + truncated_text.size() };
-            copy_from_content(content);
-            std::copy(truncated_text.begin(), truncated_text.end(), _content_maybe.data() + _content_maybe.size() - truncated_text.size());
-        }
-        _view = std::string_view(_content_maybe);
-    }
-
-    std::string_view as_view() const { return _view; }
-    sstring take_as_sstring() && {
-        if (_content_maybe.empty() && !_view.empty()) {
-            return sstring{_view};
-        }
-        return std::move(_content_maybe);
-    }
-};
-
-// `truncated_content_view` will produce an object representing a view to a passed content
-// possibly truncated at some length. The value returned is used in two ways:
-// - to print it in logs (use `as_view()` method for this)
-// - to pass it to tracing object, where it will be stored and used later
-//   (use `take_as_sstring()` method as this produces a copy in form of a sstring)
-// `truncated_content` delays constructing `sstring` object until it's actually needed.
-// `truncated_content` is valid as long as passed `content` is alive.
-// if the content is truncated, `<truncated>` will be appended at the maximum size limit
-// and total size will be `max_users_query_size_in_trace_output() + strlen("<truncated>")`.
-static truncated_content truncated_content_view(const chunked_content& content, size_t max_size) {
-    return truncated_content{content, max_size};
 }
 
-static tracing::trace_state_ptr maybe_trace_query(service::client_state& client_state, std::string_view username, std::string_view op, const chunked_content& query, size_t max_users_query_size_in_trace_output) {
+static tracing::trace_state_ptr maybe_trace_query(service::client_state& client_state, std::string_view username, std::string_view op, const chunked_content& query) {
     tracing::trace_state_ptr trace_state;
     tracing::tracing& tracing_instance = tracing::tracing::get_local_tracing_instance();
     if (tracing_instance.trace_next_query() || tracing_instance.slow_query_tracing_enabled()) {
         trace_state = create_tracing_session(tracing_instance);
+        std::string buf;
         tracing::add_session_param(trace_state, "alternator_op", op);
-        tracing::add_query(trace_state, truncated_content_view(query, max_users_query_size_in_trace_output).take_as_sstring());
+        tracing::add_query(trace_state, truncated_content_view(query, buf));
         tracing::begin(trace_state, seastar::format("Alternator {}", op), client_state.get_client_address());
         if (!username.empty()) {
             tracing::set_username(trace_state, auth::authenticated_user(username));
@@ -545,215 +465,37 @@ static tracing::trace_state_ptr maybe_trace_query(service::client_state& client_
     return trace_state;
 }
 
-// This read_entire_stream() is similar to Seastar's read_entire_stream()
-// which reads the given content_stream until its end into non-contiguous
-// memory. The difference is that this implementation takes an extra length
-// limit, and throws an error if we read more than this limit.
-// This length-limited variant would not have been needed if Seastar's HTTP
-// server's set_content_length_limit() worked in every case, but unfortunately
-// it does not - it only works if the request has a Content-Length header (see
-// issue #8196). In contrast this function can limit the request's length no
-// matter how it's encoded. We need this limit to protect Alternator from
-// oversized requests that can deplete memory.
-static future<chunked_content>
-read_entire_stream(input_stream<char>& inp, size_t length_limit) {
-    chunked_content ret;
-    // We try to read length_limit + 1 bytes, so that we can throw an
-    // exception if we managed to read more than length_limit.
-    ssize_t remain = length_limit + 1;
-    do {
-        temporary_buffer<char> buf = co_await inp.read_up_to(remain);
-        if (buf.empty()) {
-            break;
-        }
-        remain -= buf.size();
-        ret.push_back(std::move(buf));
-    } while (remain > 0);
-    // If we read the full length_limit + 1 bytes, we went over the limit:
-    if (remain <= 0) {
-        // By throwing here an error, we may send a reply (the error message)
-        // without having read the full request body. Seastar's httpd will
-        // realize that we have not read the entire content stream, and
-        // correctly mark the connection unreusable, i.e., close it.
-        // This means we are currently exposed to issue #12166 caused by
-        // Seastar issue 1325), where the client may get an RST instead of
-        // a FIN, and may rarely get a "Connection reset by peer" before
-        // reading the error we send.
-        throw api_error::payload_too_large(fmt::format("Request content length limit of {} bytes exceeded", length_limit));
-    }
-    co_return ret;
-}
-
-// safe_gzip_stream is an exception-safe wrapper for zlib's z_stream.
-// The "z_stream" struct is used by zlib to hold state while decompressing a
-// stream of data. It allocates memory which must be freed with inflateEnd(),
-// which the destructor of this class does.
-class safe_gzip_zstream {
-    z_stream _zs;
-public:
-    // If gzip is true, decode a gzip header (for "Content-Encoding: gzip").
-    // Otherwise, a zlib header (for "Content-Encoding: deflate").
-    safe_gzip_zstream(bool gzip = true) {
-        memset(&_zs, 0, sizeof(_zs));
-        if (inflateInit2(&_zs, gzip ? 16 + MAX_WBITS : MAX_WBITS) != Z_OK) {
-            // Should only happen if memory allocation fails
-            throw std::bad_alloc();
-        }
-    }
-    ~safe_gzip_zstream() {
-        inflateEnd(&_zs);
-    }
-    z_stream* operator->() {
-        return &_zs;
-    }
-    z_stream* get() {
-        return &_zs;
-    }
-    void reset() {
-        inflateReset(&_zs);
-    }
-};
-
-// ungzip() takes a chunked_content of a compressed request body, and returns
-// the uncompressed content as a chunked_content. If gzip is true, we expect
-// gzip header (for "Content-Encoding: gzip"), if gzip is false, we expect a
-// zlib header (for "Content-Encoding: deflate").
-// If the uncompressed content exceeds length_limit, an error is thrown.
-static future<chunked_content>
-ungzip(chunked_content&& compressed_body, size_t length_limit, bool gzip = true) {
-    chunked_content ret;
-    // output_buf can be any size - when uncompressing input_buf, it doesn't
-    // need to fit in a single output_buf, we'll use multiple output_buf for
-    // a single input_buf if needed.
-    constexpr size_t OUTPUT_BUF_SIZE = 4096;
-    temporary_buffer<char> output_buf;
-    safe_gzip_zstream strm(gzip);
-    bool complete_stream = false; // empty input is not a valid gzip/deflate
-    size_t total_out_bytes = 0;
-    for (const temporary_buffer<char>& input_buf : compressed_body) {
-        if (input_buf.empty()) {
-            continue;
-        }
-        complete_stream = false;
-        strm->next_in = (Bytef*) input_buf.get();
-        strm->avail_in = (uInt) input_buf.size();
-        do {
-            co_await coroutine::maybe_yield();
-            if (output_buf.empty()) {
-                output_buf = temporary_buffer<char>(OUTPUT_BUF_SIZE);
-            }
-            strm->next_out = (Bytef*) output_buf.get();
-            strm->avail_out = OUTPUT_BUF_SIZE;
-            int e = inflate(strm.get(), Z_NO_FLUSH);
-            size_t out_bytes = OUTPUT_BUF_SIZE - strm->avail_out;
-            if (out_bytes > 0) {
-                // If output_buf is nearly full, we save it as-is in ret. But
-                // if it only has little data, better copy to a small buffer.
-                if (out_bytes > OUTPUT_BUF_SIZE/2) {
-                    ret.push_back(std::move(output_buf).prefix(out_bytes));
-                    // output_buf is now empty. if this loop finds more input,
-                    // we'll allocate a new output buffer.
-                } else {
-                    ret.push_back(temporary_buffer<char>(output_buf.get(), out_bytes));
-                }
-                total_out_bytes += out_bytes;
-                if (total_out_bytes > length_limit) {
-                    throw api_error::payload_too_large(fmt::format("Request content length limit of {} bytes exceeded", length_limit));
-                }
-            }
-            if (e == Z_STREAM_END) {
-                // There may be more input after the first gzip stream - in
-                // either this input_buf or the next one. The additional input
-                // should be a second concatenated gzip. We need to allow that
-                // by resetting the gzip stream and continuing the input loop
-                // until there's no more input.
-                strm.reset();
-                if (strm->avail_in == 0) {
-                    complete_stream = true;
-                    break;
-                }
-            } else if (e != Z_OK && e != Z_BUF_ERROR) {
-                // DynamoDB returns an InternalServerError when given a bad
-                // gzip request body. See test test_broken_gzip_content
-                throw api_error::internal("Error during gzip decompression of request body");
-            }
-        } while (strm->avail_in > 0 || strm->avail_out == 0);
-    }
-    if (!complete_stream) {
-        // The gzip stream was not properly finished with Z_STREAM_END
-        throw api_error::internal("Truncated gzip in request body");
-    }
-    co_return ret;
-}
-
 future<executor::request_return_type> server::handle_api_request(std::unique_ptr<request> req) {
     _executor._stats.total_operations++;
     sstring target = req->get_header("X-Amz-Target");
     // target is DynamoDB API version followed by a dot '.' and operation type (e.g. CreateTable)
     auto dot = target.find('.');
     std::string_view op = (dot == sstring::npos) ? std::string_view() : std::string_view(target).substr(dot+1);
-    if (req->content_length > request_content_length_limit) {
-        // If we have a Content-Length header and know the request will be too
-        // long, we don't need to wait for read_entire_stream() below to
-        // discover it. And we definitely mustn't try to get_units() below for
-        // for such a size.
-        co_return api_error::payload_too_large(fmt::format("Request content length limit of {} bytes exceeded", request_content_length_limit));
-    }
     // JSON parsing can allocate up to roughly 2x the size of the raw
     // document, + a couple of bytes for maintenance.
-    // If the Content-Length of the request is not available, we assume
-    // the largest possible request (request_content_length_limit, i.e., 16 MB)
-    // and after reading the request we return_units() the excess.
-    size_t mem_estimate = (req->content_length ? req->content_length : request_content_length_limit) * 2 + 8000;
+    // TODO: consider the case where req->content_length is missing. Maybe
+    // we need to take the content_length_limit and return some of the units
+    // when we finish read_content_and_verify_signature?
+    size_t mem_estimate = req->content_length * 2 + 8000;
     auto units_fut = get_units(*_memory_limiter, mem_estimate);
     if (_memory_limiter->waiters()) {
         ++_executor._stats.requests_blocked_memory;
     }
     auto units = co_await std::move(units_fut);
     SCYLLA_ASSERT(req->content_stream);
-    chunked_content content = co_await read_entire_stream(*req->content_stream, request_content_length_limit);
-    // If the request had no Content-Length, we reserved too many units
-    // so need to return some
-    if (req->content_length == 0) {
-        size_t content_length = 0;
-        for (const auto& chunk : content) {
-            content_length += chunk.size();
-        }
-        size_t new_mem_estimate = content_length * 2 + 8000;
-        units.return_units(mem_estimate - new_mem_estimate);
-    }
+    chunked_content content = co_await util::read_entire_stream(*req->content_stream);
     auto username = co_await verify_signature(*req, content);
-    // If the request is compressed, uncompress it now, after we checked
-    // the signature (the signature is computed on the compressed content).
-    // We apply the request_content_length_limit again to the uncompressed
-    // content - we don't want to allow a tiny compressed request to
-    // expand to a huge uncompressed request.
-    sstring content_encoding = req->get_header("Content-Encoding");
-    if (content_encoding == "gzip") {
-        content = co_await ungzip(std::move(content), request_content_length_limit);
-    } else if (content_encoding == "deflate") {
-        content = co_await ungzip(std::move(content), request_content_length_limit, false);
-    } else if (!content_encoding.empty()) {
-        // DynamoDB returns a 500 error for unsupported Content-Encoding.
-        // I'm not sure if this is the best error code, but let's do it too.
-        // See the test test_garbage_content_encoding confirming this case.
-        co_return api_error::internal("Unsupported Content-Encoding");
-    }
-
     // As long as the system_clients_entry object is alive, this request will
     // be visible in the "system.clients" virtual table. When requested, this
     // entry will be formatted by server::ongoing_request::make_client_data().
-    auto user_agent_header = co_await _connection_options_keys_and_values.get_or_load(req->get_header("User-Agent"), [] (const client_options_cache_key_type&) {
-        return make_ready_future<options_cache_value_type>(options_cache_value_type{});
-    });
-
     auto system_clients_entry = _ongoing_requests.emplace(
-        req->get_client_address(), std::move(user_agent_header),
+        req->get_client_address(), req->get_header("User-Agent"),
         username, current_scheduling_group(),
         req->get_protocol_name() == "https");
 
     if (slogger.is_enabled(log_level::trace)) {
-        slogger.trace("Request: {} {} {}", op, truncated_content_view(content, _max_users_query_size_in_trace_output).as_view(), req->_headers);
+        std::string buf;
+        slogger.trace("Request: {} {} {}", op, truncated_content_view(content, buf), req->_headers);
     }
     auto callback_it = _callbacks.find(op);
     if (callback_it == _callbacks.end()) {
@@ -773,7 +515,7 @@ future<executor::request_return_type> server::handle_api_request(std::unique_ptr
     }
     co_await client_state.maybe_update_per_service_level_params();
 
-    tracing::trace_state_ptr trace_state = maybe_trace_query(client_state, username, op, content, _max_users_query_size_in_trace_output.get());
+    tracing::trace_state_ptr trace_state = maybe_trace_query(client_state, username, op, content);
     tracing::trace(trace_state, "{}", op);
 
     auto user = client_state.user();
@@ -793,7 +535,7 @@ future<executor::request_return_type> server::handle_api_request(std::unique_ptr
 void server::set_routes(routes& r) {
     api_handler* req_handler = new api_handler([this] (std::unique_ptr<request> req) mutable {
         return handle_api_request(std::move(req));
-    }, _proxy.data_dictionary().get_config());
+    });
 
     r.put(operation_type::POST, "/", req_handler);
     r.put(operation_type::GET, "/", new health_handler(_pending_requests));
@@ -824,7 +566,7 @@ server::server(executor& exec, service::storage_proxy& proxy, gms::gossiper& gos
         , _auth_service(auth_service)
         , _sl_controller(sl_controller)
         , _key_cache(1024, 1min, slogger)
-        , _max_users_query_size_in_trace_output(1024)
+        , _enforce_authorization(false)
         , _enabled_servers{}
         , _pending_requests("alternator::server::pending_requests")
         , _timeout_config(_proxy.data_dictionary().get_config())
@@ -904,39 +646,30 @@ server::server(executor& exec, service::storage_proxy& proxy, gms::gossiper& gos
     } {
 }
 
-future<> server::init(net::inet_address addr, std::optional<uint16_t> port, std::optional<uint16_t> https_port,
-        std::optional<uint16_t> port_proxy_protocol, std::optional<uint16_t> https_port_proxy_protocol,
-        std::optional<tls::credentials_builder> creds,
-        utils::updateable_value<bool> enforce_authorization, utils::updateable_value<bool> warn_authorization, utils::updateable_value<uint64_t> max_users_query_size_in_trace_output,
+future<> server::init(net::inet_address addr, std::optional<uint16_t> port, std::optional<uint16_t> https_port, std::optional<tls::credentials_builder> creds,
+        utils::updateable_value<bool> enforce_authorization, utils::updateable_value<bool> warn_authorization,
         semaphore* memory_limiter, utils::updateable_value<uint32_t> max_concurrent_requests) {
     _memory_limiter = memory_limiter;
     _enforce_authorization = std::move(enforce_authorization);
     _warn_authorization = std::move(warn_authorization);
     _max_concurrent_requests = std::move(max_concurrent_requests);
-    _max_users_query_size_in_trace_output = std::move(max_users_query_size_in_trace_output);
-    if (!port && !https_port && !port_proxy_protocol && !https_port_proxy_protocol) {
+    if (!port && !https_port) {
         return make_exception_future<>(std::runtime_error("Either regular port or TLS port"
                 " must be specified in order to init an alternator HTTP server instance"));
     }
-    return seastar::async([this, addr, port, https_port, port_proxy_protocol, https_port_proxy_protocol, creds] {
+    return seastar::async([this, addr, port, https_port, creds] {
         _executor.start().get();
 
-        if (port || port_proxy_protocol) {
+        if (port) {
             set_routes(_http_server._routes);
+            _http_server.set_content_length_limit(server::content_length_limit);
             _http_server.set_content_streaming(true);
-            if (port) {
-                _http_server.listen(socket_address{addr, *port}).get();
-            }
-            if (port_proxy_protocol) {
-                listen_options lo;
-                lo.reuse_address = true;
-                lo.proxy_protocol = true;
-                _http_server.listen(socket_address{addr, *port_proxy_protocol}, lo).get();
-            }
+            _http_server.listen(socket_address{addr, *port}).get();
             _enabled_servers.push_back(std::ref(_http_server));
         }
-        if (https_port || https_port_proxy_protocol) {
+        if (https_port) {
             set_routes(_https_server._routes);
+            _https_server.set_content_length_limit(server::content_length_limit);
             _https_server.set_content_streaming(true);
 
             if (this_shard_id() == 0) {
@@ -955,15 +688,7 @@ future<> server::init(net::inet_address addr, std::optional<uint16_t> port, std:
             } else {
                 _credentials = creds->build_server_credentials();
             }
-            if (https_port) {
-                _https_server.listen(socket_address{addr, *https_port}, _credentials).get();
-            }
-            if (https_port_proxy_protocol) {
-                listen_options lo;
-                lo.reuse_address = true;
-                lo.proxy_protocol = true;
-                _https_server.listen(socket_address{addr, *https_port_proxy_protocol}, lo, _credentials).get();
-            }
+            _https_server.listen(socket_address{addr, *https_port}, _credentials).get();
             _enabled_servers.push_back(std::ref(_https_server));
         }
     });
@@ -1036,15 +761,16 @@ client_data server::ongoing_request::make_client_data() const {
     // and keep "driver_version" unset.
     cd.driver_name = _user_agent;
     // Leave "protocol_version" unset, it has no meaning in Alternator.
-    // Leave "hostname", "ssl_protocol" and "ssl_cipher_suite" unset for Alternator.
-    // Note: CQL sets ssl_protocol and ssl_cipher_suite via generic_server::connection base class.
+    // Leave "hostname", "ssl_protocol" and "ssl_cipher_suite" unset.
+    // As reported in issue #9216, we never set these fields in CQL
+    // either (see cql_server::connection::make_client_data()).
     return cd;
 }
 
-future<utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>>> server::get_client_data() {
-    utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>> ret;
+future<utils::chunked_vector<client_data>> server::get_client_data() {
+    utils::chunked_vector<client_data> ret;
     co_await _ongoing_requests.for_each_gently([&ret] (const ongoing_request& r) {
-        ret.emplace_back(make_foreign(std::make_unique<client_data>(r.make_client_data())));
+        ret.emplace_back(r.make_client_data());
     });
     co_return ret;
 }

alternator/server.hh

@@ -28,11 +28,7 @@ namespace alternator {
 using chunked_content = rjson::chunked_content;
 
 class server : public peering_sharded_service<server> {
-    // The maximum size of a request body that Alternator will accept,
-    // in bytes. This is a safety measure to prevent Alternator from
-    // running out of memory when a client sends a very large request.
-    // DynamoDB also has the same limit set to 16 MB.
-    static constexpr size_t request_content_length_limit = 16*MB;
+    static constexpr size_t content_length_limit = 16*MB;
     using alternator_callback = std::function<future<executor::request_return_type>(executor&, executor::client_state&,
             tracing::trace_state_ptr, service_permit, rjson::value, std::unique_ptr<http::request>)>;
     using alternator_callbacks_map = std::unordered_map<std::string_view, alternator_callback>;
@@ -48,14 +44,12 @@ class server : public peering_sharded_service<server> {
     key_cache _key_cache;
     utils::updateable_value<bool> _enforce_authorization;
     utils::updateable_value<bool> _warn_authorization;
-    utils::updateable_value<uint64_t> _max_users_query_size_in_trace_output;
     utils::small_vector<std::reference_wrapper<seastar::httpd::http_server>, 2> _enabled_servers;
     named_gate _pending_requests;
     // In some places we will need a CQL updateable_timeout_config object even
     // though it isn't really relevant for Alternator which defines its own
     // timeouts separately. We can create this object only once.
     updateable_timeout_config _timeout_config;
-    client_options_cache_type _connection_options_keys_and_values;
 
     alternator_callbacks_map _callbacks;
 
@@ -89,7 +83,7 @@ class server : public peering_sharded_service<server> {
     // is called when reading the "system.clients" virtual table.
     struct ongoing_request {
         socket_address _client_address;
-        client_options_cache_entry_type _user_agent;
+        sstring _user_agent;
         sstring _username;
         scheduling_group _scheduling_group;
         bool _is_https;
@@ -100,17 +94,15 @@ class server : public peering_sharded_service<server> {
 public:
     server(executor& executor, service::storage_proxy& proxy, gms::gossiper& gossiper, auth::service& service, qos::service_level_controller& sl_controller);
 
-    future<> init(net::inet_address addr, std::optional<uint16_t> port, std::optional<uint16_t> https_port,
-            std::optional<uint16_t> port_proxy_protocol, std::optional<uint16_t> https_port_proxy_protocol,
-            std::optional<tls::credentials_builder> creds,
-            utils::updateable_value<bool> enforce_authorization, utils::updateable_value<bool> warn_authorization, utils::updateable_value<uint64_t> max_users_query_size_in_trace_output,
+    future<> init(net::inet_address addr, std::optional<uint16_t> port, std::optional<uint16_t> https_port, std::optional<tls::credentials_builder> creds,
+            utils::updateable_value<bool> enforce_authorization, utils::updateable_value<bool> warn_authorization,
             semaphore* memory_limiter, utils::updateable_value<uint32_t> max_concurrent_requests);
     future<> stop();
     // get_client_data() is called (on each shard separately) when the virtual
     // table "system.clients" is read. It is expected to generate a list of
     // clients connected to this server (on this shard). This function is
     // called by alternator::controller::get_client_data().
-    future<utils::chunked_vector<foreign_ptr<std::unique_ptr<client_data>>>> get_client_data();
+    future<utils::chunked_vector<client_data>> get_client_data();
 private:
     void set_routes(seastar::httpd::routes& r);
     // If verification succeeds, returns the authenticated user's username

alternator/stats.cc

@@ -154,18 +154,6 @@ static void register_metrics_with_optional_table(seastar::metrics::metric_groups
                     [&stats]{ return estimated_histogram_to_metrics(stats.api_operations.batch_get_item_histogram);})(op("BatchGetItem")).aggregate({seastar::metrics::shard_label}).set_skip_when_empty(),
             seastar::metrics::make_histogram("batch_item_count_histogram", seastar::metrics::description("Histogram of the number of items in a batch request"), labels,
                     [&stats]{ return estimated_histogram_to_metrics(stats.api_operations.batch_write_item_histogram);})(op("BatchWriteItem")).aggregate({seastar::metrics::shard_label}).set_skip_when_empty(),
-            seastar::metrics::make_histogram("operation_size_kb", seastar::metrics::description("Histogram of item sizes involved in a request"), labels,
-                    [&stats]{ return estimated_histogram_to_metrics(stats.operation_sizes.get_item_op_size_kb);})(op("GetItem")).aggregate({seastar::metrics::shard_label}).set_skip_when_empty(),
-            seastar::metrics::make_histogram("operation_size_kb", seastar::metrics::description("Histogram of item sizes involved in a request"), labels,
-                    [&stats]{ return estimated_histogram_to_metrics(stats.operation_sizes.put_item_op_size_kb);})(op("PutItem")).aggregate({seastar::metrics::shard_label}).set_skip_when_empty(),
-            seastar::metrics::make_histogram("operation_size_kb", seastar::metrics::description("Histogram of item sizes involved in a request"), labels,
-                    [&stats]{ return estimated_histogram_to_metrics(stats.operation_sizes.delete_item_op_size_kb);})(op("DeleteItem")).aggregate({seastar::metrics::shard_label}).set_skip_when_empty(),
-            seastar::metrics::make_histogram("operation_size_kb", seastar::metrics::description("Histogram of item sizes involved in a request"), labels,
-                    [&stats]{ return estimated_histogram_to_metrics(stats.operation_sizes.update_item_op_size_kb);})(op("UpdateItem")).aggregate({seastar::metrics::shard_label}).set_skip_when_empty(),
-            seastar::metrics::make_histogram("operation_size_kb", seastar::metrics::description("Histogram of item sizes involved in a request"), labels,
-                    [&stats]{ return estimated_histogram_to_metrics(stats.operation_sizes.batch_get_item_op_size_kb);})(op("BatchGetItem")).aggregate({seastar::metrics::shard_label}).set_skip_when_empty(),
-            seastar::metrics::make_histogram("operation_size_kb", seastar::metrics::description("Histogram of item sizes involved in a request"), labels,
-                    [&stats]{ return estimated_histogram_to_metrics(stats.operation_sizes.batch_write_item_op_size_kb);})(op("BatchWriteItem")).aggregate({seastar::metrics::shard_label}).set_skip_when_empty(),
     });
 
     seastar::metrics::label expression_label("expression");

alternator/stats.hh

@@ -79,32 +79,6 @@ public:
         utils::estimated_histogram batch_get_item_histogram{22}; // a histogram that covers the range 1 - 100
         utils::estimated_histogram batch_write_item_histogram{22}; // a histogram that covers the range 1 - 100
     } api_operations;
-    // Operation size metrics
-    struct {
-        // Item size statistics collected per table and aggregated per node.
-        // Each histogram covers the range 0 - 446. Resolves #25143.
-        // A size is the retrieved item's size.
-        utils::estimated_histogram get_item_op_size_kb{30};
-        // A size is the maximum of the new item's size and the old item's size.
-        utils::estimated_histogram put_item_op_size_kb{30};
-        // A size is the deleted item's size. If the deleted item's size is
-        // unknown (i.e. read-before-write wasn't necessary and it wasn't
-        // forced by a configuration option), it won't be recorded on the
-        // histogram.
-        utils::estimated_histogram delete_item_op_size_kb{30};
-        // A size is the maximum of existing item's size and the estimated size
-        // of the update. This will be changed to the maximum of the existing item's
-        // size and the new item's size in a subsequent PR.
-        utils::estimated_histogram update_item_op_size_kb{30};
-
-        // A size is the sum of the sizes of all items per table. This means
-        // that a single BatchGetItem / BatchWriteItem updates the histogram
-        // for each table that it has items in.
-        // The sizes are the retrieved items' sizes grouped per table.
-        utils::estimated_histogram batch_get_item_op_size_kb{30};
-        // The sizes are the the written items' sizes grouped per table.
-        utils::estimated_histogram batch_write_item_op_size_kb{30};
-    } operation_sizes;
     // Count of authentication and authorization failures, counted if either
     // alternator_enforce_authorization or alternator_warn_authorization are
     // set to true. If both are false, no authentication or authorization
@@ -163,8 +137,4 @@ struct table_stats {
 };
 void register_metrics(seastar::metrics::metric_groups& metrics, const stats& stats);
 
-inline uint64_t bytes_to_kb_ceil(uint64_t bytes) {
-    return (bytes + 1023) / 1024;
-}
-
 }

alternator/streams.cc

@@ -13,6 +13,7 @@
 
 #include <seastar/json/formatter.hh>
 
+#include "auth/permission.hh"
 #include "db/config.hh"
 
 #include "cdc/log.hh"
@@ -126,7 +127,7 @@ public:
     }
 };
 
-} // namespace alternator
+}
 
 template<typename ValueType>
 struct rapidjson::internal::TypeHelper<ValueType, alternator::stream_arn>
@@ -296,7 +297,7 @@ sequence_number::sequence_number(std::string_view v)
     }())
 {}
 
-} // namespace alternator
+}
 
 template<typename ValueType>
 struct rapidjson::internal::TypeHelper<ValueType, alternator::shard_id>
@@ -356,7 +357,7 @@ static stream_view_type cdc_options_to_steam_view_type(const cdc::options& opts)
     return type;
 }
 
-} // namespace alternator
+}
 
 template<typename ValueType>
 struct rapidjson::internal::TypeHelper<ValueType, alternator::stream_view_type>
@@ -475,10 +476,10 @@ future<executor::request_return_type> executor::describe_stream(client_state& cl
         } else {
             status = "ENABLED";
         }
-    }
+    } 
 
     auto ttl = std::chrono::seconds(opts.ttl());
-
+    
     rjson::add(stream_desc, "StreamStatus", rjson::from_string(status));
 
     stream_view_type type = cdc_options_to_steam_view_type(opts);
@@ -714,7 +715,7 @@ future<executor::request_return_type> executor::get_shard_iterator(client_state&
 
     auto type = rjson::get<shard_iterator_type>(request, "ShardIteratorType");
     auto seq_num = rjson::get_opt<sequence_number>(request, "SequenceNumber");
-
+    
     if (type < shard_iterator_type::TRIM_HORIZON && !seq_num) {
         throw api_error::validation("Missing required parameter \"SequenceNumber\"");
     }
@@ -724,7 +725,7 @@ future<executor::request_return_type> executor::get_shard_iterator(client_state&
 
     auto stream_arn = rjson::get<alternator::stream_arn>(request, "StreamArn");
     auto db = _proxy.data_dictionary();
-
+    
     schema_ptr schema = nullptr;
     std::optional<shard_id> sid;
 
@@ -789,7 +790,7 @@ struct event_id {
         return os;
     }
 };
-} // namespace alternator
+}
 
 template<typename ValueType>
 struct rapidjson::internal::TypeHelper<ValueType, alternator::event_id>
@@ -940,7 +941,7 @@ future<executor::request_return_type> executor::get_records(client_state& client
                 rjson::add(record, "awsRegion", rjson::from_string(dc_name));
                 rjson::add(record, "eventID", event_id(iter.shard.id, *timestamp));
                 rjson::add(record, "eventSource", "scylladb:alternator");
-                rjson::add(record, "eventVersion", "1.1");
+                rjson::add(record, "eventVersion", "1.0");
                 rjson::push_back(records, std::move(record));
                 record = rjson::empty_object();
                 --limit;
@@ -999,16 +1000,6 @@ future<executor::request_return_type> executor::get_records(client_state& client
             case cdc::operation::insert:
                 rjson::add(record, "eventName", "INSERT");
                 break;
-            case cdc::operation::service_row_delete:
-            case cdc::operation::service_partition_delete:
-            {
-                auto user_identity = rjson::empty_object();
-                rjson::add(user_identity, "Type", "Service");
-                rjson::add(user_identity, "PrincipalId", "dynamodb.amazonaws.com");
-                rjson::add(record, "userIdentity", std::move(user_identity));
-                rjson::add(record, "eventName", "REMOVE");
-                break;
-            }
             default:
                 rjson::add(record, "eventName", "REMOVE");
                 break;
@@ -1073,7 +1064,9 @@ bool executor::add_stream_options(const rjson::value& stream_specification, sche
     }
 
     if (stream_enabled->GetBool()) {
-        if (!sp.features().alternator_streams) {
+        auto db = sp.data_dictionary();
+
+        if (!db.features().alternator_streams) {
             throw api_error::validation("StreamSpecification: alternator streams feature not enabled in cluster.");
         }
 
@@ -1132,4 +1125,4 @@ void executor::supplement_table_stream_info(rjson::value& descr, const schema& s
     }
 }
 
-} // namespace alternator
+}

alternator/ttl.cc

@@ -17,7 +17,6 @@
 #include <seastar/core/lowres_clock.hh>
 #include <seastar/coroutine/maybe_yield.hh>
 
-#include "cdc/log.hh"
 #include "exceptions/exceptions.hh"
 #include "gms/gossiper.hh"
 #include "gms/inet_address.hh"
@@ -68,7 +67,7 @@ extern const sstring TTL_TAG_KEY;
 
 future<executor::request_return_type> executor::update_time_to_live(client_state& client_state, service_permit permit, rjson::value request) {
     _stats.api_operations.update_time_to_live++;
-    if (!_proxy.features().alternator_ttl) {
+    if (!_proxy.data_dictionary().features().alternator_ttl) {
         co_return api_error::unknown_operation("UpdateTimeToLive not yet supported. Experimental support is available if the 'alternator-ttl' experimental feature is enabled on all nodes.");
     }
 
@@ -93,7 +92,7 @@ future<executor::request_return_type> executor::update_time_to_live(client_state
     if (v->GetStringLength() < 1 || v->GetStringLength() > 255) {
         co_return api_error::validation("The length of AttributeName must be between 1 and 255");
     }
-    sstring attribute_name = rjson::to_sstring(*v);
+    sstring attribute_name(v->GetString(), v->GetStringLength());
 
     co_await verify_permission(_enforce_authorization, _warn_authorization, client_state, schema, auth::permission::ALTER, _stats);
     co_await db::modify_tags(_mm, schema->ks_name(), schema->cf_name(), [&](std::map<sstring, sstring>& tags_map) {
@@ -293,12 +292,7 @@ static future<> expire_item(service::storage_proxy& proxy,
         db::consistency_level::LOCAL_QUORUM,
         executor::default_timeout(), // FIXME - which timeout?
         qs.get_trace_state(), qs.get_permit(),
-        db::allow_per_partition_rate_limit::no,
-        false,
-        cdc::per_request_options{
-            .is_system_originated = true,
-        }
-    );
+        db::allow_per_partition_rate_limit::no);
 }
 
 static size_t random_offset(size_t min, size_t max) {

api/CMakeLists.txt

@@ -31,7 +31,6 @@ set(swagger_files
   api-doc/column_family.json
   api-doc/commitlog.json
   api-doc/compaction_manager.json
-  api-doc/client_routes.json
   api-doc/config.json
   api-doc/cql_server_test.json
   api-doc/endpoint_snitch_info.json
@@ -69,7 +68,6 @@ target_sources(api
   PRIVATE
     api.cc
     cache_service.cc
-    client_routes.cc
     collectd.cc
     column_family.cc
     commitlog.cc
@@ -108,8 +106,5 @@ target_link_libraries(api
     wasmtime_bindings
     absl::headers)
 
-if (Scylla_USE_PRECOMPILED_HEADER_USE)
-  target_precompile_headers(api REUSE_FROM scylla-precompiled-header)
-endif()
 check_headers(check-headers api
   GLOB_RECURSE ${CMAKE_CURRENT_SOURCE_DIR}/*.hh)

api/api-doc/client_routes.def.json

@@ -1,23 +0,0 @@
-    , "client_routes_entry": {
-        "id": "client_routes_entry",
-        "summary": "An entry storing client routes",
-        "properties": {
-            "connection_id": {"type": "string"},
-            "host_id": {"type": "string", "format": "uuid"},
-            "address": {"type": "string"},
-            "port": {"type": "integer"},
-            "tls_port": {"type": "integer"},
-            "alternator_port": {"type": "integer"},
-            "alternator_https_port": {"type": "integer"}
-        },
-        "required": ["connection_id", "host_id", "address"]
-    }
-    , "client_routes_key": {
-        "id": "client_routes_key",
-        "summary": "A key of client_routes_entry",
-        "properties": {
-            "connection_id": {"type": "string"},
-            "host_id": {"type": "string", "format": "uuid"}
-        }
-    }
-

api/api-doc/client_routes.json

@@ -1,74 +0,0 @@
-    , "/v2/client-routes":{
-        "get": {
-            "description":"List all client route entries",
-            "operationId":"get_client_routes",
-            "tags":["client_routes"],
-            "produces":[
-                "application/json"
-            ],
-            "parameters":[],
-            "responses":{
-                "200":{
-                    "schema":{
-                        "type":"array",
-                        "items":{ "$ref":"#/definitions/client_routes_entry" }
-                    }
-                },
-                "default":{
-                    "description":"unexpected error",
-                    "schema":{"$ref":"#/definitions/ErrorModel"}
-                }
-            }
-        },
-        "post": {
-            "description":"Upsert one or more client route entries",
-            "operationId":"set_client_routes",
-            "tags":["client_routes"],
-            "parameters":[
-                {
-                    "name":"body",
-                    "in":"body",
-                    "required":true,
-                    "schema":{
-                        "type":"array",
-                        "items":{ "$ref":"#/definitions/client_routes_entry" }
-                    }
-                }
-            ],
-            "responses":{
-                "200":{ "description": "OK" },
-                "default":{
-                    "description":"unexpected error",
-                    "schema":{ "$ref":"#/definitions/ErrorModel" }
-                }
-            }
-        },
-        "delete": {
-            "description":"Delete one or more client route entries",
-            "operationId":"delete_client_routes",
-            "tags":["client_routes"],
-            "parameters":[
-                {
-                    "name":"body",
-                    "in":"body",
-                    "required":true,
-                    "schema":{
-                        "type":"array",
-                        "items":{ "$ref":"#/definitions/client_routes_key" }
-                    }
-                }
-            ],
-            "responses":{
-                "200":{
-                    "description": "OK"
-                },
-                "default":{
-                    "description":"unexpected error",
-                    "schema":{
-                        "$ref":"#/definitions/ErrorModel"
-                    }
-                }
-            }
-        }
-    }
-

api/api-doc/storage_service.json

@@ -220,25 +220,6 @@
             }
          ]
       },
-      {
-         "path":"/storage_service/nodes/excluded",
-         "operations":[
-            {
-               "method":"GET",
-               "summary":"Retrieve host ids of nodes which are marked as excluded",
-               "type":"array",
-               "items":{
-                  "type":"string"
-               },
-               "nickname":"get_excluded_nodes",
-               "produces":[
-                  "application/json"
-               ],
-               "parameters":[
-               ]
-            }
-         ]
-      },
       {
          "path":"/storage_service/nodes/joining",
          "operations":[
@@ -613,50 +594,6 @@
             }
          ]
       },
-      {
-         "path": "/storage_service/natural_endpoints/v2/{keyspace}",
-         "operations": [
-            {
-               "method": "GET",
-               "summary":"This method returns the N endpoints that are responsible for storing the specified key i.e for replication. the endpoint responsible for this key",
-               "type": "array",
-               "items": {
-                  "type": "string"
-               },
-               "nickname": "get_natural_endpoints_v2",
-               "produces": [
-                  "application/json"
-               ],
-               "parameters": [
-                  {
-                     "name": "keyspace",
-                     "description": "The keyspace to query about.",
-                     "required": true,
-                     "allowMultiple": false,
-                     "type": "string",
-                     "paramType": "path"
-                  },
-                  {
-                     "name": "cf",
-                     "description": "Column family name.",
-                     "required": true,
-                     "allowMultiple": false,
-                     "type": "string",
-                     "paramType": "query"
-                  },
-                  {
-                     "name": "key_component",
-                     "description": "Each component of the key for which we need to find the endpoint (e.g. ?key_component=part1&key_component=part2).",
-                     "required": true,
-                     "allowMultiple": true,
-                     "type": "string",
-                     "paramType": "query"
-                  }
-               ]
-            }
-         ]
-      },
-
       {
          "path":"/storage_service/cdc_streams_check_and_repair",
          "operations":[
@@ -1195,14 +1132,6 @@
                      "allowMultiple":false,
                      "type":"string",
                      "paramType":"query"
-                  },
-                  {
-                     "name": "drop_unfixable_sstables",
-                     "description": "When set to true, drop unfixable sstables. Applies only to scrub mode SEGREGATE.",
-                     "required":false,
-                     "allowMultiple":false,
-                     "type":"boolean",
-                     "paramType":"query"
                   }
                ]
             }
@@ -1622,30 +1551,6 @@
             }
          ]
       },
-      {
-         "path":"/storage_service/exclude_node",
-         "operations":[
-            {
-               "method":"POST",
-               "summary":"Marks the node as permanently down (excluded).",
-               "type":"void",
-               "nickname":"exclude_node",
-               "produces":[
-                  "application/json"
-               ],
-               "parameters":[
-                  {
-                     "name":"hosts",
-                     "description":"Comma-separated list of host ids to exclude",
-                     "required":true,
-                     "allowMultiple":false,
-                     "type":"string",
-                     "paramType":"query"
-                  }
-               ]
-            }
-         ]
-      },
       {
          "path":"/storage_service/removal_status",
          "operations":[

api/api.cc

@@ -37,7 +37,6 @@
 #include "raft.hh"
 #include "gms/gossip_address_map.hh"
 #include "service_levels.hh"
-#include "client_routes.hh"
 
 logging::logger apilog("api");
 
@@ -68,11 +67,9 @@ future<> set_server_init(http_context& ctx) {
         rb02->set_api_doc(r);
         rb02->register_api_file(r, "swagger20_header");
         rb02->register_api_file(r, "metrics");
-        rb02->register_api_file(r, "client_routes");
         rb->register_function(r, "system",
                 "The system related API");
         rb02->add_definitions_file(r, "metrics");
-        rb02->add_definitions_file(r, "client_routes");
         set_system(ctx, r);
         rb->register_function(r, "error_injection",
             "The error injection API");
@@ -132,16 +129,6 @@ future<> unset_server_storage_service(http_context& ctx) {
     return ctx.http_server.set_routes([&ctx] (routes& r) { unset_storage_service(ctx, r); });
 }
 
-future<> set_server_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr) {
-    return ctx.http_server.set_routes([&ctx, &cr] (routes& r) {
-        set_client_routes(ctx, r, cr);
-    });
-}
-
-future<> unset_server_client_routes(http_context& ctx) {
-    return ctx.http_server.set_routes([&ctx] (routes& r) { unset_client_routes(ctx, r); });
-}
-
 future<> set_load_meter(http_context& ctx, service::load_meter& lm) {
     return ctx.http_server.set_routes([&ctx, &lm] (routes& r) { set_load_meter(ctx, r, lm); });
 }
@@ -229,10 +216,10 @@ future<> unset_server_gossip(http_context& ctx) {
     });
 }
 
-future<> set_server_column_family(http_context& ctx, sharded<replica::database>& db) {
+future<> set_server_column_family(http_context& ctx, sharded<replica::database>& db, sharded<db::system_keyspace>& sys_ks) {
     co_await register_api(ctx, "column_family",
-                "The column family API", [&db] (http_context& ctx, routes& r) {
-                    set_column_family(ctx, r, db);
+                "The column family API", [&db, &sys_ks] (http_context& ctx, routes& r) {
+                    set_column_family(ctx, r, db, sys_ks);
                 });
     co_await register_api(ctx, "cache_service",
             "The cache service API", [&db] (http_context& ctx, routes& r) {

api/api_init.hh

@@ -29,7 +29,6 @@ class storage_proxy;
 class storage_service;
 class raft_group0_client;
 class raft_group_registry;
-class client_routes_service;
 
 } // namespace service
 
@@ -59,6 +58,7 @@ class sstables_format_selector;
 namespace view {
 class view_builder;
 }
+class system_keyspace;
 }
 namespace netw { class messaging_service; }
 class repair_service;
@@ -100,8 +100,6 @@ future<> set_server_snitch(http_context& ctx, sharded<locator::snitch_ptr>& snit
 future<> unset_server_snitch(http_context& ctx);
 future<> set_server_storage_service(http_context& ctx, sharded<service::storage_service>& ss, service::raft_group0_client&);
 future<> unset_server_storage_service(http_context& ctx);
-future<> set_server_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr);
-future<> unset_server_client_routes(http_context& ctx);
 future<> set_server_sstables_loader(http_context& ctx, sharded<sstables_loader>& sst_loader);
 future<> unset_server_sstables_loader(http_context& ctx);
 future<> set_server_view_builder(http_context& ctx, sharded<db::view::view_builder>& vb, sharded<gms::gossiper>& g);
@@ -120,7 +118,7 @@ future<> set_server_token_metadata(http_context& ctx, sharded<locator::shared_to
 future<> unset_server_token_metadata(http_context& ctx);
 future<> set_server_gossip(http_context& ctx, sharded<gms::gossiper>& g);
 future<> unset_server_gossip(http_context& ctx);
-future<> set_server_column_family(http_context& ctx, sharded<replica::database>& db);
+future<> set_server_column_family(http_context& ctx, sharded<replica::database>& db, sharded<db::system_keyspace>& sys_ks);
 future<> unset_server_column_family(http_context& ctx);
 future<> set_server_messaging_service(http_context& ctx, sharded<netw::messaging_service>& ms);
 future<> unset_server_messaging_service(http_context& ctx);

api/client_routes.cc

@@ -1,176 +0,0 @@
-/*
- * Copyright (C) 2025-present ScyllaDB
- *
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
- #include <seastar/http/short_streams.hh>
-
-#include "client_routes.hh"
-#include "api/api.hh"
-#include "service/storage_service.hh"
-#include "service/client_routes.hh"
-#include "utils/rjson.hh"
-
-
-#include "api/api-doc/client_routes.json.hh"
-
-using namespace seastar::httpd;
-using namespace std::chrono_literals;
-using namespace json;
-
-extern logging::logger apilog;
-
-namespace api {
-
-static void validate_client_routes_endpoint(sharded<service::client_routes_service>& cr, sstring endpoint_name) {
-    if (!cr.local().get_feature_service().client_routes) {
-        apilog.warn("{}: called before the cluster feature was enabled", endpoint_name);
-        throw std::runtime_error(fmt::format("{} requires all nodes to support the CLIENT_ROUTES cluster feature", endpoint_name));
-    }
-}
-
-static sstring parse_string(const char* name, rapidjson::Value const& v) {
-    const auto it = v.FindMember(name);
-    if (it == v.MemberEnd()) {
-        throw bad_param_exception(fmt::format("Missing '{}'", name));
-    }
-    if (!it->value.IsString()) {
-        throw bad_param_exception(fmt::format("'{}' must be a string", name));
-    }
-    return {it->value.GetString(), it->value.GetStringLength()};
-}
-
-static std::optional<uint32_t> parse_port(const char* name, rapidjson::Value const& v) {
-    const auto it = v.FindMember(name);
-    if (it == v.MemberEnd()) {
-        return std::nullopt;
-    }
-    if (!it->value.IsInt()) {
-        throw bad_param_exception(fmt::format("'{}' must be an integer", name));
-    }
-    auto port = it->value.GetInt();
-    if (port < 1 || port > 65535) {
-        throw bad_param_exception(fmt::format("'{}' value={} is outside the allowed port range", name, port));
-    }
-    return port;
-}
-
-static std::vector<service::client_routes_service::client_route_entry> parse_set_client_array(const rapidjson::Document& root) {
-    if (!root.IsArray()) {
-        throw bad_param_exception("Body must be a JSON array");
-    }
-
-    std::vector<service::client_routes_service::client_route_entry> v;
-    v.reserve(root.GetArray().Size());
-    for (const auto& element : root.GetArray()) {
-        if (!element.IsObject()) { throw bad_param_exception("Each element must be object"); }
-
-        const auto port = parse_port("port", element);
-        const auto tls_port = parse_port("tls_port", element);
-        const auto alternator_port = parse_port("alternator_port", element);
-        const auto alternator_https_port = parse_port("alternator_https_port", element);
-
-        if (!port.has_value() && !tls_port.has_value() && !alternator_port.has_value() && !alternator_https_port.has_value()) {
-            throw bad_param_exception("At least one port field ('port', 'tls_port', 'alternator_port', 'alternator_https_port') must be specified");
-        }
-
-        v.emplace_back(
-            parse_string("connection_id", element),
-            utils::UUID{parse_string("host_id", element)},
-            parse_string("address", element),
-            port,
-            tls_port,
-            alternator_port,
-            alternator_https_port
-        );
-    }
-
-    return v;
-}
-
-static
-future<json::json_return_type>
-rest_set_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr, std::unique_ptr<http::request> req) {
-    validate_client_routes_endpoint(cr, "rest_set_client_routes");
-
-    rapidjson::Document root;
-    auto content = co_await util::read_entire_stream_contiguous(*req->content_stream);
-    root.Parse(content.c_str());
-
-    co_await cr.local().set_client_routes(parse_set_client_array(root));
-    co_return seastar::json::json_void();
-}
-
-static std::vector<service::client_routes_service::client_route_key> parse_delete_client_array(const rapidjson::Document& root) {
-    if (!root.IsArray()) {
-        throw bad_param_exception("Body must be a JSON array");
-    }
-
-    std::vector<service::client_routes_service::client_route_key> v;
-    v.reserve(root.GetArray().Size());
-    for (const auto& element : root.GetArray()) {
-        v.emplace_back(
-            parse_string("connection_id", element),
-            utils::UUID{parse_string("host_id", element)}
-        );
-    }
-
-    return v;
-}
-
-static
-future<json::json_return_type>
-rest_delete_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr, std::unique_ptr<http::request> req) {
-    validate_client_routes_endpoint(cr, "delete_client_routes");
-
-    rapidjson::Document root;
-    auto content = co_await util::read_entire_stream_contiguous(*req->content_stream);
-    root.Parse(content.c_str());
-
-    co_await cr.local().delete_client_routes(parse_delete_client_array(root));
-    co_return seastar::json::json_void();
-}
-
-static
-future<json::json_return_type>
-rest_get_client_routes(http_context& ctx, sharded<service::client_routes_service>& cr, std::unique_ptr<http::request> req) {
-    validate_client_routes_endpoint(cr, "get_client_routes");
-
-    co_return co_await cr.invoke_on(0, [] (service::client_routes_service& cr) -> future<json::json_return_type> {
-        co_return json::json_return_type(stream_range_as_array(co_await cr.get_client_routes(), [](const service::client_routes_service::client_route_entry & entry) {
-            seastar::httpd::client_routes_json::client_routes_entry obj;
-            obj.connection_id = entry.connection_id;
-            obj.host_id = fmt::to_string(entry.host_id);
-            obj.address = entry.address;
-            if (entry.port.has_value()) { obj.port = entry.port.value(); }
-            if (entry.tls_port.has_value()) { obj.tls_port = entry.tls_port.value(); }
-            if (entry.alternator_port.has_value()) { obj.alternator_port = entry.alternator_port.value(); }
-            if (entry.alternator_https_port.has_value()) { obj.alternator_https_port = entry.alternator_https_port.value(); }
-            return obj;
-        }));
-    });
-}
-
-void set_client_routes(http_context& ctx, routes& r, sharded<service::client_routes_service>& cr) {
-    seastar::httpd::client_routes_json::set_client_routes.set(r, [&ctx, &cr] (std::unique_ptr<seastar::http::request> req) {
-        return rest_set_client_routes(ctx, cr, std::move(req));
-    });
-    seastar::httpd::client_routes_json::delete_client_routes.set(r, [&ctx, &cr] (std::unique_ptr<seastar::http::request> req) {
-        return rest_delete_client_routes(ctx, cr, std::move(req));
-    });
-    seastar::httpd::client_routes_json::get_client_routes.set(r, [&ctx, &cr] (std::unique_ptr<seastar::http::request> req) {
-        return rest_get_client_routes(ctx, cr, std::move(req));
-    });
-}
-
-void unset_client_routes(http_context& ctx, routes& r) {
-    seastar::httpd::client_routes_json::set_client_routes.unset(r);
-    seastar::httpd::client_routes_json::delete_client_routes.unset(r);
-    seastar::httpd::client_routes_json::get_client_routes.unset(r);
-}
-
-}

api/client_routes.hh

@@ -1,20 +0,0 @@
-/*
- * Copyright (C) 2025-present ScyllaDB
- *
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-#pragma once
-
-#include <seastar/core/sharded.hh>
-#include <seastar/json/json_elements.hh>
-#include "api/api_init.hh"
-
-namespace api {
-
-void set_client_routes(http_context& ctx, httpd::routes& r, sharded<service::client_routes_service>& cr);
-void unset_client_routes(http_context& ctx, httpd::routes& r);
-
-}

api/column_family.cc

@@ -18,6 +18,7 @@
 #include "utils/assert.hh"
 #include "utils/estimated_histogram.hh"
 #include <algorithm>
+#include "db/system_keyspace.hh"
 #include "db/data_listeners.hh"
 #include "storage_service.hh"
 #include "compaction/compaction_manager.hh"
@@ -66,13 +67,6 @@ static future<json::json_return_type>  get_cf_stats(sharded<replica::database>&
     }, std::plus<int64_t>());
 }
 
-static future<json::json_return_type>  get_cf_stats(sharded<replica::database>& db,
-        std::function<int64_t(const replica::column_family_stats&)> f) {
-    return map_reduce_cf(db, int64_t(0), [f](const replica::column_family& cf) {
-        return f(cf.get_stats());
-    }, std::plus<int64_t>());
-}
-
 static future<json::json_return_type> for_tables_on_all_shards(sharded<replica::database>& db, std::vector<table_info> tables, std::function<future<>(replica::table&)> set) {
     return do_with(std::move(tables), [&db, set] (const std::vector<table_info>& tables) {
         return db.invoke_on_all([&tables, set] (replica::database& db) {
@@ -342,7 +336,7 @@ uint64_t accumulate_on_active_memtables(replica::table& t, noncopyable_function<
     return ret;
 }
 
-void set_column_family(http_context& ctx, routes& r, sharded<replica::database>& db) {
+void set_column_family(http_context& ctx, routes& r, sharded<replica::database>& db, sharded<db::system_keyspace>& sys_ks) {
     cf::get_column_family_name.set(r, [&db] (const_req req){
         std::vector<sstring> res;
         const replica::database::tables_metadata& meta = db.local().get_tables_metadata();
@@ -943,6 +937,30 @@ void set_column_family(http_context& ctx, routes& r, sharded<replica::database>&
         return set_tables_tombstone_gc(db, std::move(tables), false);
     });
 
+    cf::get_built_indexes.set(r, [&db, &sys_ks](std::unique_ptr<http::request> req) {
+        auto [ks, cf_name] = parse_fully_qualified_cf_name(req->get_path_param("name"));
+        // Use of load_built_views() as filtering table should be in sync with
+        // built_indexes_virtual_reader filtering with BUILT_VIEWS table
+        return sys_ks.local().load_built_views().then([ks, cf_name, &db](const std::vector<db::system_keyspace::view_name>& vb) mutable {
+            std::set<sstring> vp;
+            for (auto b : vb) {
+                if (b.first == ks) {
+                    vp.insert(b.second);
+                }
+            }
+            std::vector<sstring> res;
+            auto uuid = validate_table(db.local(), ks, cf_name);
+            replica::column_family& cf = db.local().find_column_family(uuid);
+            res.reserve(cf.get_index_manager().list_indexes().size());
+            for (auto&& i : cf.get_index_manager().list_indexes()) {
+                if (vp.contains(secondary_index::index_table_name(i.metadata().name()))) {
+                    res.emplace_back(i.metadata().name());
+                }
+            }
+            return make_ready_future<json::json_return_type>(res);
+        });
+    });
+
     cf::get_compression_metadata_off_heap_memory_used.set(r, [](const_req) {
         // FIXME
         // Currently there are no information on the compression
@@ -1073,14 +1091,10 @@ void set_column_family(http_context& ctx, routes& r, sharded<replica::database>&
     });
 
     ss::get_load.set(r, [&db] (std::unique_ptr<http::request> req) {
-        return get_cf_stats(db, [](const replica::column_family_stats& stats) {
-            return stats.live_disk_space_used.on_disk;
-        });
+        return get_cf_stats(db, &replica::column_family_stats::live_disk_space_used);
     });
     ss::get_metrics_load.set(r, [&db] (std::unique_ptr<http::request> req) {
-        return get_cf_stats(db, [](const replica::column_family_stats& stats) {
-            return stats.live_disk_space_used.on_disk;
-        });
+        return get_cf_stats(db, &replica::column_family_stats::live_disk_space_used);
     });
 
     ss::get_keyspaces.set(r, [&db] (const_req req) {
@@ -1201,6 +1215,7 @@ void unset_column_family(http_context& ctx, routes& r) {
     cf::disable_tombstone_gc.unset(r);
     ss::enable_tombstone_gc.unset(r);
     ss::disable_tombstone_gc.unset(r);
+    cf::get_built_indexes.unset(r);
     cf::get_compression_metadata_off_heap_memory_used.unset(r);
     cf::get_compression_parameters.unset(r);
     cf::get_compression_ratio.unset(r);

api/column_family.hh

@@ -13,9 +13,13 @@
 #include <any>
 #include "api/api_init.hh"
 
+namespace db {
+class system_keyspace;
+}
+
 namespace api {
 
-void set_column_family(http_context& ctx, httpd::routes& r, sharded<replica::database>& db);
+void set_column_family(http_context& ctx, httpd::routes& r, sharded<replica::database>& db, sharded<db::system_keyspace>& sys_ks);
 void unset_column_family(http_context& ctx, httpd::routes& r);
 
 table_info parse_table_info(const sstring& name, const replica::database& db);

api/error_injection.cc

@@ -21,10 +21,10 @@ namespace hf = httpd::error_injection_json;
 
 void set_error_injection(http_context& ctx, routes& r) {
 
-    hf::enable_injection.set(r, [](std::unique_ptr<request> req) -> future<json::json_return_type> {
+    hf::enable_injection.set(r, [](std::unique_ptr<request> req) {
         sstring injection = req->get_path_param("injection");
         bool one_shot = req->get_query_param("one_shot") == "True";
-        auto params = co_await util::read_entire_stream_contiguous(*req->content_stream);
+        auto params = req->content;
 
         const size_t max_params_size = 1024 * 1024;
         if (params.size() > max_params_size) {
@@ -39,11 +39,12 @@ void set_error_injection(http_context& ctx, routes& r) {
                 : rjson::parse_to_map<utils::error_injection_parameters>(params);
 
             auto& errinj = utils::get_local_injector();
-            co_await errinj.enable_on_all(injection, one_shot, std::move(parameters));
+            return errinj.enable_on_all(injection, one_shot, std::move(parameters)).then([] {
+                return make_ready_future<json::json_return_type>(json::json_void());
+            });
         } catch (const rjson::error& e) {
             throw httpd::bad_param_exception(format("Failed to parse injections parameters: {}", e.what()));
         }
-        co_return json::json_void();
     });
 
     hf::get_enabled_injections_on_all.set(r, [](std::unique_ptr<request> req) {

api/storage_service.cc

@@ -37,7 +37,6 @@
 #include "gms/gossiper.hh"
 #include "db/system_keyspace.hh"
 #include <seastar/http/exception.hh>
-#include <seastar/http/short_streams.hh>
 #include <seastar/core/coroutine.hh>
 #include <seastar/coroutine/parallel_for_each.hh>
 #include <seastar/coroutine/exception.hh>
@@ -274,13 +273,6 @@ scrub_info parse_scrub_options(const http_context& ctx, std::unique_ptr<http::re
         throw httpd::bad_param_exception(fmt::format("Unknown argument for 'quarantine_mode' parameter: {}", quarantine_mode_str));
     }
 
-    if(req_param<bool>(*req, "drop_unfixable_sstables", false)) {
-        if(scrub_mode != compaction::compaction_type_options::scrub::mode::segregate) {
-            throw httpd::bad_param_exception("The 'drop_unfixable_sstables' parameter is only valid when 'scrub_mode' is 'SEGREGATE'");
-        }
-        info.opts.drop_unfixable = compaction::compaction_type_options::scrub::drop_unfixable_sstables::yes;
-    }
-
     return info;
 }
 
@@ -507,8 +499,9 @@ void set_sstables_loader(http_context& ctx, routes& r, sharded<sstables_loader>&
         auto scope = parse_stream_scope(req->get_query_param("scope"));
         auto primary_replica_only = validate_bool_x(req->get_query_param("primary_replica_only"), false);
 
-        rjson::chunked_content content = co_await util::read_entire_stream(*req->content_stream);
-        rjson::value parsed = rjson::parse(std::move(content));
+        // TODO: the http_server backing the API does not use content streaming
+        // should use it for better performance
+        rjson::value parsed = rjson::parse(req->content);
         if (!parsed.IsArray()) {
             throw httpd::bad_param_exception("malformatted sstables in body");
         }
@@ -536,31 +529,10 @@ void set_view_builder(http_context& ctx, routes& r, sharded<db::view::view_build
         });
     });
 
-    cf::get_built_indexes.set(r, [&vb](std::unique_ptr<http::request> req) -> future<json::json_return_type> {
-        auto [ks, cf_name] = parse_fully_qualified_cf_name(req->get_path_param("name"));
-        // Use of load_built_views() as filtering table should be in sync with
-        // built_indexes_virtual_reader filtering with BUILT_VIEWS table
-        std::vector<db::system_keyspace::view_name> vn = co_await vb.local().get_sys_ks().load_built_views();
-        std::set<sstring> vp;
-        for (auto b : vn) {
-            if (b.first == ks) {
-                vp.insert(b.second);
-            }
-        }
-        replica::database& db = vb.local().get_db();
-        auto uuid = validate_table(db, ks, cf_name);
-        replica::column_family& cf = db.find_column_family(uuid);
-        co_return cf.get_index_manager().list_indexes()
-                | std::views::transform([] (const auto& i) { return i.metadata().name(); })
-                | std::views::filter([&vp] (const auto& n) { return vp.contains(secondary_index::index_table_name(n)); })
-                | std::ranges::to<std::vector>();
-    });
-
 }
 
 void unset_view_builder(http_context& ctx, routes& r) {
     ss::view_build_statuses.unset(r);
-    cf::get_built_indexes.unset(r);
 }
 
 static future<json::json_return_type> describe_ring_as_json(sharded<service::storage_service>& ss, sstring keyspace) {
@@ -740,14 +712,6 @@ rest_get_natural_endpoints(http_context& ctx, sharded<service::storage_service>&
         return res | std::views::transform([] (auto& ep) { return fmt::to_string(ep); }) | std::ranges::to<std::vector>();
 }
 
-static
-json::json_return_type
-rest_get_natural_endpoints_v2(http_context& ctx, sharded<service::storage_service>& ss, const_req req) {
-        auto keyspace = validate_keyspace(ctx, req);
-        auto res = ss.local().get_natural_endpoints(keyspace, req.get_query_param("cf"), req.get_query_param_array("key_component"));
-        return res | std::views::transform([] (auto& ep) { return fmt::to_string(ep); }) | std::ranges::to<std::vector>();
-}
-
 static
 future<json::json_return_type>
 rest_cdc_streams_check_and_repair(sharded<service::storage_service>& ss, std::unique_ptr<http::request> req) {
@@ -772,7 +736,7 @@ rest_cleanup_all(http_context& ctx, sharded<service::storage_service>& ss, std::
             if (!ss.is_topology_coordinator_enabled()) {
                 co_return false;
             }
-            co_await ss.do_clusterwide_vnodes_cleanup();
+            co_await ss.do_cluster_cleanup();
             co_return true;
         });
         if (done) {
@@ -869,25 +833,6 @@ rest_remove_node(sharded<service::storage_service>& ss, std::unique_ptr<http::re
         });
 }
 
-static
-future<json::json_return_type>
-rest_exclude_node(sharded<service::storage_service>& ss, std::unique_ptr<http::request> req) {
-    auto hosts = utils::split_comma_separated_list(req->get_query_param("hosts"))
-        | std::views::transform([] (const sstring& s) { return locator::host_id(utils::UUID(s)); })
-        | std::ranges::to<std::vector<locator::host_id>>();
-
-    auto& topo = ss.local().get_token_metadata().get_topology();
-    for (auto host : hosts) {
-        if (!topo.has_node(host)) {
-            throw bad_param_exception(fmt::format("Host ID {} does not belong to this cluster", host));
-        }
-    }
-
-    apilog.info("exclude_node: hosts={}", hosts);
-    co_await ss.local().mark_excluded(hosts);
-    co_return json_void();
-}
-
 static
 future<json::json_return_type>
 rest_get_removal_status(sharded<service::storage_service>& ss, std::unique_ptr<http::request> req) {
@@ -1805,7 +1750,6 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     ss::describe_ring.set(r, rest_bind(rest_describe_ring, ctx, ss));
     ss::get_current_generation_number.set(r, rest_bind(rest_get_current_generation_number, ss));
     ss::get_natural_endpoints.set(r, rest_bind(rest_get_natural_endpoints, ctx, ss));
-    ss::get_natural_endpoints_v2.set(r, rest_bind(rest_get_natural_endpoints_v2, ctx, ss));
     ss::cdc_streams_check_and_repair.set(r, rest_bind(rest_cdc_streams_check_and_repair, ss));
     ss::cleanup_all.set(r, rest_bind(rest_cleanup_all, ctx, ss));
     ss::reset_cleanup_needed.set(r, rest_bind(rest_reset_cleanup_needed, ctx, ss));
@@ -1814,7 +1758,6 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     ss::decommission.set(r, rest_bind(rest_decommission, ss));
     ss::move.set(r, rest_bind(rest_move, ss));
     ss::remove_node.set(r, rest_bind(rest_remove_node, ss));
-    ss::exclude_node.set(r, rest_bind(rest_exclude_node, ss));
     ss::get_removal_status.set(r, rest_bind(rest_get_removal_status, ss));
     ss::force_remove_completion.set(r, rest_bind(rest_force_remove_completion, ss));
     ss::set_logging_level.set(r, rest_bind(rest_set_logging_level));
@@ -1893,7 +1836,6 @@ void unset_storage_service(http_context& ctx, routes& r) {
     ss::decommission.unset(r);
     ss::move.unset(r);
     ss::remove_node.unset(r);
-    ss::exclude_node.unset(r);
     ss::get_removal_status.unset(r);
     ss::force_remove_completion.unset(r);
     ss::set_logging_level.unset(r);

api/system.cc

@@ -54,8 +54,7 @@ void set_system(http_context& ctx, routes& r) {
 
     hm::set_metrics_config.set(r, [](std::unique_ptr<http::request> req) -> future<json::json_return_type> {
         rapidjson::Document doc;
-        auto content = co_await util::read_entire_stream_contiguous(*req->content_stream);
-        doc.Parse(content.c_str());
+        doc.Parse(req->content.c_str());
         if (!doc.IsArray()) {
             throw bad_param_exception("Expected a json array");
         }
@@ -88,19 +87,21 @@ void set_system(http_context& ctx, routes& r) {
                 relabels[i].expr = element["regex"].GetString();
             }
         }
-        bool failed = false;
-        co_await smp::invoke_on_all([&relabels, &failed] {
-            return metrics::set_relabel_configs(relabels).then([&failed](const metrics::metric_relabeling_result& result) {
-                if (result.metrics_relabeled_due_to_collision > 0) {
-                    failed = true;
+        return do_with(std::move(relabels), false, [](const std::vector<seastar::metrics::relabel_config>& relabels, bool& failed) {
+            return smp::invoke_on_all([&relabels, &failed] {
+                return metrics::set_relabel_configs(relabels).then([&failed](const metrics::metric_relabeling_result& result) {
+                    if (result.metrics_relabeled_due_to_collision > 0) {
+                        failed = true;
+                    }
+                    return;
+                });
+            }).then([&failed](){
+                if (failed) {
+                    throw bad_param_exception("conflicts found during relabeling");
                 }
-                return;
+                return make_ready_future<json::json_return_type>(seastar::json::json_void());
             });
         });
-        if (failed) {
-            throw bad_param_exception("conflicts found during relabeling");
-        }
-        co_return seastar::json::json_void();
     });
 
     hs::get_system_uptime.set(r, [](const_req req) {

api/task_manager.cc

@@ -9,7 +9,6 @@
 #include <seastar/core/chunked_fifo.hh>
 #include <seastar/core/coroutine.hh>
 #include <seastar/coroutine/exception.hh>
-#include <seastar/coroutine/maybe_yield.hh>
 #include <seastar/http/exception.hh>
 
 #include "task_manager.hh"
@@ -265,7 +264,7 @@ void set_task_manager(http_context& ctx, routes& r, sharded<tasks::task_manager>
                 if (id) {
                     module->unregister_task(id);
                 }
-                co_await coroutine::maybe_yield();
+                co_await maybe_yield();
             }
         });
         co_return json_void();

api/tasks.cc

@@ -73,7 +73,7 @@ static future<shared_ptr<compaction::cleanup_keyspace_compaction_task_impl>> for
         co_return nullptr;
     }
     apilog.info("force_keyspace_cleanup: keyspace={} tables={}", keyspace, table_infos);
-    if (!co_await ss.local().is_vnodes_cleanup_allowed(keyspace)) {
+    if (!co_await ss.local().is_cleanup_allowed(keyspace)) {
         auto msg = "Can not perform cleanup operation when topology changes";
         apilog.warn("force_keyspace_cleanup: keyspace={} tables={}: {}", keyspace, table_infos, msg);
         co_await coroutine::return_exception(std::runtime_error(msg));

api/token_metadata.cc

@@ -62,17 +62,6 @@ void set_token_metadata(http_context& ctx, routes& r, sharded<locator::shared_to
         return addr | std::ranges::to<std::vector>();
     });
 
-    ss::get_excluded_nodes.set(r, [&tm](const_req req) {
-        const auto& local_tm = *tm.local().get();
-        std::vector<sstring> eps;
-        local_tm.get_topology().for_each_node([&] (auto& node) {
-            if (node.is_excluded()) {
-                eps.push_back(node.host_id().to_sstring());
-            }
-        });
-        return eps;
-    });
-
     ss::get_joining_nodes.set(r, [&tm, &g](const_req req) {
         const auto& local_tm = *tm.local().get();
         const auto& points = local_tm.get_bootstrap_tokens();
@@ -141,7 +130,6 @@ void unset_token_metadata(http_context& ctx, routes& r) {
     ss::get_leaving_nodes.unset(r);
     ss::get_moving_nodes.unset(r);
     ss::get_joining_nodes.unset(r);
-    ss::get_excluded_nodes.unset(r);
     ss::get_host_id_map.unset(r);
     httpd::endpoint_snitch_info_json::get_datacenter.unset(r);
     httpd::endpoint_snitch_info_json::get_rack.unset(r);

audit/CMakeLists.txt

@@ -5,7 +5,6 @@ target_sources(scylla_audit
   PRIVATE
     audit.cc
     audit_cf_storage_helper.cc
-    audit_composite_storage_helper.cc
     audit_syslog_storage_helper.cc)
 target_include_directories(scylla_audit
   PUBLIC
@@ -17,7 +16,4 @@ target_link_libraries(scylla_audit
   PRIVATE
     cql3)
 
-if (Scylla_USE_PRECOMPILED_HEADER_USE)
-  target_precompile_headers(scylla_audit REUSE_FROM scylla-precompiled-header)
-endif()
 add_whole_archive(audit scylla_audit)

audit/audit.cc

@@ -13,11 +13,9 @@
 #include "cql3/statements/batch_statement.hh"
 #include "cql3/statements/modification_statement.hh"
 #include "storage_helper.hh"
-#include "audit_cf_storage_helper.hh"
-#include "audit_syslog_storage_helper.hh"
-#include "audit_composite_storage_helper.hh"
 #include "audit.hh"
 #include "../db/config.hh"
+#include "utils/class_registrator.hh"
 
 #include <boost/algorithm/string/split.hpp>
 #include <boost/algorithm/string/trim.hpp>
@@ -28,47 +26,6 @@ namespace audit {
 
 logging::logger logger("audit");
 
-static std::set<sstring> parse_audit_modes(const sstring& data) {
-    std::set<sstring> result;
-    if (!data.empty()) {
-        std::vector<sstring> audit_modes;
-        boost::split(audit_modes, data, boost::is_any_of(","));
-        if (audit_modes.empty()) {
-            return {};
-        }
-        for (sstring& audit_mode : audit_modes) {
-            boost::trim(audit_mode);
-            if (audit_mode == "none") {
-                return {};
-            }
-            if (audit_mode != "table" && audit_mode != "syslog") {
-                throw audit_exception(fmt::format("Bad configuration: invalid 'audit': {}", audit_mode));
-            }
-            result.insert(std::move(audit_mode));
-        }
-    }
-    return result;
-}
-
-static std::unique_ptr<storage_helper> create_storage_helper(const std::set<sstring>& audit_modes, cql3::query_processor& qp, service::migration_manager& mm) {
-    SCYLLA_ASSERT(!audit_modes.empty() && !audit_modes.contains("none"));
-
-    std::vector<std::unique_ptr<storage_helper>> helpers;
-    for (const sstring& audit_mode : audit_modes) {
-        if (audit_mode == "table") {
-            helpers.emplace_back(std::make_unique<audit_cf_storage_helper>(qp, mm));
-        } else if (audit_mode == "syslog") {
-            helpers.emplace_back(std::make_unique<audit_syslog_storage_helper>(qp, mm));
-        }
-    }
-
-    SCYLLA_ASSERT(!helpers.empty());
-    if (helpers.size() == 1) {
-        return std::move(helpers.front());
-    }
-    return std::make_unique<audit_composite_storage_helper>(std::move(helpers));
-}
-
 static sstring category_to_string(statement_category category)
 {
     switch (category) {
@@ -146,9 +103,7 @@ static std::set<sstring> parse_audit_keyspaces(const sstring& data) {
 }
 
 audit::audit(locator::shared_token_metadata& token_metadata,
-             cql3::query_processor& qp,
-             service::migration_manager& mm,
-             std::set<sstring>&& audit_modes,
+             sstring&& storage_helper_name,
              std::set<sstring>&& audited_keyspaces,
              std::map<sstring, std::set<sstring>>&& audited_tables,
              category_set&& audited_categories,
@@ -157,21 +112,28 @@ audit::audit(locator::shared_token_metadata& token_metadata,
     , _audited_keyspaces(std::move(audited_keyspaces))
     , _audited_tables(std::move(audited_tables))
     , _audited_categories(std::move(audited_categories))
+    , _storage_helper_class_name(std::move(storage_helper_name))
     , _cfg(cfg)
     , _cfg_keyspaces_observer(cfg.audit_keyspaces.observe([this] (sstring const& new_value){ update_config<std::set<sstring>>(new_value, parse_audit_keyspaces, _audited_keyspaces); }))
     , _cfg_tables_observer(cfg.audit_tables.observe([this] (sstring const& new_value){ update_config<std::map<sstring, std::set<sstring>>>(new_value, parse_audit_tables, _audited_tables); }))
     , _cfg_categories_observer(cfg.audit_categories.observe([this] (sstring const& new_value){ update_config<category_set>(new_value, parse_audit_categories, _audited_categories); }))
-{
-    _storage_helper_ptr = create_storage_helper(std::move(audit_modes), qp, mm);
-}
+{ }
 
 audit::~audit() = default;
 
-future<> audit::start_audit(const db::config& cfg, sharded<locator::shared_token_metadata>& stm, sharded<cql3::query_processor>& qp, sharded<service::migration_manager>& mm) {
-    std::set<sstring> audit_modes = parse_audit_modes(cfg.audit());
-    if (audit_modes.empty()) {
+future<> audit::create_audit(const db::config& cfg, sharded<locator::shared_token_metadata>& stm) {
+    sstring storage_helper_name;
+    if (cfg.audit() == "table") {
+        storage_helper_name = "audit_cf_storage_helper";
+    } else if (cfg.audit() == "syslog") {
+        storage_helper_name = "audit_syslog_storage_helper";
+    } else if (cfg.audit() == "none") {
+        // Audit is off
         logger.info("Audit is disabled");
+
         return make_ready_future<>();
+    } else {
+        throw audit_exception(fmt::format("Bad configuration: invalid 'audit': {}", cfg.audit()));
     }
     category_set audited_categories = parse_audit_categories(cfg.audit_categories());
     std::map<sstring, std::set<sstring>> audited_tables = parse_audit_tables(cfg.audit_tables());
@@ -181,20 +143,19 @@ future<> audit::start_audit(const db::config& cfg, sharded<locator::shared_token
                 cfg.audit(), cfg.audit_categories(), cfg.audit_keyspaces(), cfg.audit_tables());
 
     return audit_instance().start(std::ref(stm),
-                                  std::ref(qp),
-                                  std::ref(mm),
-                                  std::move(audit_modes),
+                                  std::move(storage_helper_name),
                                   std::move(audited_keyspaces),
                                   std::move(audited_tables),
                                   std::move(audited_categories),
-                                  std::cref(cfg))
-    .then([&cfg] {
-        if (!audit_instance().local_is_initialized()) {
-            return make_ready_future<>();
-        }
-        return audit_instance().invoke_on_all([&cfg] (audit& local_audit) {
-            return local_audit.start(cfg);
-        });
+                                  std::cref(cfg));
+}
+
+future<> audit::start_audit(const db::config& cfg, sharded<cql3::query_processor>& qp, sharded<service::migration_manager>& mm) {
+    if (!audit_instance().local_is_initialized()) {
+        return make_ready_future<>();
+    }
+    return audit_instance().invoke_on_all([&cfg, &qp, &mm] (audit& local_audit) {
+        return local_audit.start(cfg, qp.local(), mm.local());
     });
 }
 
@@ -220,7 +181,15 @@ audit_info_ptr audit::create_no_audit_info() {
     return audit_info_ptr();
 }
 
-future<> audit::start(const db::config& cfg) {
+future<> audit::start(const db::config& cfg, cql3::query_processor& qp, service::migration_manager& mm) {
+    try {
+        _storage_helper_ptr = create_object<storage_helper>(_storage_helper_class_name, qp, mm);
+    } catch (no_such_class& e) {
+        logger.error("Can't create audit storage helper {}: not supported", _storage_helper_class_name);
+        throw;
+    } catch (...) {
+        throw;
+    }
     return _storage_helper_ptr->start(cfg);
 }
 

audit/audit.hh

@@ -102,6 +102,7 @@ class audit final : public seastar::async_sharded_service<audit> {
     std::map<sstring, std::set<sstring>> _audited_tables;
     category_set _audited_categories;
 
+    sstring _storage_helper_class_name;
     std::unique_ptr<storage_helper> _storage_helper_ptr;
 
     const db::config& _cfg;
@@ -124,20 +125,18 @@ public:
     static audit& local_audit_instance() {
         return audit_instance().local();
     }
-    static future<> start_audit(const db::config& cfg, sharded<locator::shared_token_metadata>& stm, sharded<cql3::query_processor>& qp, sharded<service::migration_manager>& mm);
+    static future<> create_audit(const db::config& cfg, sharded<locator::shared_token_metadata>& stm);
+    static future<> start_audit(const db::config& cfg, sharded<cql3::query_processor>& qp, sharded<service::migration_manager>& mm);
     static future<> stop_audit();
     static audit_info_ptr create_audit_info(statement_category cat, const sstring& keyspace, const sstring& table);
     static audit_info_ptr create_no_audit_info();
-    audit(locator::shared_token_metadata& stm,
-          cql3::query_processor& qp,
-          service::migration_manager& mm,
-          std::set<sstring>&& audit_modes,
+    audit(locator::shared_token_metadata& stm, sstring&& storage_helper_name,
           std::set<sstring>&& audited_keyspaces,
           std::map<sstring, std::set<sstring>>&& audited_tables,
           category_set&& audited_categories,
           const db::config& cfg);
     ~audit();
-    future<> start(const db::config& cfg);
+    future<> start(const db::config& cfg, cql3::query_processor& qp, service::migration_manager& mm);
     future<> stop();
     future<> shutdown();
     bool should_log(const audit_info* audit_info) const;

audit/audit_cf_storage_helper.cc

@@ -11,11 +11,11 @@
 #include "cql3/query_processor.hh"
 #include "data_dictionary/keyspace_metadata.hh"
 #include "utils/UUID_gen.hh"
+#include "utils/class_registrator.hh"
 #include "cql3/query_options.hh"
 #include "cql3/statements/ks_prop_defs.hh"
 #include "service/migration_manager.hh"
 #include "service/storage_proxy.hh"
-#include "locator/abstract_replication_strategy.hh"
 
 namespace audit {
 
@@ -64,8 +64,8 @@ future<> audit_cf_storage_helper::migrate_audit_table(service::group0_guard grou
             data_dictionary::database db = _qp.db();
             cql3::statements::ks_prop_defs old_ks_prop_defs;
             auto old_ks_metadata = old_ks_prop_defs.as_ks_metadata_update(
-                    ks->metadata(), *_qp.proxy().get_token_metadata_ptr(), db.features(), db.get_config());
-            locator::replication_strategy_config_options strategy_opts;
+                    ks->metadata(), *_qp.proxy().get_token_metadata_ptr(), db.features());
+            std::map<sstring, sstring> strategy_opts;
             for (const auto &dc: _qp.proxy().get_token_metadata_ptr()->get_topology().get_datacenters())
                 strategy_opts[dc] = "3";
 
@@ -73,7 +73,6 @@ future<> audit_cf_storage_helper::migrate_audit_table(service::group0_guard grou
                                                                    "org.apache.cassandra.locator.NetworkTopologyStrategy",
                                                                    strategy_opts,
                                                                    std::nullopt, // initial_tablets
-                                                                   std::nullopt, // consistency_option
                                                                    old_ks_metadata->durable_writes(),
                                                                    old_ks_metadata->get_storage_options(),
                                                                    old_ks_metadata->tables());
@@ -197,4 +196,7 @@ cql3::query_options audit_cf_storage_helper::make_login_data(socket_address node
     return cql3::query_options(cql3::default_cql_config, db::consistency_level::ONE, std::nullopt, std::move(values), false, cql3::query_options::specific_options::DEFAULT);
 }
 
+using registry = class_registrator<storage_helper, audit_cf_storage_helper, cql3::query_processor&, service::migration_manager&>;
+static registry registrator1("audit_cf_storage_helper");
+
 }

audit/audit_composite_storage_helper.cc

@@ -1,68 +0,0 @@
-/*
- * Copyright (C) 2025 ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#include <seastar/core/loop.hh>
-#include <seastar/core/future-util.hh>
-
-#include "audit/audit_composite_storage_helper.hh"
-
-#include "utils/class_registrator.hh"
-
-namespace audit {
-
-audit_composite_storage_helper::audit_composite_storage_helper(std::vector<std::unique_ptr<storage_helper>>&& storage_helpers)
-    : _storage_helpers(std::move(storage_helpers))
-{}
-
-future<> audit_composite_storage_helper::start(const db::config& cfg) {
-    auto res = seastar::parallel_for_each(
-        _storage_helpers,
-        [&cfg] (std::unique_ptr<storage_helper>& h) {
-            return h->start(cfg);
-        }
-    );
-    return res;
-}
-
-future<> audit_composite_storage_helper::stop() {
-    auto res = seastar::parallel_for_each(
-        _storage_helpers,
-        [] (std::unique_ptr<storage_helper>& h) {
-            return h->stop();
-        }
-    );
-    return res;
-}
-
-future<> audit_composite_storage_helper::write(const audit_info* audit_info,
-                                               socket_address node_ip,
-                                               socket_address client_ip,
-                                               db::consistency_level cl,
-                                               const sstring& username,
-                                               bool error) {
-    return seastar::parallel_for_each(
-        _storage_helpers,
-        [audit_info, node_ip, client_ip, cl, &username, error](std::unique_ptr<storage_helper>& h) {
-            return h->write(audit_info, node_ip, client_ip, cl, username, error);
-        }
-    );
-}
-
-future<> audit_composite_storage_helper::write_login(const sstring& username,
-                                                     socket_address node_ip,
-                                                     socket_address client_ip,
-                                                     bool error) {
-    return seastar::parallel_for_each(
-        _storage_helpers,
-        [&username, node_ip, client_ip, error](std::unique_ptr<storage_helper>& h) {
-            return h->write_login(username, node_ip, client_ip, error);
-        }
-    );
-}
-
-} // namespace audit

audit/audit_composite_storage_helper.hh

@@ -1,37 +0,0 @@
-/*
- * Copyright (C) 2025 ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-#pragma once
-
-#include "audit/audit.hh"
-#include <seastar/core/future.hh>
-
-#include "storage_helper.hh"
-
-namespace audit {
-
-class audit_composite_storage_helper : public storage_helper {
-    std::vector<std::unique_ptr<storage_helper>> _storage_helpers;
-
-public:
-    explicit audit_composite_storage_helper(std::vector<std::unique_ptr<storage_helper>>&&);
-    virtual ~audit_composite_storage_helper() = default;
-    virtual future<> start(const db::config& cfg) override;
-    virtual future<> stop() override;
-    virtual future<> write(const audit_info* audit_info,
-                           socket_address node_ip,
-                           socket_address client_ip,
-                           db::consistency_level cl,
-                           const sstring& username,
-                           bool error) override;
-    virtual future<> write_login(const sstring& username,
-                                 socket_address node_ip,
-                                 socket_address client_ip,
-                                 bool error) override;
-};
-
-} // namespace audit

audit/audit_syslog_storage_helper.cc

@@ -21,6 +21,7 @@
 #include <fmt/chrono.h>
 
 #include "cql3/query_processor.hh"
+#include "utils/class_registrator.hh"
 
 namespace cql3 {
 
@@ -142,4 +143,7 @@ future<> audit_syslog_storage_helper::write_login(const sstring& username,
     co_await syslog_send_helper(msg.c_str());
 }
 
+using registry = class_registrator<storage_helper, audit_syslog_storage_helper, cql3::query_processor&, service::migration_manager&>;
+static registry registrator1("audit_syslog_storage_helper");
+
 }

auth/CMakeLists.txt

@@ -9,7 +9,6 @@ target_sources(scylla_auth
     allow_all_authorizer.cc
     authenticated_user.cc
     authenticator.cc
-    cache.cc
     certificate_authenticator.cc
     common.cc
     default_authorizer.cc
@@ -45,8 +44,5 @@ target_link_libraries(scylla_auth
 
 add_whole_archive(auth scylla_auth)
 
-if (Scylla_USE_PRECOMPILED_HEADER_USE)
-  target_precompile_headers(scylla_auth REUSE_FROM scylla-precompiled-header)
-endif()
 check_headers(check-headers scylla_auth
   GLOB_RECURSE ${CMAKE_CURRENT_SOURCE_DIR}/*.hh)

auth/allow_all_authenticator.cc

@@ -9,6 +9,7 @@
 #include "auth/allow_all_authenticator.hh"
 
 #include "service/migration_manager.hh"
+#include "utils/alien_worker.hh"
 #include "utils/class_registrator.hh"
 
 namespace auth {
@@ -22,6 +23,6 @@ static const class_registrator<
         cql3::query_processor&,
         ::service::raft_group0_client&,
         ::service::migration_manager&,
-        cache&> registration("org.apache.cassandra.auth.AllowAllAuthenticator");
+        utils::alien_worker&> registration("org.apache.cassandra.auth.AllowAllAuthenticator");
 
 }

auth/allow_all_authenticator.hh

@@ -12,8 +12,8 @@
 
 #include "auth/authenticated_user.hh"
 #include "auth/authenticator.hh"
-#include "auth/cache.hh"
 #include "auth/common.hh"
+#include "utils/alien_worker.hh"
 
 namespace cql3 {
 class query_processor;
@@ -29,7 +29,7 @@ extern const std::string_view allow_all_authenticator_name;
 
 class allow_all_authenticator final : public authenticator {
 public:
-    allow_all_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&) {
+    allow_all_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&) {
     }
 
     virtual future<> start() override {

auth/cache.cc

@@ -1,188 +0,0 @@
-/*
- * Copyright (C) 2017-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#include "auth/cache.hh"
-#include "auth/common.hh"
-#include "auth/roles-metadata.hh"
-#include "cql3/query_processor.hh"
-#include "cql3/untyped_result_set.hh"
-#include "db/consistency_level_type.hh"
-#include "db/system_keyspace.hh"
-#include "schema/schema.hh"
-#include <iterator>
-#include <seastar/core/abort_source.hh>
-#include <seastar/coroutine/maybe_yield.hh>
-#include <seastar/core/format.hh>
-
-namespace auth {
-
-logging::logger logger("auth-cache");
-
-cache::cache(cql3::query_processor& qp, abort_source& as) noexcept
-    : _current_version(0)
-    , _qp(qp)
-    , _loading_sem(1)
-    , _as(as) {
-}
-
-lw_shared_ptr<const cache::role_record> cache::get(const role_name_t& role) const noexcept {
-    auto it = _roles.find(role);
-    if (it == _roles.end()) {
-        return {};
-    }
-    return it->second;
-}
-
-future<lw_shared_ptr<cache::role_record>> cache::fetch_role(const role_name_t& role) const {
-    auto rec = make_lw_shared<role_record>();
-    rec->version = _current_version;
-
-    auto fetch = [this, &role](const sstring& q) {
-        return _qp.execute_internal(q, db::consistency_level::LOCAL_ONE,
-                internal_distributed_query_state(), {role},
-                cql3::query_processor::cache_internal::yes);
-    };
-    // roles
-    {
-        static const sstring q = format("SELECT * FROM {}.{} WHERE role = ?", db::system_keyspace::NAME, meta::roles_table::name);
-        auto rs = co_await fetch(q);
-        if (!rs->empty()) {
-            auto& r = rs->one();
-            rec->is_superuser = r.get_or<bool>("is_superuser", false);
-            rec->can_login = r.get_or<bool>("can_login", false);
-            rec->salted_hash = r.get_or<sstring>("salted_hash", "");
-            if (r.has("member_of")) {
-                auto mo = r.get_set<sstring>("member_of");
-                rec->member_of.insert(
-                        std::make_move_iterator(mo.begin()),
-                        std::make_move_iterator(mo.end()));
-            }
-        } else {
-            // role got deleted
-            co_return nullptr;
-        }
-    }
-    // members
-    {
-        static const sstring q = format("SELECT role, member FROM {}.{} WHERE role = ?", db::system_keyspace::NAME, ROLE_MEMBERS_CF);
-        auto rs = co_await fetch(q);
-        for (const auto& r : *rs) {
-            rec->members.insert(r.get_as<sstring>("member"));
-            co_await coroutine::maybe_yield();
-        }
-    }
-    // attributes
-    {
-        static const sstring q = format("SELECT role, name, value FROM {}.{} WHERE role = ?", db::system_keyspace::NAME, ROLE_ATTRIBUTES_CF);
-        auto rs = co_await fetch(q);
-        for (const auto& r : *rs) {
-            rec->attributes[r.get_as<sstring>("name")] =
-                    r.get_as<sstring>("value");
-            co_await coroutine::maybe_yield();
-        }
-    }
-    // permissions
-    {
-        static const sstring q = format("SELECT role, resource, permissions FROM {}.{} WHERE role = ?", db::system_keyspace::NAME, PERMISSIONS_CF);
-        auto rs = co_await fetch(q);
-        for (const auto& r : *rs) {
-            auto resource = r.get_as<sstring>("resource");
-            auto perms_strings = r.get_set<sstring>("permissions");
-            std::unordered_set<sstring> perms_set(perms_strings.begin(), perms_strings.end());
-            auto pset = permissions::from_strings(perms_set);
-            rec->permissions[std::move(resource)] = std::move(pset);
-            co_await coroutine::maybe_yield();
-        }
-    }
-    co_return rec;
-}
-
-future<> cache::prune_all() noexcept {
-    for (auto it = _roles.begin(); it != _roles.end(); ) {
-        if (it->second->version != _current_version) {
-            _roles.erase(it++);
-            co_await coroutine::maybe_yield();
-        } else {
-            ++it;
-        }
-    }
-    co_return;
-}
-
-future<> cache::load_all() {
-    if (legacy_mode(_qp)) {
-        co_return;
-    }
-    SCYLLA_ASSERT(this_shard_id() == 0);
-    auto units = co_await get_units(_loading_sem, 1, _as);
-
-    ++_current_version;
-
-    logger.info("Loading all roles");
-    const uint32_t page_size = 128;
-    auto loader = [this](const cql3::untyped_result_set::row& r) -> future<stop_iteration> {
-        const auto name = r.get_as<sstring>("role");
-        auto role = co_await fetch_role(name);
-        if (role) {
-            _roles[name] = role;
-        }
-        co_return stop_iteration::no;
-    };
-    co_await _qp.query_internal(format("SELECT * FROM {}.{}",
-            db::system_keyspace::NAME, meta::roles_table::name),
-            db::consistency_level::LOCAL_ONE, {}, page_size, loader);
-
-    co_await prune_all();
-    for (const auto& [name, role] : _roles) {
-        co_await distribute_role(name, role);
-    }
-    co_await container().invoke_on_others([this](cache& c) -> future<> {
-        c._current_version = _current_version;
-        co_await c.prune_all();
-    });
-}
-
-future<> cache::load_roles(std::unordered_set<role_name_t> roles) {
-    if (legacy_mode(_qp)) {
-        co_return;
-    }
-    SCYLLA_ASSERT(this_shard_id() == 0);
-    auto units = co_await get_units(_loading_sem, 1, _as);
-
-    for (const auto& name : roles) {
-        logger.info("Loading role {}", name);
-        auto role = co_await fetch_role(name);
-         if (role) {
-            _roles[name] = role;
-        } else {
-            _roles.erase(name);
-        }
-        co_await distribute_role(name, role);
-    }
-}
-
-future<> cache::distribute_role(const role_name_t& name, lw_shared_ptr<role_record> role) {
-    auto role_ptr = role.get();
-    co_await container().invoke_on_others([&name, role_ptr](cache& c) {
-        if (!role_ptr) {
-            c._roles.erase(name);
-            return;
-        }
-        auto role_copy = make_lw_shared<role_record>(*role_ptr);
-        c._roles[name] = std::move(role_copy);
-    });
-}
-
-bool cache::includes_table(const table_id& id) noexcept {
-    return id == db::system_keyspace::roles()->id()
-            || id == db::system_keyspace::role_members()->id()
-            || id == db::system_keyspace::role_attributes()->id()
-            || id == db::system_keyspace::role_permissions()->id();
-}
-
-} // namespace auth

auth/cache.hh

@@ -1,65 +0,0 @@
-/*
- * Copyright (C) 2025-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#pragma once
-
-#include <seastar/core/abort_source.hh>
-#include <unordered_set>
-#include <unordered_map>
-
-#include <seastar/core/sstring.hh>
-#include <seastar/core/future.hh>
-#include <seastar/core/sharded.hh>
-#include <seastar/core/shared_ptr.hh>
-#include <seastar/core/semaphore.hh>
-
-#include <absl/container/flat_hash_map.h>
-
-#include "auth/permission.hh"
-#include "auth/common.hh"
-
-namespace cql3 { class query_processor; }
-
-namespace auth {
-
-class cache : public peering_sharded_service<cache> {
-public:
-    using role_name_t = sstring;
-    using version_tag_t = char;
-
-	struct role_record {
-        bool can_login = false;
-        bool is_superuser = false;
-        std::unordered_set<role_name_t> member_of;
-        std::unordered_set<role_name_t> members;
-        sstring salted_hash;
-        std::unordered_map<sstring, sstring> attributes;
-        std::unordered_map<sstring, permission_set> permissions;
-        version_tag_t version; // used for seamless cache reloads
-    };
-
-    explicit cache(cql3::query_processor& qp, abort_source& as) noexcept;
-    lw_shared_ptr<const role_record> get(const role_name_t& role) const noexcept;
-    future<> load_all();
-    future<> load_roles(std::unordered_set<role_name_t> roles);
-    static bool includes_table(const table_id&) noexcept;
-
-private:
-    using roles_map = absl::flat_hash_map<role_name_t, lw_shared_ptr<role_record>>;
-    roles_map _roles;
-    version_tag_t _current_version;
-    cql3::query_processor& _qp;
-    semaphore _loading_sem;
-    abort_source& _as;
-
-    future<lw_shared_ptr<role_record>> fetch_role(const role_name_t& role) const;
-    future<> prune_all() noexcept;
-    future<> distribute_role(const role_name_t& name, const lw_shared_ptr<role_record> role);
-};
-
-} // namespace auth

auth/certificate_authenticator.cc

@@ -8,7 +8,6 @@
  */
 
 #include "auth/certificate_authenticator.hh"
-#include "auth/cache.hh"
 
 #include <boost/regex.hpp>
 #include <fmt/ranges.h>
@@ -35,13 +34,13 @@ static const class_registrator<auth::authenticator
     , cql3::query_processor&
     , ::service::raft_group0_client&
     , ::service::migration_manager&
-    , auth::cache&> cert_auth_reg(CERT_AUTH_NAME);
+    , 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&, auth::cache&)
+auth::certificate_authenticator::certificate_authenticator(cql3::query_processor& qp, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&)
     : _queries([&] {
         auto& conf = qp.db().get_config();
         auto queries = conf.auth_certificate_role_queries();
@@ -76,9 +75,9 @@ auth::certificate_authenticator::certificate_authenticator(cql3::query_processor
                         throw std::invalid_argument(fmt::format("Invalid source: {}", map.at(cfg_source_attr)));
                     }
                     continue;
-                } catch (const std::out_of_range&) {
+                } catch (std::out_of_range&) {
                     // just fallthrough
-                } catch (const boost::regex_error&) {
+                } catch (boost::regex_error&) {
                     std::throw_with_nested(std::invalid_argument(fmt::format("Invalid query expression: {}", map.at(cfg_query_attr))));
                 }
             }

auth/certificate_authenticator.hh

@@ -10,6 +10,7 @@
 #pragma once
 
 #include "auth/authenticator.hh"
+#include "utils/alien_worker.hh"
 #include <boost/regex_fwd.hpp>  // IWYU pragma: keep
 
 namespace cql3 {
@@ -25,15 +26,13 @@ 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&, cache&);
+    certificate_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&);
     ~certificate_authenticator();
 
     future<> start() override;

auth/common.cc

@@ -94,7 +94,7 @@ static future<> create_legacy_metadata_table_if_missing_impl(
         try {
             co_return co_await mm.announce(co_await ::service::prepare_new_column_family_announcement(qp.proxy(), table, ts),
                     std::move(group0_guard), format("auth: create {} metadata table", table->cf_name()));
-        } catch (const exceptions::already_exists_exception&) {}
+        } catch (exceptions::already_exists_exception&) {}
     }
 }
 

auth/common.hh

@@ -48,10 +48,6 @@ extern constinit const std::string_view AUTH_PACKAGE_NAME;
 
 } // namespace meta
 
-constexpr std::string_view PERMISSIONS_CF = "role_permissions";
-constexpr std::string_view ROLE_MEMBERS_CF = "role_members";
-constexpr std::string_view ROLE_ATTRIBUTES_CF = "role_attributes";
-
 // This is a helper to check whether auth-v2 is on.
 bool legacy_mode(cql3::query_processor& qp);
 

auth/default_authorizer.cc

@@ -37,6 +37,7 @@ std::string_view default_authorizer::qualified_java_name() const {
 static constexpr std::string_view ROLE_NAME = "role";
 static constexpr std::string_view RESOURCE_NAME = "resource";
 static constexpr std::string_view PERMISSIONS_NAME = "permissions";
+static constexpr std::string_view PERMISSIONS_CF = "role_permissions";
 
 static logging::logger alogger("default_authorizer");
 
@@ -256,7 +257,7 @@ future<> default_authorizer::revoke_all(std::string_view role_name, ::service::g
         } else {
             co_await collect_mutations(_qp, mc, query, {sstring(role_name)});
         }
-    } catch (const exceptions::request_execution_exception& e) {
+    } catch (exceptions::request_execution_exception& e) {
         alogger.warn("CassandraAuthorizer failed to revoke all permissions of {}: {}", role_name, e);
     }
 }
@@ -293,13 +294,13 @@ future<> default_authorizer::revoke_all_legacy(const resource& resource) {
                                 [resource](auto ep) {
                     try {
                         std::rethrow_exception(ep);
-                    } catch (const exceptions::request_execution_exception& e) {
+                    } catch (exceptions::request_execution_exception& e) {
                         alogger.warn("CassandraAuthorizer failed to revoke all permissions on {}: {}", resource, e);
                     }
 
                 });
             });
-        } catch (const exceptions::request_execution_exception& e) {
+        } catch (exceptions::request_execution_exception& e) {
             alogger.warn("CassandraAuthorizer failed to revoke all permissions on {}: {}", resource, e);
             return make_ready_future();
         }

auth/ldap_role_manager.cc

@@ -83,18 +83,17 @@ static const class_registrator<
     ldap_role_manager,
     cql3::query_processor&,
     ::service::raft_group0_client&,
-    ::service::migration_manager&,
-    cache&> registration(ldap_role_manager_full_name);
+    ::service::migration_manager&> registration(ldap_role_manager_full_name);
 
 ldap_role_manager::ldap_role_manager(
         std::string_view query_template, std::string_view target_attr, std::string_view bind_name, std::string_view bind_password,
-        cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm, cache& cache)
-        : _std_mgr(qp, rg0c, mm, cache), _group0_client(rg0c), _query_template(query_template), _target_attr(target_attr), _bind_name(bind_name)
+        cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm)
+        : _std_mgr(qp, rg0c, mm), _group0_client(rg0c), _query_template(query_template), _target_attr(target_attr), _bind_name(bind_name)
         , _bind_password(bind_password)
         , _connection_factory(bind(std::mem_fn(&ldap_role_manager::reconnect), std::ref(*this))) {
 }
 
-ldap_role_manager::ldap_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm, cache& cache)
+ldap_role_manager::ldap_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm)
     : ldap_role_manager(
             qp.db().get_config().ldap_url_template(),
             qp.db().get_config().ldap_attr_role(),
@@ -102,8 +101,7 @@ ldap_role_manager::ldap_role_manager(cql3::query_processor& qp, ::service::raft_
             qp.db().get_config().ldap_bind_passwd(),
             qp,
             rg0c,
-            mm,
-            cache) {
+            mm) {
 }
 
 std::string_view ldap_role_manager::qualified_java_name() const noexcept {

auth/ldap_role_manager.hh

@@ -14,7 +14,6 @@
 
 #include "ent/ldap/ldap_connection.hh"
 #include "standard_role_manager.hh"
-#include "auth/cache.hh"
 
 namespace auth {
 
@@ -44,13 +43,12 @@ class ldap_role_manager : public role_manager {
             std::string_view bind_password, ///< LDAP bind credentials.
             cql3::query_processor& qp, ///< Passed to standard_role_manager.
             ::service::raft_group0_client& rg0c, ///< Passed to standard_role_manager.
-            ::service::migration_manager& mm, ///< Passed to standard_role_manager.
-            cache& cache ///< Passed to standard_role_manager.
+            ::service::migration_manager& mm ///< Passed to standard_role_manager.
     );
 
     /// Retrieves LDAP configuration entries from qp and invokes the other constructor.  Required by
     /// class_registrator<role_manager>.
-    ldap_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm, cache& cache);
+    ldap_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& rg0c, ::service::migration_manager& mm);
 
     /// Thrown when query-template parsing fails.
     struct url_error : public std::runtime_error {

auth/maintenance_socket_role_manager.cc

@@ -11,7 +11,6 @@
 #include <seastar/core/future.hh>
 #include <stdexcept>
 #include <string_view>
-#include "auth/cache.hh"
 #include "cql3/description.hh"
 #include "utils/class_registrator.hh"
 
@@ -24,8 +23,7 @@ static const class_registrator<
         maintenance_socket_role_manager,
         cql3::query_processor&,
         ::service::raft_group0_client&,
-        ::service::migration_manager&,
-        cache&> registration(sstring{maintenance_socket_role_manager_name});
+        ::service::migration_manager&> registration(sstring{maintenance_socket_role_manager_name});
 
 
 std::string_view maintenance_socket_role_manager::qualified_java_name() const noexcept {

auth/maintenance_socket_role_manager.hh

@@ -8,7 +8,6 @@
 
 #pragma once
 
-#include "auth/cache.hh"
 #include "auth/resource.hh"
 #include "auth/role_manager.hh"
 #include <seastar/core/future.hh>
@@ -30,7 +29,7 @@ extern const std::string_view maintenance_socket_role_manager_name;
 // system_auth keyspace, which may be not yet created when the maintenance socket starts listening.
 class maintenance_socket_role_manager final : public role_manager {
 public:
-    maintenance_socket_role_manager(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&) {}
+    maintenance_socket_role_manager(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&) {}
 
     virtual std::string_view qualified_java_name() const noexcept override;
 

auth/password_authenticator.cc

@@ -49,7 +49,7 @@ static const class_registrator<
         cql3::query_processor&,
         ::service::raft_group0_client&,
         ::service::migration_manager&,
-        cache&> password_auth_reg("org.apache.cassandra.auth.PasswordAuthenticator");
+        utils::alien_worker&> password_auth_reg("org.apache.cassandra.auth.PasswordAuthenticator");
 
 static thread_local auto rng_for_salt = std::default_random_engine(std::random_device{}());
 
@@ -63,13 +63,13 @@ std::string password_authenticator::default_superuser(const db::config& cfg) {
 password_authenticator::~password_authenticator() {
 }
 
-password_authenticator::password_authenticator(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, cache& cache)
+password_authenticator::password_authenticator(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, utils::alien_worker& hashing_worker)
     : _qp(qp)
     , _group0_client(g0)
     , _migration_manager(mm)
-    , _cache(cache)
     , _stopped(make_ready_future<>()) 
     , _superuser(default_superuser(qp.db().get_config()))
+    , _hashing_worker(hashing_worker)
 {}
 
 static bool has_salted_hash(const cql3::untyped_result_set_row& row) {
@@ -315,31 +315,24 @@ future<authenticated_user> password_authenticator::authenticate(
     const sstring password = credentials.at(PASSWORD_KEY);
 
     try {
-        std::optional<sstring> salted_hash;
-        if (legacy_mode(_qp)) {
-            salted_hash = co_await get_password_hash(username);
-            if (!salted_hash) {
-                throw exceptions::authentication_exception("Username and/or password are incorrect");
-            }
-        } else {
-            auto role = _cache.get(username);
-            if (!role || role->salted_hash.empty()) {
-                throw exceptions::authentication_exception("Username and/or password are incorrect");
-            }
-            salted_hash = role->salted_hash;
+        const std::optional<sstring> salted_hash = co_await get_password_hash(username);
+        if (!salted_hash) {
+            throw exceptions::authentication_exception("Username and/or password are incorrect");
         }
-        const bool password_match = co_await passwords::check(password, *salted_hash);
+        const bool password_match = co_await _hashing_worker.submit<bool>([password = std::move(password), salted_hash = std::move(salted_hash)]{
+            return passwords::check(password, *salted_hash);
+        });
         if (!password_match) {
             throw exceptions::authentication_exception("Username and/or password are incorrect");
         }
         co_return username;
-    } catch (const std::system_error &) {
+    } catch (std::system_error &) {
         std::throw_with_nested(exceptions::authentication_exception("Could not verify password"));
-    } catch (const exceptions::request_execution_exception& e) {
+    } catch (exceptions::request_execution_exception& e) {
         std::throw_with_nested(exceptions::authentication_exception(e.what()));
-    } catch (const exceptions::authentication_exception& e) {
+    } catch (exceptions::authentication_exception& e) {
         std::throw_with_nested(e);
-    } catch (const exceptions::unavailable_exception& e) {
+    } catch (exceptions::unavailable_exception& e) {
         std::throw_with_nested(exceptions::authentication_exception(e.get_message()));
     } catch (...) {
         std::throw_with_nested(exceptions::authentication_exception("authentication failed"));

auth/password_authenticator.hh

@@ -16,8 +16,8 @@
 #include "db/consistency_level_type.hh"
 #include "auth/authenticator.hh"
 #include "auth/passwords.hh"
-#include "auth/cache.hh"
 #include "service/raft/raft_group0_client.hh"
+#include "utils/alien_worker.hh"
 
 namespace db {
     class config;
@@ -41,19 +41,19 @@ class password_authenticator : public authenticator {
     cql3::query_processor& _qp;
     ::service::raft_group0_client& _group0_client;
     ::service::migration_manager& _migration_manager;
-    cache& _cache;
     future<> _stopped;
     abort_source _as;
     std::string _superuser; // default superuser name from the config (may or may not be present in roles table)
     shared_promise<> _superuser_created_promise;
     // We used to also support bcrypt, SHA-256, and MD5 (ref. scylladb#24524).
     constexpr static auth::passwords::scheme _scheme = passwords::scheme::sha_512;
+    utils::alien_worker& _hashing_worker;
 
 public:
     static db::consistency_level consistency_for_user(std::string_view role_name);
     static std::string default_superuser(const db::config&);
 
-    password_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&);
+    password_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&);
 
     ~password_authenticator();
 

auth/passwords.cc

@@ -7,8 +7,6 @@
  */
 
 #include "auth/passwords.hh"
-#include "utils/crypt_sha512.hh"
-#include <seastar/core/coroutine.hh>
 
 #include <cerrno>
 
@@ -23,46 +21,25 @@ static thread_local crypt_data tlcrypt = {};
 
 namespace detail {
 
-void verify_hashing_output(const char * res) {
-    if (!res || (res[0] == '*')) {
-        throw std::system_error(errno, std::system_category());
-    }
-}
-
 void verify_scheme(scheme scheme) {
     const sstring random_part_of_salt = "aaaabbbbccccdddd";
 
     const sstring salt = sstring(prefix_for_scheme(scheme)) + random_part_of_salt;
     const char* e = crypt_r("fisk", salt.c_str(), &tlcrypt);
-    try {
-        verify_hashing_output(e);
-    } catch (const std::system_error& ex) {
-        throw no_supported_schemes();
+
+    if (e && (e[0] != '*')) {
+        return;
     }
+
+    throw no_supported_schemes();
 }
 
 sstring hash_with_salt(const sstring& pass, const sstring& salt) {
     auto res = crypt_r(pass.c_str(), salt.c_str(), &tlcrypt);
-    verify_hashing_output(res);
-    return res;
-}
-
-seastar::future<sstring> hash_with_salt_async(const sstring& pass, const sstring& salt) {
-    sstring res;
-    // Only SHA-512 hashes for passphrases shorter than 256 bytes can be computed using
-    // the __crypt_sha512 method. For other computations, we fall back to the
-    // crypt_r implementation from `<crypt.h>`, which can stall.
-    if (salt.starts_with(prefix_for_scheme(scheme::sha_512)) && pass.size() <= 255) {
-        char buf[128];
-        const char * output_ptr = co_await __crypt_sha512(pass.c_str(), salt.c_str(), buf);
-        verify_hashing_output(output_ptr);
-        res = output_ptr;
-    } else {
-        const char * output_ptr = crypt_r(pass.c_str(), salt.c_str(), &tlcrypt);
-        verify_hashing_output(output_ptr);
-        res = output_ptr;
+    if (!res || (res[0] == '*')) {
+        throw std::system_error(errno, std::system_category());
     }
-    co_return res;
+    return res;
 }
 
 std::string_view prefix_for_scheme(scheme c) noexcept {
@@ -81,9 +58,8 @@ no_supported_schemes::no_supported_schemes()
         : std::runtime_error("No allowed hashing schemes are supported on this system") {
 }
 
-seastar::future<bool> check(const sstring& pass, const sstring& salted_hash) {
-    const auto pwd_hash = co_await detail::hash_with_salt_async(pass, salted_hash);
-    co_return pwd_hash == salted_hash;
+bool check(const sstring& pass, const sstring& salted_hash) {
+    return detail::hash_with_salt(pass, salted_hash) == salted_hash;
 }
 
 } // namespace auth::passwords

auth/passwords.hh

@@ -11,7 +11,6 @@
 #include <random>
 #include <stdexcept>
 
-#include <seastar/core/future.hh>
 #include <seastar/core/sstring.hh>
 
 #include "seastarx.hh"
@@ -76,23 +75,11 @@ sstring generate_salt(RandomNumberEngine& g, scheme scheme) {
 
 ///
 /// Hash a password combined with an implementation-specific salt string.
-/// Deprecated in favor of `hash_with_salt_async`. This function is still used
-/// when generating password hashes for storage to ensure that
-/// `hash_with_salt` and `hash_with_salt_async` produce identical results,
-/// preserving backward compatibility.
 ///
 /// \throws \ref std::system_error when an unexpected implementation-specific error occurs.
 ///
 sstring hash_with_salt(const sstring& pass, const sstring& salt);
 
-///
-/// Async version of `hash_with_salt` that returns a future.
-/// If possible, hashing uses `coroutine::maybe_yield` to prevent reactor stalls.
-///
-/// \throws \ref std::system_error when an unexpected implementation-specific error occurs.
-///
-seastar::future<sstring> hash_with_salt_async(const sstring& pass, const sstring& salt);
-
 } // namespace detail
 
 ///
@@ -120,6 +107,6 @@ sstring hash(const sstring& pass, RandomNumberEngine& g, scheme scheme) {
 ///
 /// \throws \ref std::system_error when an unexpected implementation-specific error occurs.
 ///
-seastar::future<bool> check(const sstring& pass, const sstring& salted_hash);
+bool check(const sstring& pass, const sstring& salted_hash);
 
 } // namespace auth::passwords

auth/saslauthd_authenticator.cc

@@ -35,9 +35,9 @@ static const class_registrator<
         cql3::query_processor&,
         ::service::raft_group0_client&,
         ::service::migration_manager&,
-        cache&> saslauthd_auth_reg("com.scylladb.auth.SaslauthdAuthenticator");
+        utils::alien_worker&> saslauthd_auth_reg("com.scylladb.auth.SaslauthdAuthenticator");
 
-saslauthd_authenticator::saslauthd_authenticator(cql3::query_processor& qp, ::service::raft_group0_client&, ::service::migration_manager&, cache&)
+saslauthd_authenticator::saslauthd_authenticator(cql3::query_processor& qp, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&)
     : _socket_path(qp.db().get_config().saslauthd_socket_path())
 {}
 

auth/saslauthd_authenticator.hh

@@ -11,7 +11,7 @@
 #pragma once
 
 #include "auth/authenticator.hh"
-#include "auth/cache.hh"
+#include "utils/alien_worker.hh"
 
 namespace cql3 {
 class query_processor;
@@ -29,7 +29,7 @@ namespace auth {
 class saslauthd_authenticator : public authenticator {
     sstring _socket_path; ///< Path to the domain socket on which saslauthd is listening.
 public:
-    saslauthd_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&);
+    saslauthd_authenticator(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, utils::alien_worker&);
 
     future<> start() override;
 

auth/service.cc

@@ -17,7 +17,6 @@
 #include <chrono>
 
 #include <seastar/core/future-util.hh>
-#include <seastar/core/shard_id.hh>
 #include <seastar/core/sharded.hh>
 #include <seastar/core/shared_ptr.hh>
 
@@ -158,7 +157,6 @@ static future<> validate_role_exists(const service& ser, std::string_view role_n
 
 service::service(
         utils::loading_cache_config c,
-        cache& cache,
         cql3::query_processor& qp,
         ::service::raft_group0_client& g0,
         ::service::migration_notifier& mn,
@@ -168,7 +166,6 @@ service::service(
         maintenance_socket_enabled used_by_maintenance_socket)
             : _loading_cache_config(std::move(c))
             , _permissions_cache(nullptr)
-            , _cache(cache)
             , _qp(qp)
             , _group0_client(g0)
             , _mnotifier(mn)
@@ -191,16 +188,15 @@ service::service(
         ::service::migration_manager& mm,
         const service_config& sc,
         maintenance_socket_enabled used_by_maintenance_socket,
-        cache& cache)
+        utils::alien_worker& hashing_worker)
             : service(
                       std::move(c),
-                      cache,
                       qp,
                       g0,
                       mn,
                       create_object<authorizer>(sc.authorizer_java_name, qp, g0, mm),
-                      create_object<authenticator>(sc.authenticator_java_name, qp, g0, mm, cache),
-                      create_object<role_manager>(sc.role_manager_java_name, qp, g0, mm, cache),
+                      create_object<authenticator>(sc.authenticator_java_name, qp, g0, mm, hashing_worker),
+                      create_object<role_manager>(sc.role_manager_java_name, qp, g0, mm),
                       used_by_maintenance_socket) {
 }
 
@@ -219,13 +215,12 @@ future<> service::create_legacy_keyspace_if_missing(::service::migration_manager
                     meta::legacy::AUTH_KS,
                     "org.apache.cassandra.locator.SimpleStrategy",
                     opts,
-                    std::nullopt,
                     std::nullopt);
 
             try {
                 co_return co_await mm.announce(::service::prepare_new_keyspace_announcement(db.real_database(), ksm, ts),
                         std::move(group0_guard), seastar::format("auth_service: create {} keyspace", meta::legacy::AUTH_KS));
-            } catch (const ::service::group0_concurrent_modification&) {
+            } catch (::service::group0_concurrent_modification&) {
                 log.info("Concurrent operation is detected while creating {} keyspace, retrying.", meta::legacy::AUTH_KS);
             }
         }
@@ -236,9 +231,6 @@ future<> service::start(::service::migration_manager& mm, db::system_keyspace& s
     auto auth_version = co_await sys_ks.get_auth_version();
     // version is set in query processor to be easily available in various places we call auth::legacy_mode check.
     _qp.auth_version = auth_version;
-    if (this_shard_id() == 0) {
-        co_await _cache.load_all();
-    }
     if (!_used_by_maintenance_socket) {
         // this legacy keyspace is only used by cqlsh
         // it's needed when executing `list roles` or `list users`

auth/service.hh

@@ -21,12 +21,12 @@
 #include "auth/authorizer.hh"
 #include "auth/permission.hh"
 #include "auth/permissions_cache.hh"
-#include "auth/cache.hh"
 #include "auth/role_manager.hh"
 #include "auth/common.hh"
 #include "cql3/description.hh"
 #include "seastarx.hh"
 #include "service/raft/raft_group0_client.hh"
+#include "utils/alien_worker.hh"
 #include "utils/observable.hh"
 #include "utils/serialized_action.hh"
 #include "service/maintenance_mode.hh"
@@ -77,7 +77,6 @@ public:
 class service final : public seastar::peering_sharded_service<service> {
     utils::loading_cache_config _loading_cache_config;
     std::unique_ptr<permissions_cache> _permissions_cache;
-    cache& _cache;
 
     cql3::query_processor& _qp;
 
@@ -108,7 +107,6 @@ class service final : public seastar::peering_sharded_service<service> {
 public:
     service(
             utils::loading_cache_config,
-            cache& cache,
             cql3::query_processor&,
             ::service::raft_group0_client&,
             ::service::migration_notifier&,
@@ -130,7 +128,7 @@ public:
             ::service::migration_manager&,
             const service_config&,
             maintenance_socket_enabled,
-            cache&);
+            utils::alien_worker&);
 
     future<> start(::service::migration_manager&, db::system_keyspace&);
 

auth/standard_role_manager.cc

@@ -41,6 +41,21 @@
 
 namespace auth {
 
+namespace meta {
+
+namespace role_members_table {
+
+constexpr std::string_view name{"role_members" , 12};
+
+}
+
+namespace role_attributes_table {
+
+constexpr std::string_view name{"role_attributes", 15};
+
+}
+
+}
 
 static logging::logger log("standard_role_manager");
 
@@ -49,8 +64,7 @@ static const class_registrator<
         standard_role_manager,
         cql3::query_processor&,
         ::service::raft_group0_client&,
-        ::service::migration_manager&,
-        cache&> registration("org.apache.cassandra.auth.CassandraRoleManager");
+        ::service::migration_manager&> registration("org.apache.cassandra.auth.CassandraRoleManager");
 
 struct record final {
     sstring name;
@@ -107,11 +121,10 @@ static bool has_can_login(const cql3::untyped_result_set_row& row) {
     return row.has("can_login") && !(boolean_type->deserialize(row.get_blob_unfragmented("can_login")).is_null());
 }
 
-standard_role_manager::standard_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, cache& cache)
+standard_role_manager::standard_role_manager(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm)
     : _qp(qp)
     , _group0_client(g0)
     , _migration_manager(mm)
-    , _cache(cache)
     , _stopped(make_ready_future<>())
     , _superuser(password_authenticator::default_superuser(qp.db().get_config()))
 {}
@@ -123,7 +136,7 @@ std::string_view standard_role_manager::qualified_java_name() const noexcept {
 const resource_set& standard_role_manager::protected_resources() const {
     static const resource_set resources({
             make_data_resource(meta::legacy::AUTH_KS, meta::roles_table::name),
-            make_data_resource(meta::legacy::AUTH_KS, ROLE_MEMBERS_CF)});
+            make_data_resource(meta::legacy::AUTH_KS, meta::role_members_table::name)});
 
     return resources;
 }
@@ -147,7 +160,7 @@ future<> standard_role_manager::create_legacy_metadata_tables_if_missing() const
             "  PRIMARY KEY (role, member)"
             ")",
             meta::legacy::AUTH_KS,
-            ROLE_MEMBERS_CF);
+            meta::role_members_table::name);
     static const sstring create_role_attributes_query = seastar::format(
             "CREATE TABLE {}.{} ("
             "  role text,"
@@ -156,7 +169,7 @@ future<> standard_role_manager::create_legacy_metadata_tables_if_missing() const
             "  PRIMARY KEY(role, name)"
             ")",
             meta::legacy::AUTH_KS,
-            ROLE_ATTRIBUTES_CF);
+            meta::role_attributes_table::name);
     return when_all_succeed(
             create_legacy_metadata_table_if_missing(
                     meta::roles_table::name,
@@ -164,12 +177,12 @@ future<> standard_role_manager::create_legacy_metadata_tables_if_missing() const
                     create_roles_query,
                     _migration_manager),
             create_legacy_metadata_table_if_missing(
-                    ROLE_MEMBERS_CF,
+                    meta::role_members_table::name,
                     _qp,
                     create_role_members_query,
                     _migration_manager),
             create_legacy_metadata_table_if_missing(
-                    ROLE_ATTRIBUTES_CF,
+                    meta::role_attributes_table::name,
                     _qp,
                     create_role_attributes_query,
                     _migration_manager)).discard_result();
@@ -192,7 +205,7 @@ future<> standard_role_manager::legacy_create_default_role_if_missing() {
                 {_superuser},
                 cql3::query_processor::cache_internal::no).discard_result();
         log.info("Created default superuser role '{}'.", _superuser);
-    } catch (const exceptions::unavailable_exception& e) {
+    } catch(const exceptions::unavailable_exception& e) {
         log.warn("Skipped default role setup: some nodes were not ready; will retry");
         throw e;
     }
@@ -416,7 +429,7 @@ future<> standard_role_manager::drop(std::string_view role_name, ::service::grou
     const auto revoke_from_members = [this, role_name, &mc] () -> future<> {
         const sstring query = seastar::format("SELECT member FROM {}.{} WHERE role = ?",
                 get_auth_ks_name(_qp),
-                ROLE_MEMBERS_CF);
+                meta::role_members_table::name);
         const auto members = co_await _qp.execute_internal(
                 query,
                 consistency_for_role(role_name),
@@ -448,7 +461,7 @@ future<> standard_role_manager::drop(std::string_view role_name, ::service::grou
     const auto remove_attributes_of = [this, role_name, &mc] () -> future<> {
         const sstring query = seastar::format("DELETE FROM {}.{} WHERE role = ?",
                 get_auth_ks_name(_qp),
-                ROLE_ATTRIBUTES_CF);
+                meta::role_attributes_table::name);
         if (legacy_mode(_qp)) {
             co_await _qp.execute_internal(query, {sstring(role_name)},
                 cql3::query_processor::cache_internal::yes).discard_result();
@@ -504,7 +517,7 @@ standard_role_manager::legacy_modify_membership(
             case membership_change::add: {
                 const sstring insert_query = seastar::format("INSERT INTO {}.{} (role, member) VALUES (?, ?)",
                         get_auth_ks_name(_qp),
-                        ROLE_MEMBERS_CF);
+                        meta::role_members_table::name);
                 co_return co_await _qp.execute_internal(
                         insert_query,
                         consistency_for_role(role_name),
@@ -516,7 +529,7 @@ standard_role_manager::legacy_modify_membership(
             case membership_change::remove: {
                 const sstring delete_query = seastar::format("DELETE FROM {}.{} WHERE role = ? AND member = ?",
                         get_auth_ks_name(_qp),
-                        ROLE_MEMBERS_CF);
+                        meta::role_members_table::name);
                 co_return co_await _qp.execute_internal(
                         delete_query,
                         consistency_for_role(role_name),
@@ -554,12 +567,12 @@ standard_role_manager::modify_membership(
     case membership_change::add:
         modify_role_members = seastar::format("INSERT INTO {}.{} (role, member) VALUES (?, ?)",
                 get_auth_ks_name(_qp),
-                ROLE_MEMBERS_CF);
+                meta::role_members_table::name);
         break;
     case membership_change::remove:
         modify_role_members = seastar::format("DELETE FROM {}.{} WHERE role = ? AND member = ?",
                 get_auth_ks_name(_qp),
-                ROLE_MEMBERS_CF);
+                meta::role_members_table::name);
         break;
     default:
         on_internal_error(log, format("unknown membership_change value: {}", int(ch)));
@@ -653,7 +666,7 @@ future<role_set> standard_role_manager::query_granted(std::string_view grantee_n
 future<role_to_directly_granted_map> standard_role_manager::query_all_directly_granted(::service::query_state& qs) {
     const sstring query = seastar::format("SELECT * FROM {}.{}",
             get_auth_ks_name(_qp),
-            ROLE_MEMBERS_CF);
+            meta::role_members_table::name);
 
     const auto results = co_await _qp.execute_internal(
             query,
@@ -718,21 +731,15 @@ future<bool> standard_role_manager::is_superuser(std::string_view role_name) {
 }
 
 future<bool> standard_role_manager::can_login(std::string_view role_name) {
-    if (legacy_mode(_qp)) {
-       const auto r = co_await require_record(_qp, role_name);
-       co_return r.can_login;
-    }
-    auto role = _cache.get(sstring(role_name));
-    if (!role) {
-        throw nonexistant_role(role_name);
-    }
-    co_return role->can_login;
+    return require_record(_qp, role_name).then([](record r) {
+        return r.can_login;
+    });
 }
 
 future<std::optional<sstring>> standard_role_manager::get_attribute(std::string_view role_name, std::string_view attribute_name, ::service::query_state& qs) {
     const sstring query = seastar::format("SELECT name, value FROM {}.{} WHERE role = ? AND name = ?",
             get_auth_ks_name(_qp),
-            ROLE_ATTRIBUTES_CF);
+            meta::role_attributes_table::name);
     const auto result_set = co_await _qp.execute_internal(query, db::consistency_level::ONE, qs, {sstring(role_name), sstring(attribute_name)}, cql3::query_processor::cache_internal::yes);
     if (!result_set->empty()) {
         const cql3::untyped_result_set_row &row = result_set->one();
@@ -763,7 +770,7 @@ future<> standard_role_manager::set_attribute(std::string_view role_name, std::s
     }
     const sstring query = seastar::format("INSERT INTO {}.{} (role, name, value)  VALUES (?, ?, ?)",
             get_auth_ks_name(_qp),
-            ROLE_ATTRIBUTES_CF);
+            meta::role_attributes_table::name);
     if (legacy_mode(_qp)) {
         co_await _qp.execute_internal(query, {sstring(role_name), sstring(attribute_name), sstring(attribute_value)}, cql3::query_processor::cache_internal::yes).discard_result();
     } else {
@@ -778,7 +785,7 @@ future<> standard_role_manager::remove_attribute(std::string_view role_name, std
     }
     const sstring query = seastar::format("DELETE FROM {}.{} WHERE role = ? AND name = ?",
             get_auth_ks_name(_qp),
-            ROLE_ATTRIBUTES_CF);
+            meta::role_attributes_table::name);
     if (legacy_mode(_qp)) {
         co_await _qp.execute_internal(query, {sstring(role_name), sstring(attribute_name)}, cql3::query_processor::cache_internal::yes).discard_result();
     } else {

auth/standard_role_manager.hh

@@ -10,7 +10,6 @@
 
 #include "auth/common.hh"
 #include "auth/role_manager.hh"
-#include "auth/cache.hh"
 
 #include <string_view>
 
@@ -37,14 +36,13 @@ class standard_role_manager final : public role_manager {
     cql3::query_processor& _qp;
     ::service::raft_group0_client& _group0_client;
     ::service::migration_manager& _migration_manager;
-    cache& _cache;
     future<> _stopped;
     abort_source _as;
     std::string _superuser;
     shared_promise<> _superuser_created_promise;
 
 public:
-    standard_role_manager(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&, cache&);
+    standard_role_manager(cql3::query_processor&, ::service::raft_group0_client&, ::service::migration_manager&);
 
     virtual std::string_view qualified_java_name() const noexcept override;
 

auth/transitional.cc

@@ -13,7 +13,6 @@
 #include "auth/authorizer.hh"
 #include "auth/default_authorizer.hh"
 #include "auth/password_authenticator.hh"
-#include "auth/cache.hh"
 #include "auth/permission.hh"
 #include "service/raft/raft_group0_client.hh"
 #include "utils/class_registrator.hh"
@@ -38,8 +37,8 @@ class transitional_authenticator : public authenticator {
 public:
     static const sstring PASSWORD_AUTHENTICATOR_NAME;
 
-    transitional_authenticator(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, cache& cache)
-            : transitional_authenticator(std::make_unique<password_authenticator>(qp, g0, mm, cache)) {
+    transitional_authenticator(cql3::query_processor& qp, ::service::raft_group0_client& g0, ::service::migration_manager& mm, utils::alien_worker& hashing_worker)
+            : transitional_authenticator(std::make_unique<password_authenticator>(qp, g0, mm, hashing_worker)) {
     }
     transitional_authenticator(std::unique_ptr<authenticator> a)
             : _authenticator(std::move(a)) {
@@ -81,7 +80,7 @@ public:
         }).handle_exception([](auto ep) {
             try {
                 std::rethrow_exception(ep);
-            } catch (const exceptions::authentication_exception&) {
+            } catch (exceptions::authentication_exception&) {
                 // return anon user
                 return make_ready_future<authenticated_user>(anonymous_user());
             }
@@ -126,7 +125,7 @@ public:
             virtual bytes evaluate_response(bytes_view client_response) override {
                 try {
                     return _sasl->evaluate_response(client_response);
-                } catch (const exceptions::authentication_exception&) {
+                } catch (exceptions::authentication_exception&) {
                     _complete = true;
                     return {};
                 }
@@ -141,7 +140,7 @@ public:
                     return _sasl->get_authenticated_user().handle_exception([](auto ep) {
                         try {
                             std::rethrow_exception(ep);
-                        } catch (const exceptions::authentication_exception&) {
+                        } catch (exceptions::authentication_exception&) {
                             // return anon user
                             return make_ready_future<authenticated_user>(anonymous_user());
                         }
@@ -241,7 +240,7 @@ static const class_registrator<
         cql3::query_processor&,
         ::service::raft_group0_client&,
         ::service::migration_manager&,
-        auth::cache&> transitional_authenticator_reg(auth::PACKAGE_NAME + "TransitionalAuthenticator");
+        utils::alien_worker&> transitional_authenticator_reg(auth::PACKAGE_NAME + "TransitionalAuthenticator");
 
 static const class_registrator<
         auth::authorizer,

backlog_controller.hh

@@ -15,7 +15,6 @@
 #include <cmath>
 
 #include "seastarx.hh"
-#include "backlog_controller_fwd.hh"
 
 // Simple proportional controller to adjust shares for processes for which a backlog can be clearly
 // defined.
@@ -129,21 +128,11 @@ public:
     static constexpr unsigned normalization_factor = 30;
     static constexpr float disable_backlog = std::numeric_limits<double>::infinity();
     static constexpr float backlog_disabled(float backlog) { return std::isinf(backlog); }
-    static inline const std::vector<backlog_controller::control_point> default_control_points = {
-            backlog_controller::control_point{0.0, 50}, {1.5, 100}, {normalization_factor, default_compaction_maximum_shares}};
-    compaction_controller(backlog_controller::scheduling_group sg, float static_shares, std::optional<float> max_shares,
-        std::chrono::milliseconds interval, std::function<float()> current_backlog)
+    compaction_controller(backlog_controller::scheduling_group sg, float static_shares, std::chrono::milliseconds interval, std::function<float()> current_backlog)
         : backlog_controller(std::move(sg), std::move(interval),
-          default_control_points,
+          std::vector<backlog_controller::control_point>({{0.0, 50}, {1.5, 100} , {normalization_factor, 1000}}),
           std::move(current_backlog),
           static_shares
         )
-    {
-        if (max_shares) {
-            set_max_shares(*max_shares);
-        }
-    }
-
-    // Updates the maximum output value for control points.
-    void set_max_shares(float max_shares);
+    {}
 };

backlog_controller_fwd.hh

@@ -1,13 +0,0 @@
-/*
- * Copyright (C) 2025-present ScyllaDB
- */
-
-/*
- * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
- */
-
-#pragma once
-
-#include <cstdint>
-
-static constexpr uint64_t default_compaction_maximum_shares = 1000;

cdc/CMakeLists.txt

@@ -17,8 +17,5 @@ target_link_libraries(cdc
   PRIVATE
     replica)
 
-if (Scylla_USE_PRECOMPILED_HEADER_USE)
-  target_precompile_headers(cdc REUSE_FROM scylla-precompiled-header)
-endif()
 check_headers(check-headers cdc
   GLOB_RECURSE ${CMAKE_CURRENT_SOURCE_DIR}/*.hh)

cdc/generation.cc

@@ -204,7 +204,7 @@ future<topology_description> topology_description::clone_async() const {
 
     for (const auto& entry : _entries) {
         vec.push_back(entry);
-        co_await coroutine::maybe_yield();
+        co_await seastar::maybe_yield();
     }
 
     co_return topology_description{std::move(vec)};

cdc/log.cc

@@ -23,9 +23,7 @@
 #include "bytes.hh"
 #include "index/vector_index.hh"
 #include "locator/abstract_replication_strategy.hh"
-#include "locator/topology.hh"
 #include "replica/database.hh"
-#include "db/config.hh"
 #include "db/schema_tables.hh"
 #include "gms/feature_service.hh"
 #include "schema/schema.hh"
@@ -64,9 +62,9 @@ logging::logger cdc_log("cdc");
 
 namespace {
 
-shared_ptr<locator::abstract_replication_strategy> generate_replication_strategy(const keyspace_metadata& ksm, const locator::topology& topo) {
-    locator::replication_strategy_params params(ksm.strategy_options(), ksm.initial_tablets(), ksm.consistency_option());
-    return locator::abstract_replication_strategy::create_replication_strategy(ksm.strategy_name(), params, topo);
+shared_ptr<locator::abstract_replication_strategy> generate_replication_strategy(const keyspace_metadata& ksm) {
+    locator::replication_strategy_params params(ksm.strategy_options(), ksm.initial_tablets());
+    return locator::abstract_replication_strategy::create_replication_strategy(ksm.strategy_name(), params);
 }
 
 // When dropping a column from a CDC log table, we set the drop timestamp
@@ -204,7 +202,7 @@ public:
             check_that_cdc_log_table_does_not_exist(db, schema, logname);
             ensure_that_table_has_no_counter_columns(schema);
             if (!db.features().cdc_with_tablets) {
-                ensure_that_table_uses_vnodes(ksm, schema, db.get_token_metadata().get_topology());
+                ensure_that_table_uses_vnodes(ksm, schema);
             }
 
             // in seastar thread
@@ -251,7 +249,7 @@ public:
             check_for_attempt_to_create_nested_cdc_log(db, new_schema);
             ensure_that_table_has_no_counter_columns(new_schema);
             if (!db.features().cdc_with_tablets) {
-                ensure_that_table_uses_vnodes(*keyspace.metadata(), new_schema, db.get_token_metadata().get_topology());
+                ensure_that_table_uses_vnodes(*keyspace.metadata(), new_schema);
             }
 
             std::optional<table_id> maybe_id = log_schema ? std::make_optional(log_schema->id()) : std::nullopt;
@@ -318,8 +316,7 @@ public:
         lowres_clock::time_point timeout,
         utils::chunked_vector<mutation>&& mutations,
         tracing::trace_state_ptr tr_state,
-        db::consistency_level write_cl,
-        per_request_options options
+        db::consistency_level write_cl
     );
 
     template<typename Iter>
@@ -353,8 +350,8 @@ private:
     // Until we support CDC with tablets (issue #16317), we can't allow this
     // to be attempted - in particular the log table we try to create will not
     // have tablets, and will cause a failure.
-    static void ensure_that_table_uses_vnodes(const keyspace_metadata& ksm, const schema& schema, const locator::topology& topo) {
-        auto rs = generate_replication_strategy(ksm, topo);
+    static void ensure_that_table_uses_vnodes(const keyspace_metadata& ksm, const schema& schema) {
+        auto rs = generate_replication_strategy(ksm);
         if (rs->uses_tablets()) {
             throw exceptions::invalid_request_exception(format("Cannot create CDC log for a table {}.{}, because the keyspace uses tablets, and not all nodes support the CDC with tablets feature.",
                 schema.ks_name(), schema.cf_name()));
@@ -587,9 +584,11 @@ bytes log_data_column_deleted_elements_name_bytes(const bytes& column_name) {
     return to_bytes(cdc_deleted_elements_column_prefix) + column_name;
 }
 
-static void set_default_properties_log_table(schema_builder& b, const schema& s,
-        const replica::database& db, const keyspace_metadata& ksm)
+static schema_ptr create_log_schema(const schema& s, const replica::database& db,
+        const keyspace_metadata& ksm, api::timestamp_type timestamp, std::optional<table_id> uuid, schema_ptr old)
 {
+    schema_builder b(s.ks_name(), log_name(s.cf_name()));
+    b.with_partitioner(cdc::cdc_partitioner::classname);
     b.set_compaction_strategy(compaction::compaction_strategy_type::time_window);
     b.set_comment(fmt::format("CDC log for {}.{}", s.ks_name(), s.cf_name()));
     auto ttl_seconds = s.cdc_options().ttl();
@@ -615,22 +614,13 @@ static void set_default_properties_log_table(schema_builder& b, const schema& s,
                         std::to_string(std::max(1, window_seconds / 2))},
         });
     }
-    b.set_caching_options(caching_options::get_disabled_caching_options());
-
-    auto rs = generate_replication_strategy(ksm, db.get_token_metadata().get_topology());
-    auto tombstone_gc_ext = seastar::make_shared<tombstone_gc_extension>(get_default_tombstone_gc_mode(*rs, db.get_token_metadata(), false));
-    b.add_extension(tombstone_gc_extension::NAME, std::move(tombstone_gc_ext));
-}
-
-static void add_columns_to_cdc_log(schema_builder& b, const schema& s,
-        const api::timestamp_type timestamp, const schema_ptr old)
-{
     b.with_column(log_meta_column_name_bytes("stream_id"), bytes_type, column_kind::partition_key);
     b.with_column(log_meta_column_name_bytes("time"), timeuuid_type, column_kind::clustering_key);
     b.with_column(log_meta_column_name_bytes("batch_seq_no"), int32_type, column_kind::clustering_key);
     b.with_column(log_meta_column_name_bytes("operation"), data_type_for<operation_native_type>());
     b.with_column(log_meta_column_name_bytes("ttl"), long_type);
     b.with_column(log_meta_column_name_bytes("end_of_batch"), boolean_type);
+    b.set_caching_options(caching_options::get_disabled_caching_options());
 
     auto validate_new_column = [&] (const sstring& name) {
         // When dropping a column from a CDC log table, we set the drop timestamp to be
@@ -700,28 +690,15 @@ static void add_columns_to_cdc_log(schema_builder& b, const schema& s,
     add_columns(s.clustering_key_columns());
     add_columns(s.static_columns(), true);
     add_columns(s.regular_columns(), true);
-}
-
-static schema_ptr create_log_schema(const schema& s, const replica::database& db,
-        const keyspace_metadata& ksm, api::timestamp_type timestamp, std::optional<table_id> uuid, schema_ptr old)
-{
-    schema_builder b(s.ks_name(), log_name(s.cf_name()));
-
-    b.with_partitioner(cdc::cdc_partitioner::classname);
-
-    if (old) {
-        // If the user reattaches the log table, do not change its properties.
-        b.set_properties(old->get_properties());
-    } else {
-        set_default_properties_log_table(b, s, db, ksm);
-    }
-
-    add_columns_to_cdc_log(b, s, timestamp, old);
 
     if (uuid) {
         b.set_uuid(*uuid);
     }
 
+    auto rs = generate_replication_strategy(ksm);
+    auto tombstone_gc_ext = seastar::make_shared<tombstone_gc_extension>(get_default_tombstone_gc_mode(*rs, db.get_token_metadata(), false));
+    b.add_extension(tombstone_gc_extension::NAME, std::move(tombstone_gc_ext));
+
     /**
      * #10473 - if we are redefining the log table, we need to ensure any dropped
      * columns are registered in "dropped_columns" table, otherwise clients will not
@@ -952,6 +929,9 @@ static managed_bytes merge(const abstract_type& type, const managed_bytes_opt& p
     throw std::runtime_error(format("cdc merge: unknown type {}", type.name()));
 }
 
+using cell_map = std::unordered_map<const column_definition*, managed_bytes_opt>;
+using row_states_map = std::unordered_map<clustering_key, cell_map, clustering_key::hashing, clustering_key::equality>;
+
 static managed_bytes_opt get_col_from_row_state(const cell_map* state, const column_definition& cdef) {
     if (state) {
         if (auto it = state->find(&cdef); it != state->end()) {
@@ -961,12 +941,7 @@ static managed_bytes_opt get_col_from_row_state(const cell_map* state, const col
     return std::nullopt;
 }
 
-cell_map* get_row_state(row_states_map& row_states, const clustering_key& ck) {
-    auto it = row_states.find(ck);
-    return it == row_states.end() ? nullptr : &it->second;
-}
-
-const cell_map* get_row_state(const row_states_map& row_states, const clustering_key& ck) {
+static cell_map* get_row_state(row_states_map& row_states, const clustering_key& ck) {
     auto it = row_states.find(ck);
     return it == row_states.end() ? nullptr : &it->second;
 }
@@ -1419,13 +1394,6 @@ struct process_row_visitor {
 };
 
 struct process_change_visitor {
-    const per_request_options& _request_options;
-    // The types of the operations used for row / partition deletes. Introduced
-    // to differentiate service operations (e.g. operation::service_row_delete
-    // vs operation::row_delete).
-    const operation _row_delete_op = operation::row_delete;
-    const operation _partition_delete_op = operation::partition_delete;
-
     stats::part_type_set& _touched_parts;
 
     log_mutation_builder& _builder;
@@ -1436,8 +1404,6 @@ struct process_change_visitor {
     row_states_map& _clustering_row_states;
     cell_map& _static_row_state;
 
-    const bool _is_update = false;
-
     const bool _generate_delta_values = true;
 
     void static_row_cells(auto&& visit_row_cells) {
@@ -1461,13 +1427,12 @@ struct process_change_visitor {
 
         struct clustering_row_cells_visitor : public process_row_visitor {
             operation _cdc_op = operation::update;
-            operation _marker_op = operation::insert;
 
             using process_row_visitor::process_row_visitor;
 
             void marker(const row_marker& rm) {
                 _ttl_column = get_ttl(rm);
-                _cdc_op = _marker_op;
+                _cdc_op = operation::insert;
             }
         };
 
@@ -1475,9 +1440,6 @@ struct process_change_visitor {
                 log_ck, _touched_parts, _builder,
                 _enable_updating_state, &ckey, get_row_state(_clustering_row_states, ckey),
                 _clustering_row_states, _generate_delta_values);
-        if (_is_update && _request_options.alternator) {
-            v._marker_op = operation::update;
-        }
         visit_row_cells(v);
 
         if (_enable_updating_state) {
@@ -1494,7 +1456,7 @@ struct process_change_visitor {
     void clustered_row_delete(const clustering_key& ckey, const tombstone&) {
         _touched_parts.set<stats::part_type::ROW_DELETE>();
 
-        auto log_ck = _builder.allocate_new_log_row(_row_delete_op);
+        auto log_ck = _builder.allocate_new_log_row(operation::row_delete);
         _builder.set_clustering_columns(log_ck, ckey);
 
         if (_enable_updating_state && get_row_state(_clustering_row_states, ckey)) {
@@ -1538,7 +1500,7 @@ struct process_change_visitor {
 
     void partition_delete(const tombstone&) {
         _touched_parts.set<stats::part_type::PARTITION_DELETE>();
-        auto log_ck = _builder.allocate_new_log_row(_partition_delete_op);
+        auto log_ck = _builder.allocate_new_log_row(operation::partition_delete);
         if (_enable_updating_state) {
             _clustering_row_states.clear();
         }
@@ -1553,7 +1515,6 @@ private:
     schema_ptr _schema;
     dht::decorated_key _dk;
     schema_ptr _log_schema;
-    const per_request_options& _options;
 
     /**
      * #6070, #6084
@@ -1631,11 +1592,6 @@ private:
 
     row_states_map _clustering_row_states;
     cell_map _static_row_state;
-    // True if the mutated row existed before applying the mutation. In other
-    // words, if the preimage is enabled and it isn't empty (otherwise, we
-    // assume that the row is non-existent). Used for Alternator Streams (see
-    // #6918).
-    bool _is_update = false;
 
     const bool _uses_tablets;
 
@@ -1648,12 +1604,11 @@ private:
     stats::part_type_set _touched_parts;
 
 public:
-    transformer(db_context ctx, schema_ptr s, dht::decorated_key dk, const per_request_options& options)
+    transformer(db_context ctx, schema_ptr s, dht::decorated_key dk)
         : _ctx(ctx)
         , _schema(std::move(s))
         , _dk(std::move(dk))
-        , _log_schema(_schema->cdc_schema() ? _schema->cdc_schema() : ctx._proxy.get_db().local().find_schema(_schema->ks_name(), log_name(_schema->cf_name())))
-        , _options(options)
+        , _log_schema(ctx._proxy.get_db().local().find_schema(_schema->ks_name(), log_name(_schema->cf_name())))
         , _clustering_row_states(0, clustering_key::hashing(*_schema), clustering_key::equality(*_schema))
         , _uses_tablets(ctx._proxy.get_db().local().find_keyspace(_schema->ks_name()).uses_tablets())
     {
@@ -1668,7 +1623,7 @@ public:
     }
 
     void produce_preimage(const clustering_key* ck, const one_kind_column_set& columns_to_include) override {
-        // if we want full preimage, just ignore the affected columns and include everything. 
+        // iff we want full preimage, just ignore the affected columns and include everything. 
         generate_image(operation::pre_image, ck, _schema->cdc_options().full_preimage() ? nullptr : &columns_to_include);
     };
 
@@ -1754,15 +1709,11 @@ public:
     void process_change(const mutation& m) override {
         SCYLLA_ASSERT(_builder);
         process_change_visitor v {
-            ._request_options = _options,
-            ._row_delete_op = _options.is_system_originated ? operation::service_row_delete : operation::row_delete,
-            ._partition_delete_op = _options.is_system_originated ? operation::service_partition_delete : operation::partition_delete,
             ._touched_parts = _touched_parts,
             ._builder = *_builder,
             ._enable_updating_state = _enable_updating_state,
             ._clustering_row_states = _clustering_row_states,
             ._static_row_state = _static_row_state,
-            ._is_update = _is_update,
             ._generate_delta_values = generate_delta_values(_builder->base_schema())
         };
         cdc::inspect_mutation(m, v);
@@ -1773,10 +1724,6 @@ public:
         _builder->end_record();
     }
 
-    const row_states_map& clustering_row_states() const override {
-        return _clustering_row_states;
-    }
-
     // Takes and returns generated cdc log mutations and associated statistics about parts touched during transformer's lifetime.
     // The `transformer` object on which this method was called on should not be used anymore.
     std::tuple<utils::chunked_vector<mutation>, stats::part_type_set> finish() && {
@@ -1793,8 +1740,7 @@ public:
             const mutation& m)
     {
         auto& p = m.partition();
-        const bool no_ck_schema_partition_deletion = m.schema()->clustering_key_size() == 0 && bool(p.partition_tombstone());
-        if (p.clustered_rows().empty() && p.static_row().empty() && !no_ck_schema_partition_deletion) {
+        if (p.clustered_rows().empty() && p.static_row().empty()) {
             return make_ready_future<lw_shared_ptr<cql3::untyped_result_set>>();
         }
 
@@ -1843,12 +1789,12 @@ public:
                 });
             }
         }
-        if (!p.clustered_rows().empty() || no_ck_schema_partition_deletion) {
+        if (!p.clustered_rows().empty()) {
             const bool has_row_delete = std::any_of(p.clustered_rows().begin(), p.clustered_rows().end(), [] (const rows_entry& re) {
                 return re.row().deleted_at();
             });
             // for postimage we need everything...
-            if (has_row_delete || _schema->cdc_options().postimage() || _schema->cdc_options().full_preimage() || no_ck_schema_partition_deletion) {
+            if (has_row_delete || _schema->cdc_options().postimage() || _schema->cdc_options().full_preimage()) {
                 for (const column_definition& c: _schema->regular_columns()) {
                     regular_columns.emplace_back(c.id);
                     columns.emplace_back(&c);
@@ -1900,7 +1846,6 @@ public:
                     _static_row_state[&c] = std::move(*maybe_cell_view);
                 }
             }
-            _is_update = true;
         }
 
         if (static_only) {
@@ -1964,7 +1909,7 @@ transform_mutations(utils::chunked_vector<mutation>& muts, decltype(muts.size())
 } // namespace cdc
 
 future<std::tuple<utils::chunked_vector<mutation>, lw_shared_ptr<cdc::operation_result_tracker>>>
-cdc::cdc_service::impl::augment_mutation_call(lowres_clock::time_point timeout, utils::chunked_vector<mutation>&& mutations, tracing::trace_state_ptr tr_state, db::consistency_level write_cl, per_request_options options) {
+cdc::cdc_service::impl::augment_mutation_call(lowres_clock::time_point timeout, utils::chunked_vector<mutation>&& mutations, tracing::trace_state_ptr tr_state, db::consistency_level write_cl) {
     // we do all this because in the case of batches, we can have mixed schemas.
     auto e = mutations.end();
     auto i = std::find_if(mutations.begin(), e, [](const mutation& m) {
@@ -1978,9 +1923,9 @@ cdc::cdc_service::impl::augment_mutation_call(lowres_clock::time_point timeout,
     tracing::trace(tr_state, "CDC: Started generating mutations for log rows");
     mutations.reserve(2 * mutations.size());
 
-    return do_with(std::move(mutations), service::query_state(service::client_state::for_internal_calls(), empty_service_permit()), operation_details{}, std::move(options),
-            [this, tr_state = std::move(tr_state), write_cl] (utils::chunked_vector<mutation>& mutations, service::query_state& qs, operation_details& details, per_request_options& options) {
-        return transform_mutations(mutations, 1, [this, &mutations, &qs, tr_state = tr_state, &details, write_cl, &options] (int idx) mutable {
+    return do_with(std::move(mutations), service::query_state(service::client_state::for_internal_calls(), empty_service_permit()), operation_details{},
+            [this, tr_state = std::move(tr_state), write_cl] (utils::chunked_vector<mutation>& mutations, service::query_state& qs, operation_details& details) {
+        return transform_mutations(mutations, 1, [this, &mutations, &qs, tr_state = tr_state, &details, write_cl] (int idx) mutable {
             auto& m = mutations[idx];
             auto s = m.schema();
 
@@ -1988,17 +1933,12 @@ cdc::cdc_service::impl::augment_mutation_call(lowres_clock::time_point timeout,
                 return make_ready_future<>();
             }
 
-            const bool alternator_increased_compatibility = options.alternator && options.alternator_streams_increased_compatibility;
-            transformer trans(_ctxt, s, m.decorated_key(), options);
+            transformer trans(_ctxt, s, m.decorated_key());
 
             auto f = make_ready_future<lw_shared_ptr<cql3::untyped_result_set>>(nullptr);
-            if (options.preimage && !options.preimage->empty()) {
-                // Preimage has been fetched by upper layers.
-                tracing::trace(tr_state, "CDC: Using a prefetched preimage");
-                f = make_ready_future<lw_shared_ptr<cql3::untyped_result_set>>(options.preimage);
-            } else if (s->cdc_options().preimage() || s->cdc_options().postimage() || alternator_increased_compatibility) {
+            if (s->cdc_options().preimage() || s->cdc_options().postimage()) {
                 // Note: further improvement here would be to coalesce the pre-image selects into one
-                // if a batch contains several modifications to the same table. Otoh, batch is rare(?)
+                // iff a batch contains several modifications to the same table. Otoh, batch is rare(?)
                 // so this is premature.
                 tracing::trace(tr_state, "CDC: Selecting preimage for {}", m.decorated_key());
                 f = trans.pre_image_select(qs.get_client_state(), write_cl, m).then_wrapped([this] (future<lw_shared_ptr<cql3::untyped_result_set>> f) {
@@ -2013,7 +1953,7 @@ cdc::cdc_service::impl::augment_mutation_call(lowres_clock::time_point timeout,
                 tracing::trace(tr_state, "CDC: Preimage not enabled for the table, not querying current value of {}", m.decorated_key());
             }
 
-            return f.then([alternator_increased_compatibility, trans = std::move(trans), &mutations, idx, tr_state, &details, &options] (lw_shared_ptr<cql3::untyped_result_set> rs) mutable {
+            return f.then([trans = std::move(trans), &mutations, idx, tr_state, &details] (lw_shared_ptr<cql3::untyped_result_set> rs) mutable {
                 auto& m = mutations[idx];
                 auto& s = m.schema();
 
@@ -2028,13 +1968,13 @@ cdc::cdc_service::impl::augment_mutation_call(lowres_clock::time_point timeout,
                 details.had_preimage |= preimage;
                 details.had_postimage |= postimage;
                 tracing::trace(tr_state, "CDC: Generating log mutations for {}", m.decorated_key());
-                if (should_split(m, options)) {
+                if (should_split(m)) {
                     tracing::trace(tr_state, "CDC: Splitting {}", m.decorated_key());
                     details.was_split = true;
-                    process_changes_with_splitting(m, trans, preimage, postimage, alternator_increased_compatibility);
+                    process_changes_with_splitting(m, trans, preimage, postimage);
                 } else {
                     tracing::trace(tr_state, "CDC: No need to split {}", m.decorated_key());
-                    process_changes_without_splitting(m, trans, preimage, postimage, alternator_increased_compatibility);
+                    process_changes_without_splitting(m, trans, preimage, postimage);
                 }
                 auto [log_mut, touched_parts] = std::move(trans).finish();
                 const int generated_count = log_mut.size();
@@ -2059,11 +1999,11 @@ bool cdc::cdc_service::needs_cdc_augmentation(const utils::chunked_vector<mutati
 }
 
 future<std::tuple<utils::chunked_vector<mutation>, lw_shared_ptr<cdc::operation_result_tracker>>>
-cdc::cdc_service::augment_mutation_call(lowres_clock::time_point timeout, utils::chunked_vector<mutation>&& mutations, tracing::trace_state_ptr tr_state, db::consistency_level write_cl, per_request_options options) {
+cdc::cdc_service::augment_mutation_call(lowres_clock::time_point timeout, utils::chunked_vector<mutation>&& mutations, tracing::trace_state_ptr tr_state, db::consistency_level write_cl) {
     if (utils::get_local_injector().enter("sleep_before_cdc_augmentation")) {
-        return seastar::sleep(std::chrono::milliseconds(100)).then([this, timeout, mutations = std::move(mutations), tr_state = std::move(tr_state), write_cl, options = std::move(options)] () mutable {
-            return _impl->augment_mutation_call(timeout, std::move(mutations), std::move(tr_state), write_cl, std::move(options));
+        return seastar::sleep(std::chrono::milliseconds(100)).then([this, timeout, mutations = std::move(mutations), tr_state = std::move(tr_state), write_cl] () mutable {
+            return _impl->augment_mutation_call(timeout, std::move(mutations), std::move(tr_state), write_cl);
         });
     }
-    return _impl->augment_mutation_call(timeout, std::move(mutations), std::move(tr_state), write_cl, std::move(options));
+    return _impl->augment_mutation_call(timeout, std::move(mutations), std::move(tr_state), write_cl);
 }

cdc/log.hh

@@ -21,7 +21,6 @@
 #include <seastar/core/shared_ptr.hh>
 #include <seastar/core/sstring.hh>
 
-#include "cql3/untyped_result_set.hh"
 #include "mutation/timestamp.hh"
 #include "tracing/trace_state.hh"
 #include "utils/UUID.hh"
@@ -52,40 +51,6 @@ class database;
 
 namespace cdc {
 
-using cell_map = std::unordered_map<const column_definition*, managed_bytes_opt>;
-using row_states_map = std::unordered_map<clustering_key, cell_map, clustering_key::hashing, clustering_key::equality>;
-
-// cdc log table operation
-enum class operation : int8_t {
-    // note: these values will eventually be read by a third party, probably not privvy to this
-    // enum decl, so don't change the constant values (or the datatype).
-    pre_image = 0, update = 1, insert = 2, row_delete = 3, partition_delete = 4,
-    range_delete_start_inclusive = 5, range_delete_start_exclusive = 6, range_delete_end_inclusive = 7, range_delete_end_exclusive = 8,
-    post_image = 9,
-
-    // Operations initiated internally by Scylla. Currently used only by Alternator
-    service_row_delete = -3, service_partition_delete = -4,
-};
-
-struct per_request_options {
-    // The value of the base row before current operation, queried by higher
-    // layers than CDC. We assume that CDC could have seen the row in this
-    // state, i.e. the value isn't 'stale'/'too recent'.
-    lw_shared_ptr<cql3::untyped_result_set> preimage;
-    // Whether this mutation is a result of an internal operation initiated by
-    // Scylla. Currently, only TTL expiration implementation for Alternator
-    // uses this.
-    const bool is_system_originated = false;
-    // True if this mutation was emitted by Alternator.
-    const bool alternator = false;
-    // Sacrifice performance for the sake of better compatibility with DynamoDB
-    // Streams. It's important for correctness that
-    // alternator_streams_increased_compatibility config flag be read once per
-    // request, because it's live-updateable. As a result, the flag may change
-    // between reads.
-    const bool alternator_streams_increased_compatibility = false;
-};
-
 struct operation_result_tracker;
 class db_context;
 class metadata;
@@ -115,9 +80,8 @@ public:
         lowres_clock::time_point timeout,
         utils::chunked_vector<mutation>&& mutations,
         tracing::trace_state_ptr tr_state,
-        db::consistency_level write_cl,
-        per_request_options options = {}
-    );
+        db::consistency_level write_cl
+        );
     bool needs_cdc_augmentation(const utils::chunked_vector<mutation>&) const;
 };
 
@@ -129,6 +93,15 @@ struct db_context final {
         : _proxy(proxy), _migration_notifier(notifier), _cdc_metadata(cdc_meta) {}
 };
 
+// cdc log table operation
+enum class operation : int8_t {
+    // note: these values will eventually be read by a third party, probably not privvy to this
+    // enum decl, so don't change the constant values (or the datatype).
+    pre_image = 0, update = 1, insert = 2, row_delete = 3, partition_delete = 4,
+    range_delete_start_inclusive = 5, range_delete_start_exclusive = 6, range_delete_end_inclusive = 7, range_delete_end_exclusive = 8,
+    post_image = 9,
+};
+
 bool is_log_for_some_table(const replica::database& db, const sstring& ks_name, const std::string_view& table_name);
 
 schema_ptr get_base_table(const replica::database&, const schema&);
@@ -153,7 +126,4 @@ bool is_cdc_metacolumn_name(const sstring& name);
 
 utils::UUID generate_timeuuid(api::timestamp_type t);
 
-cell_map* get_row_state(row_states_map& row_states, const clustering_key& ck);
-const cell_map* get_row_state(const row_states_map& row_states, const clustering_key& ck);
-
 } // namespace cdc

cdc/split.cc

@@ -6,28 +6,15 @@
  * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
  */
 
-#include "bytes.hh"
-#include "bytes_fwd.hh"
-#include "mutation/atomic_cell.hh"
-#include "mutation/atomic_cell_or_collection.hh"
-#include "mutation/collection_mutation.hh"
 #include "mutation/mutation.hh"
-#include "mutation/tombstone.hh"
 #include "schema/schema.hh"
 
-#include <seastar/core/sstring.hh>
 #include "types/concrete_types.hh"
-#include "types/types.hh"
 #include "types/user.hh"
 
 #include "split.hh"
 #include "log.hh"
 #include "change_visitor.hh"
-#include "utils/managed_bytes.hh"
-#include <string_view>
-#include <unordered_map>
-
-extern logging::logger cdc_log;
 
 struct atomic_column_update {
     column_id id;
@@ -124,15 +111,6 @@ struct batch {
                 ret.insert(std::make_pair(change.key, all_columns));
             }
         }
-        // While deleting a full partition avoids row-by-row logging for performance
-        // reasons, we must explicitly log single-row deletions for tables without a
-        // clustering key. This ensures consistent behavior with deletions of single
-        // rows from tables with a clustering key. See issue #26382.
-        if (partition_deletions && s.clustering_key_size() == 0) {
-            cdc::one_kind_column_set all_columns{s.regular_columns_count()};
-            all_columns.set(0, s.regular_columns_count(), true);
-            ret.emplace(clustering_key::make_empty(), all_columns);
-        }
 
         auto process_change_type = [&] (const auto& changes) {
             for (const auto& change : changes) {
@@ -503,8 +481,6 @@ struct should_split_visitor {
     // Otherwise we store the change's ttl.
     std::optional<gc_clock::duration> _ttl = std::nullopt;
 
-    virtual ~should_split_visitor() = default;
-
     inline bool finished() const { return _result; }
     inline void stop() { _result = true; }
 
@@ -527,7 +503,7 @@ struct should_split_visitor {
 
     void collection_tombstone(const tombstone& t) { visit(t.timestamp + 1); }
 
-    virtual void live_collection_cell(bytes_view, const atomic_cell_view& cell) {
+    void live_collection_cell(bytes_view, const atomic_cell_view& cell) {
         if (_had_row_marker) {
             // nonatomic updates cannot be expressed with an INSERT.
             return stop();
@@ -537,7 +513,7 @@ struct should_split_visitor {
     void dead_collection_cell(bytes_view, const atomic_cell_view& cell) { visit(cell); }
     void collection_column(const column_definition&, auto&& visit_collection) { visit_collection(*this); }
 
-    virtual void marker(const row_marker& rm) {
+    void marker(const row_marker& rm) {
         _had_row_marker = true;
         visit(rm.timestamp(), get_ttl(rm));
     }
@@ -578,29 +554,7 @@ struct should_split_visitor {
     }
 };
 
-// This is the same as the above, but it doesn't split a row marker away from
-// an update. As a result, updates that create an item appear as a single log
-// row.
-class alternator_should_split_visitor : public should_split_visitor {
-public:
-    ~alternator_should_split_visitor() override = default;
-
-    void live_collection_cell(bytes_view, const atomic_cell_view& cell) override {
-        visit(cell.timestamp());
-    }
-
-    void marker(const row_marker& rm) override {
-        visit(rm.timestamp());
-    }
-};
-
-bool should_split(const mutation& m, const per_request_options& options) {
-    if (options.alternator) {
-        alternator_should_split_visitor v;
-        cdc::inspect_mutation(m, v);
-        return v._result || v._ts == api::missing_timestamp;
-    }
-
+bool should_split(const mutation& m) {
     should_split_visitor v;
 
     cdc::inspect_mutation(m, v);
@@ -610,109 +564,8 @@ bool should_split(const mutation& m, const per_request_options& options) {
         || v._ts == api::missing_timestamp;
 }
 
-// Returns true if the row state and the atomic and nonatomic entries represent
-// an equivalent item.
-static bool entries_match_row_state(const schema_ptr& base_schema, const cell_map& row_state, const std::vector<atomic_column_update>& atomic_entries,
-        std::vector<nonatomic_column_update>& nonatomic_entries) {
-    for (const auto& update : atomic_entries) {
-        const column_definition& cdef = base_schema->column_at(column_kind::regular_column, update.id);
-        const auto it = row_state.find(&cdef);
-        if (it == row_state.end()) {
-            return false;
-        }
-        if (to_managed_bytes_opt(update.cell.value().linearize()) != it->second) {
-            return false;
-        }
-    }
-    if (nonatomic_entries.empty()) {
-        return true;
-    }
-
-    for (const auto& update : nonatomic_entries) {
-        const column_definition& cdef = base_schema->column_at(column_kind::regular_column, update.id);
-        const auto it = row_state.find(&cdef);
-        if (it == row_state.end()) {
-            return false;
-        }
-
-        // The only collection used by Alternator is a non-frozen map.
-        auto current_raw_map = cdef.type->deserialize(*it->second);
-        map_type_impl::native_type current_values = value_cast<map_type_impl::native_type>(current_raw_map);
-
-        if (current_values.size() != update.cells.size()) {
-            return false;
-        }
-        
-        std::unordered_map<sstring_view, bytes> current_values_map;
-        for (const auto& entry : current_values) {
-            const auto attr_name = std::string_view(value_cast<sstring>(entry.first));
-            current_values_map[attr_name] = value_cast<bytes>(entry.second);
-        }
-
-        for (const auto& [key, value] : update.cells) {
-            const auto key_str = to_string_view(key);
-            if (!value.is_live()) {
-                if (current_values_map.contains(key_str)) {
-                    return false;
-                }
-            } else if (current_values_map[key_str] != value.value().linearize()) {
-                return false;
-            }
-        }
-    }
-    return true;
-}
-
-bool should_skip(batch& changes, const mutation& base_mutation, change_processor& processor) {
-    const schema_ptr& base_schema = base_mutation.schema();
-    // Alternator doesn't use static updates and clustered range deletions.
-    if (!changes.static_updates.empty() || !changes.clustered_range_deletions.empty()) {
-        return false;
-    }
-
-    for (clustered_row_insert& u : changes.clustered_inserts) {
-        const cell_map* row_state = get_row_state(processor.clustering_row_states(), u.key);
-        if (!row_state) {
-            return false;
-        }
-        if (!entries_match_row_state(base_schema, *row_state, u.atomic_entries, u.nonatomic_entries)) {
-            return false;
-        }
-    }
-
-    for (clustered_row_update& u : changes.clustered_updates) {
-        const cell_map* row_state = get_row_state(processor.clustering_row_states(), u.key);
-        if (!row_state) {
-            return false;
-        }
-        if (!entries_match_row_state(base_schema, *row_state, u.atomic_entries, u.nonatomic_entries)) {
-            return false;
-        }
-    }
-
-    // Skip only if the row being deleted does not exist (i.e. the deletion is a no-op).
-    for (const auto& row_deletion : changes.clustered_row_deletions) {
-        if (processor.clustering_row_states().contains(row_deletion.key)) {
-            return false;
-        }
-    }
-
-    // Don't skip if the item exists.
-    //
-    // Increased DynamoDB Streams compatibility guarantees that single-item
-    // operations will read the item and store it in the clustering row states.
-    // If it is not found there, we may skip CDC. This is safe as long as the
-    // assumptions of this operation's write isolation are not violated.
-    if (changes.partition_deletions && processor.clustering_row_states().contains(clustering_key::make_empty())) {
-        return false;
-    }
-
-    cdc_log.trace("Skipping CDC log for mutation {}", base_mutation);
-    return true;
-}
-
 void process_changes_with_splitting(const mutation& base_mutation, change_processor& processor,
-        bool enable_preimage, bool enable_postimage, bool alternator_strict_compatibility) {
+        bool enable_preimage, bool enable_postimage) {
     const auto base_schema = base_mutation.schema();
     auto changes = extract_changes(base_mutation);
     auto pk = base_mutation.key();
@@ -724,6 +577,9 @@ void process_changes_with_splitting(const mutation& base_mutation, change_proces
     const auto last_timestamp = changes.rbegin()->first;
 
     for (auto& [change_ts, btch] : changes) {
+        const bool is_last = change_ts == last_timestamp;
+        processor.begin_timestamp(change_ts, is_last);
+
         clustered_column_set affected_clustered_columns_per_row{clustering_key::less_compare(*base_schema)};
         one_kind_column_set affected_static_columns{base_schema->static_columns_count()};
 
@@ -732,12 +588,6 @@ void process_changes_with_splitting(const mutation& base_mutation, change_proces
             affected_clustered_columns_per_row = btch.get_affected_clustered_columns_per_row(*base_mutation.schema());
         }
 
-        if (alternator_strict_compatibility && should_skip(btch, base_mutation, processor)) {
-            continue;
-        }
-
-        const bool is_last = change_ts == last_timestamp;
-        processor.begin_timestamp(change_ts, is_last);
         if (enable_preimage) {
             if (affected_static_columns.count() > 0) {
                 processor.produce_preimage(nullptr, affected_static_columns);
@@ -825,13 +675,7 @@ void process_changes_with_splitting(const mutation& base_mutation, change_proces
 }
 
 void process_changes_without_splitting(const mutation& base_mutation, change_processor& processor,
-        bool enable_preimage, bool enable_postimage, bool alternator_strict_compatibility) {
-    if (alternator_strict_compatibility) {
-        auto changes = extract_changes(base_mutation);
-        if (should_skip(changes.begin()->second, base_mutation, processor)) {
-            return;
-        }
-    }
+        bool enable_preimage, bool enable_postimage) {
     auto ts = find_timestamp(base_mutation);
     processor.begin_timestamp(ts, true);
 

cdc/split.hh

@@ -9,7 +9,6 @@
 #pragma once
 
 #include <boost/dynamic_bitset.hpp>  // IWYU pragma: keep
-#include "cdc/log.hh"
 #include "replica/database_fwd.hh"
 #include "mutation/timestamp.hh"
 
@@ -66,14 +65,12 @@ public:
     // Tells processor we have reached end of record - last part
     // of a given timestamp batch
     virtual void end_record() = 0;
-
-    virtual const row_states_map& clustering_row_states() const = 0;
 };
 
-bool should_split(const mutation& base_mutation, const per_request_options& options);
+bool should_split(const mutation& base_mutation);
 void process_changes_with_splitting(const mutation& base_mutation, change_processor& processor,
-        bool enable_preimage, bool enable_postimage, bool alternator_strict_compatibility);
+        bool enable_preimage, bool enable_postimage);
 void process_changes_without_splitting(const mutation& base_mutation, change_processor& processor,
-        bool enable_preimage, bool enable_postimage, bool alternator_strict_compatibility);
+        bool enable_preimage, bool enable_postimage);
 
 }